<s> import collections <EOL> import testtools <EOL> from networking_cisco . plugins . ml2 . drivers . cisco . nexus import exceptions <EOL> from networking_cisco . plugins . ml2 . drivers . cisco . nexus import nexus_db_v2 <EOL> from neutron . tests . unit import testlib_api <EOL> class CiscoNexusDbTest ( testlib_api . SqlTestCase ) : <EOL> """<STR_LIT>""" <EOL> NpbObj = collections . namedtuple ( '<STR_LIT>' , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _npb_test_obj ( self , pnum , vnum , vni = <NUM_LIT:0> , switch = '<STR_LIT>' , <EOL> instance = None , is_provider_vlan = False , <EOL> channel_group = <NUM_LIT:0> , is_native_vlan = False ) : <EOL> """<STR_LIT>""" <EOL> if pnum is '<STR_LIT>' : <EOL> port = pnum <EOL> else : <EOL> port = '<STR_LIT>' % pnum <EOL> if instance is None : <EOL> instance = '<STR_LIT>' % ( pnum , vnum ) <EOL> return self . NpbObj ( port , vnum , vni , switch , instance , <EOL> is_provider_vlan , channel_group , <EOL> is_native_vlan = False ) <EOL> def _assert_bindings_match ( self , npb , npb_obj ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( npb . port_id , npb_obj . port ) <EOL> self . assertEqual ( npb . vlan_id , npb_obj . vlan ) <EOL> self . assertEqual ( npb . switch_ip , npb_obj . switch ) <EOL> self . assertEqual ( npb . instance_id , npb_obj . instance ) <EOL> def _add_binding_to_db ( self , npb ) : <EOL> """<STR_LIT>""" <EOL> return nexus_db_v2 . add_nexusport_binding ( <EOL> npb . port , npb . vlan , npb . vni , npb . switch , npb . instance , <EOL> npb . is_provider_vlan , npb . channel_group , <EOL> npb . is_native_vlan ) <EOL> def _add_bindings_to_db ( self , npbs ) : <EOL> """<STR_LIT>""" <EOL> for npb in npbs : <EOL> nexus_db_v2 . add_nexusport_binding ( <EOL> npb . port , npb . vlan , npb . vni , npb . switch , npb . instance , <EOL> npb . is_provider_vlan , npb . channel_group , <EOL> npb . is_native_vlan ) <EOL> def _remove_binding_from_db ( self , npb ) : <EOL> """<STR_LIT>""" <EOL> return nexus_db_v2 . remove_nexusport_binding ( <EOL> npb . port , npb . vlan , npb . vni , npb . switch , npb . instance , <EOL> npb . is_provider_vlan ) <EOL> def _get_nexusport_binding ( self , npb ) : <EOL> """<STR_LIT>""" <EOL> return nexus_db_v2 . get_nexusport_binding ( <EOL> npb . port , npb . vlan , npb . switch , npb . instance ) <EOL> def _get_nexusvlan_binding ( self , npb ) : <EOL> """<STR_LIT>""" <EOL> return nexus_db_v2 . get_nexusvlan_binding ( npb . vlan , npb . switch ) <EOL> def _get_nexusvm_binding ( self , npb ) : <EOL> """<STR_LIT>""" <EOL> return nexus_db_v2 . get_nexusvm_bindings ( npb . vlan , npb . instance ) [ <NUM_LIT:0> ] <EOL> def _get_port_vlan_switch_binding ( self , npb ) : <EOL> """<STR_LIT>""" <EOL> return nexus_db_v2 . get_port_vlan_switch_binding ( <EOL> npb . port , npb . vlan , npb . switch ) <EOL> def _get_port_switch_bindings ( self , npb ) : <EOL> """<STR_LIT>""" <EOL> return nexus_db_v2 . get_port_switch_bindings ( npb . port , npb . switch ) <EOL> def test_nexusportbinding_add_remove ( self ) : <EOL> """<STR_LIT>""" <EOL> npb11 = self . _npb_test_obj ( <NUM_LIT:10> , <NUM_LIT:100> ) <EOL> npb = self . _add_binding_to_db ( npb11 ) <EOL> self . _assert_bindings_match ( npb , npb11 ) <EOL> npb = self . _remove_binding_from_db ( npb11 ) <EOL> self . assertEqual ( len ( npb ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb [ <NUM_LIT:0> ] , npb11 ) <EOL> with testtools . ExpectedException ( exceptions . NexusPortBindingNotFound ) : <EOL> self . _remove_binding_from_db ( npb11 ) <EOL> def test_nexusportbinding_get ( self ) : <EOL> """<STR_LIT>""" <EOL> npb11 = self . _npb_test_obj ( <NUM_LIT:10> , <NUM_LIT:100> ) <EOL> npb21 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:100> ) <EOL> npb22 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:200> ) <EOL> self . _add_bindings_to_db ( [ npb11 , npb21 , npb22 ] ) <EOL> npb = self . _get_nexusport_binding ( npb11 ) <EOL> self . assertEqual ( len ( npb ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb [ <NUM_LIT:0> ] , npb11 ) <EOL> npb = self . _get_nexusport_binding ( npb21 ) <EOL> self . assertEqual ( len ( npb ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb [ <NUM_LIT:0> ] , npb21 ) <EOL> npb = self . _get_nexusport_binding ( npb22 ) <EOL> self . assertEqual ( len ( npb ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb [ <NUM_LIT:0> ] , npb22 ) <EOL> with testtools . ExpectedException ( exceptions . NexusPortBindingNotFound ) : <EOL> nexus_db_v2 . get_nexusport_binding ( <EOL> npb21 . port , npb21 . vlan , npb21 . switch , "<STR_LIT>" ) <EOL> def test_nexusvlanbinding_get ( self ) : <EOL> """<STR_LIT>""" <EOL> npb11 = self . _npb_test_obj ( <NUM_LIT:10> , <NUM_LIT:100> ) <EOL> npb21 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:100> ) <EOL> npb22 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:200> ) <EOL> self . _add_bindings_to_db ( [ npb11 , npb21 , npb22 ] ) <EOL> npb_all_v100 = self . _get_nexusvlan_binding ( npb11 ) <EOL> self . assertEqual ( len ( npb_all_v100 ) , <NUM_LIT:2> ) <EOL> npb_v200 = self . _get_nexusvlan_binding ( npb22 ) <EOL> self . assertEqual ( len ( npb_v200 ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb_v200 [ <NUM_LIT:0> ] , npb22 ) <EOL> with testtools . ExpectedException ( exceptions . NexusPortBindingNotFound ) : <EOL> nexus_db_v2 . get_nexusvlan_binding ( npb21 . vlan , "<STR_LIT>" ) <EOL> def test_nexusvmbinding_get ( self ) : <EOL> """<STR_LIT>""" <EOL> npb11 = self . _npb_test_obj ( <NUM_LIT:10> , <NUM_LIT:100> ) <EOL> npb21 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:100> ) <EOL> npb22 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:200> ) <EOL> self . _add_bindings_to_db ( [ npb11 , npb21 , npb22 ] ) <EOL> npb = self . _get_nexusvm_binding ( npb21 ) <EOL> self . _assert_bindings_match ( npb , npb21 ) <EOL> npb = self . _get_nexusvm_binding ( npb22 ) <EOL> self . _assert_bindings_match ( npb , npb22 ) <EOL> with testtools . ExpectedException ( exceptions . NexusPortBindingNotFound ) : <EOL> nexus_db_v2 . get_nexusvm_bindings ( npb21 . vlan , "<STR_LIT>" ) [ <NUM_LIT:0> ] <EOL> def test_nexusportvlanswitchbinding_get ( self ) : <EOL> """<STR_LIT>""" <EOL> npb11 = self . _npb_test_obj ( <NUM_LIT:10> , <NUM_LIT:100> ) <EOL> npb21 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:100> ) <EOL> self . _add_bindings_to_db ( [ npb11 , npb21 ] ) <EOL> npb = self . _get_port_vlan_switch_binding ( npb11 ) <EOL> self . assertEqual ( len ( npb ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb [ <NUM_LIT:0> ] , npb11 ) <EOL> with testtools . ExpectedException ( exceptions . NexusPortBindingNotFound ) : <EOL> nexus_db_v2 . get_port_vlan_switch_binding ( <EOL> npb21 . port , npb21 . vlan , "<STR_LIT>" ) <EOL> def test_nexusportswitchbinding_get ( self ) : <EOL> """<STR_LIT>""" <EOL> npb11 = self . _npb_test_obj ( <NUM_LIT:10> , <NUM_LIT:100> ) <EOL> npb21 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:100> , switch = '<STR_LIT>' ) <EOL> npb22 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:200> , switch = '<STR_LIT>' ) <EOL> self . _add_bindings_to_db ( [ npb11 , npb21 , npb22 ] ) <EOL> npb = self . _get_port_switch_bindings ( npb11 ) <EOL> self . assertEqual ( len ( npb ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb [ <NUM_LIT:0> ] , npb11 ) <EOL> npb_all_p20 = self . _get_port_switch_bindings ( npb21 ) <EOL> self . assertEqual ( len ( npb_all_p20 ) , <NUM_LIT:2> ) <EOL> npb = nexus_db_v2 . get_port_switch_bindings ( npb21 . port , "<STR_LIT>" ) <EOL> self . assertIsNone ( npb ) <EOL> def test_nexusbinding_update ( self ) : <EOL> """<STR_LIT>""" <EOL> npb11 = self . _npb_test_obj ( <NUM_LIT:10> , <NUM_LIT:100> , switch = '<STR_LIT>' , instance = '<STR_LIT:test>' ) <EOL> npb21 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:100> , switch = '<STR_LIT>' , instance = '<STR_LIT:test>' ) <EOL> self . _add_bindings_to_db ( [ npb11 , npb21 ] ) <EOL> npb_all_v100 = nexus_db_v2 . get_nexusvlan_binding ( <NUM_LIT:100> , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( npb_all_v100 ) , <NUM_LIT:2> ) <EOL> npb22 = self . _npb_test_obj ( <NUM_LIT:20> , <NUM_LIT:200> , switch = '<STR_LIT>' , instance = '<STR_LIT:test>' ) <EOL> npb = nexus_db_v2 . update_nexusport_binding ( npb21 . port , <NUM_LIT:200> ) <EOL> self . _assert_bindings_match ( npb , npb22 ) <EOL> npb_all_v100 = nexus_db_v2 . get_nexusvlan_binding ( <NUM_LIT:100> , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( npb_all_v100 ) , <NUM_LIT:1> ) <EOL> self . _assert_bindings_match ( npb_all_v100 [ <NUM_LIT:0> ] , npb11 ) <EOL> npb = nexus_db_v2 . update_nexusport_binding ( npb21 . port , <NUM_LIT:0> ) <EOL> self . assertIsNone ( npb ) <EOL> npb33 = self . _npb_test_obj ( <NUM_LIT:30> , <NUM_LIT> , switch = '<STR_LIT>' , instance = '<STR_LIT:test>' ) <EOL> with testtools . ExpectedException ( exceptions . NexusPortBindingNotFound ) : <EOL> nexus_db_v2 . update_nexusport_binding ( npb33 . port , <NUM_LIT:200> ) </s>
