<s> import hashlib <EOL> import mock <EOL> import six <EOL> from networking_cisco . plugins . ml2 . drivers . cisco . n1kv import ( <EOL> constants as n1kv_const ) <EOL> from networking_cisco . plugins . ml2 . drivers . cisco . n1kv import ( <EOL> n1kv_client ) <EOL> from networking_cisco . plugins . ml2 . drivers . cisco . n1kv import ( <EOL> n1kv_sync ) <EOL> from networking_cisco . tests . unit . ml2 . drivers . cisco . n1kv . test_cisco_n1kv_mech import TestN1KVMechanismDriver <EOL> class TestN1kvSyncDriver ( TestN1KVMechanismDriver ) : <EOL> """<STR_LIT>""" <EOL> TEST_VSM_NETWORK_PROFILES = [ '<STR_LIT>' , ] <EOL> TEST_VSM_NETWORKS = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , ] <EOL> TEST_VSM_SUBNETS = [ '<STR_LIT>' , ] <EOL> TEST_VSM_PORTS = [ '<STR_LIT>' , '<STR_LIT>' , ] <EOL> TEST_NEUTRON_NETWORK_PROFILES = TEST_VSM_NETWORK_PROFILES [ : ] <EOL> TEST_NEUTRON_NETWORKS = TEST_VSM_NETWORKS [ : ] <EOL> TEST_NEUTRON_SUBNETS = TEST_VSM_SUBNETS [ : ] <EOL> TEST_NEUTRON_PORTS = TEST_VSM_PORTS [ : ] <EOL> def setUp ( self ) : <EOL> super ( TestN1kvSyncDriver , self ) . setUp ( ) <EOL> list_md5_patcher = mock . patch ( n1kv_client . __name__ + <EOL> "<STR_LIT>" ) <EOL> fake_list_md5 = list_md5_patcher . start ( ) <EOL> fake_list_md5 . return_value = self . _fake_get_vsm_md5_hashes ( ) <EOL> self . sync_driver = n1kv_sync . N1kvSyncDriver ( None ) <EOL> self . sync_driver . _get_neutron_resource = mock . MagicMock ( <EOL> side_effect = self . _fake_get_neutron_res ) <EOL> def _fake_get_neutron_res ( self , res ) : <EOL> """<STR_LIT>""" <EOL> res_list = getattr ( self , '<STR_LIT>' + res . upper ( ) ) <EOL> return [ { '<STR_LIT:id>' : x } for x in res_list ] <EOL> def _fake_get_vsm_md5_hashes ( self ) : <EOL> """<STR_LIT>""" <EOL> def calc_md5 ( resource ) : <EOL> md5 = hashlib . md5 ( ) <EOL> res_name = '<STR_LIT>' + resource . split ( '<STR_LIT>' ) [ <NUM_LIT:0> ] . upper ( ) + '<STR_LIT:S>' <EOL> for uuid in sorted ( getattr ( self , res_name ) ) : <EOL> md5 . update ( six . b ( uuid ) ) <EOL> return md5 . hexdigest ( ) <EOL> res_order = [ n1kv_const . NETWORK_PROFILE_MD5 , n1kv_const . SUBNET_MD5 , <EOL> n1kv_const . NETWORK_MD5 , n1kv_const . PORT_MD5 ] <EOL> res_md5 = { res : calc_md5 ( res ) for res in res_order } <EOL> consolidated_md5 = hashlib . md5 ( ) <EOL> for res in res_order : <EOL> consolidated_md5 . update ( six . b ( res_md5 [ res ] ) ) <EOL> res_md5 [ n1kv_const . CONSOLIDATED_MD5 ] = consolidated_md5 . hexdigest ( ) <EOL> vsm_md5_hashes = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : res_md5 <EOL> } <EOL> } <EOL> return vsm_md5_hashes <EOL> def test_md5_hash_comparison_all ( self ) : <EOL> """<STR_LIT>""" <EOL> self . sync_driver . _md5_hash_comparison ( None ) <EOL> self . assertFalse ( any ( self . sync_driver . sync_resource . values ( ) ) ) <EOL> def test_md5_hash_comparison_networks ( self ) : <EOL> """<STR_LIT>""" <EOL> self . sync_driver . _md5_hash_comparison ( None ) <EOL> self . assertFalse ( self . sync_driver . sync_resource [ n1kv_const . NETWORKS ] ) <EOL> self . TEST_NEUTRON_NETWORKS . append ( '<STR_LIT>' ) <EOL> self . sync_driver . _md5_hash_comparison ( None ) <EOL> self . assertTrue ( self . sync_driver . sync_resource [ n1kv_const . NETWORKS ] ) <EOL> self . TEST_NEUTRON_NETWORKS . pop ( ) <EOL> network = self . TEST_NEUTRON_NETWORKS . pop ( ) <EOL> self . sync_driver . _md5_hash_comparison ( None ) <EOL> self . assertTrue ( self . sync_driver . sync_resource [ n1kv_const . NETWORKS ] ) <EOL> self . TEST_NEUTRON_NETWORKS . append ( network ) <EOL> def test_md5_hash_comparison_ports ( self ) : <EOL> """<STR_LIT>""" <EOL> self . sync_driver . _md5_hash_comparison ( None ) <EOL> self . assertFalse ( self . sync_driver . sync_resource [ n1kv_const . PORTS ] ) <EOL> self . TEST_NEUTRON_PORTS . append ( '<STR_LIT>' ) <EOL> self . sync_driver . _md5_hash_comparison ( None ) <EOL> self . assertTrue ( self . sync_driver . sync_resource [ n1kv_const . PORTS ] ) <EOL> self . TEST_NEUTRON_PORTS . pop ( ) <EOL> port = self . TEST_NEUTRON_PORTS . pop ( ) <EOL> self . sync_driver . _md5_hash_comparison ( None ) <EOL> self . assertTrue ( self . sync_driver . sync_resource [ n1kv_const . PORTS ] ) <EOL> self . TEST_NEUTRON_PORTS . append ( port ) <EOL> def test_bd_sync_triggered_on_neutron_restart ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( all ( self . sync_driver . sync_bds . values ( ) ) ) </s>
