<s> import copy <EOL> import mock <EOL> from oslo_log import log as logging <EOL> from neutron . common import constants as l3_constants <EOL> from neutron . common import test_lib <EOL> from neutron import context <EOL> from neutron . extensions import providernet as pr_net <EOL> from neutron . tests . unit . extensions import test_l3 <EOL> from networking_cisco . plugins . cisco . device_manager . plugging_drivers . hw_vlan_trunking_driver import HwVLANTrunkingPlugDriver <EOL> from networking_cisco . tests . unit . cisco . l3 import ( <EOL> test_l3_router_appliance_plugin ) <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class TestHwVLANTrunkingPlugDriver ( <EOL> test_l3_router_appliance_plugin . L3RouterApplianceTestCaseBase , <EOL> test_l3 . L3NatTestCaseMixin ) : <EOL> configure_routertypes = False <EOL> router_type = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> super ( TestHwVLANTrunkingPlugDriver , self ) . setUp ( create_mgmt_nw = False ) <EOL> self . _old_config_files = copy . copy ( test_lib . test_config . get ( <EOL> '<STR_LIT>' ) ) <EOL> self . _add_device_manager_plugin_ini_file ( ) <EOL> self . _add_router_plugin_ini_file ( ) <EOL> self . setup_config ( ) <EOL> def tearDown ( self ) : <EOL> if self . _old_config_files is None : <EOL> test_lib . test_config . pop ( '<STR_LIT>' , None ) <EOL> else : <EOL> test_lib . test_config [ '<STR_LIT>' ] = self . _old_config_files <EOL> super ( TestHwVLANTrunkingPlugDriver , self ) . tearDown ( ) <EOL> def test_create_hosting_device_resources ( self ) : <EOL> tenant_id = '<STR_LIT>' <EOL> ctx = context . Context ( '<STR_LIT>' , tenant_id , is_admin = True ) <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> mgmt_context = { '<STR_LIT>' : None } <EOL> res = plugging_driver . create_hosting_device_resources ( <EOL> ctx , "<STR_LIT>" , tenant_id , mgmt_context , <NUM_LIT:2> ) <EOL> self . assertIsNone ( res [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( len ( res ) , <NUM_LIT:1> ) <EOL> def test_create_hosting_device_resources_no_mgmt_context ( self ) : <EOL> tenant_id = '<STR_LIT>' <EOL> ctx = context . Context ( '<STR_LIT>' , tenant_id , is_admin = True ) <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> res = plugging_driver . create_hosting_device_resources ( <EOL> ctx , "<STR_LIT>" , tenant_id , None , <NUM_LIT:2> ) <EOL> self . assertIsNone ( res [ '<STR_LIT>' ] , res ) <EOL> self . assertEqual ( len ( res ) , <NUM_LIT:1> ) <EOL> def test_get_hosting_device_resources_by_complementary_id ( self ) : <EOL> tenant_id = '<STR_LIT>' <EOL> ctx = context . Context ( '<STR_LIT>' , tenant_id , is_admin = True ) <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> mgmt_context = { '<STR_LIT>' : None } <EOL> res = plugging_driver . create_hosting_device_resources ( <EOL> ctx , "<STR_LIT>" , tenant_id , mgmt_context , <NUM_LIT:1> ) <EOL> with self . port ( ) , self . port ( device_id = '<STR_LIT>' ) , self . port ( <EOL> tenant_id = tenant_id ) : <EOL> res_get = plugging_driver . get_hosting_device_resources ( <EOL> ctx , '<STR_LIT>' , '<STR_LIT>' , tenant_id , None ) <EOL> self . assertIsNone ( res_get [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( len ( res ) , <NUM_LIT:1> ) <EOL> def test_get_hosting_device_resources_by_device_id ( self ) : <EOL> tenant_id = '<STR_LIT>' <EOL> ctx = context . Context ( '<STR_LIT>' , tenant_id , is_admin = True ) <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> mgmt_context = { '<STR_LIT>' : None } <EOL> res = plugging_driver . create_hosting_device_resources ( <EOL> ctx , "<STR_LIT>" , tenant_id , mgmt_context , <NUM_LIT:1> ) <EOL> hd_uuid = '<STR_LIT>' <EOL> update_spec = { '<STR_LIT:port>' : { '<STR_LIT>' : hd_uuid , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> for hd_port in self . _list ( '<STR_LIT>' ) [ '<STR_LIT>' ] : <EOL> self . _update ( '<STR_LIT>' , hd_port [ '<STR_LIT:id>' ] , update_spec ) <EOL> with self . port ( ) , self . port ( device_id = '<STR_LIT>' ) , self . port ( <EOL> tenant_id = tenant_id ) , self . port ( tenant_id = tenant_id , <EOL> device_owner = '<STR_LIT>' ) : <EOL> res_get = plugging_driver . get_hosting_device_resources ( <EOL> ctx , hd_uuid , '<STR_LIT>' , tenant_id , None ) <EOL> self . assertIsNone ( res_get [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( len ( res ) , <NUM_LIT:1> ) <EOL> def test_delete_hosting_device_resources ( self ) : <EOL> tenant_id = '<STR_LIT>' <EOL> ctx = context . Context ( '<STR_LIT>' , tenant_id , is_admin = True ) <EOL> mgmt_context = { '<STR_LIT>' : None } <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> res = plugging_driver . create_hosting_device_resources ( <EOL> ctx , "<STR_LIT>" , tenant_id , mgmt_context , <NUM_LIT:2> ) <EOL> nets = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( nets [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> subnets = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( subnets [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> ports = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( ports [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> mgmt_port = res [ '<STR_LIT>' ] <EOL> del res [ '<STR_LIT>' ] <EOL> plugging_driver . delete_hosting_device_resources ( <EOL> ctx , tenant_id , mgmt_port , ** res ) <EOL> nets = self . _list ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( len ( nets ) , <NUM_LIT:0> ) <EOL> subnets = self . _list ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( len ( subnets ) , <NUM_LIT:0> ) <EOL> ports = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( ports [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> def test_extend_hosting_port_info_adds_segmentation_id_internal ( self ) : <EOL> hosting_info = { } <EOL> fake_port_db_obj = mock . MagicMock ( ) <EOL> fake_port_db_obj . hosting_info = mock . MagicMock ( ) <EOL> fake_port_db_obj . hosting_info . segmentation_id = <NUM_LIT:50> <EOL> fake_port_db_obj . device_owner = l3_constants . DEVICE_OWNER_ROUTER_INTF <EOL> hosting_device = { '<STR_LIT:id>' : '<STR_LIT>' } <EOL> tenant_id = '<STR_LIT>' <EOL> ctx = context . Context ( '<STR_LIT>' , tenant_id , is_admin = True ) <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> plugging_driver . extend_hosting_port_info ( ctx , fake_port_db_obj , <EOL> hosting_device , hosting_info ) <EOL> self . assertEqual ( hosting_info [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( hosting_info [ '<STR_LIT>' ] , <NUM_LIT:50> ) <EOL> def test_extend_hosting_port_info_adds_segmentation_id_external ( self ) : <EOL> hosting_info = { } <EOL> fake_port_db_obj = mock . MagicMock ( ) <EOL> fake_port_db_obj . hosting_info = mock . MagicMock ( ) <EOL> fake_port_db_obj . hosting_info . segmentation_id = <NUM_LIT> <EOL> fake_port_db_obj . device_owner = l3_constants . DEVICE_OWNER_ROUTER_GW <EOL> hosting_device = { '<STR_LIT:id>' : '<STR_LIT>' } <EOL> tenant_id = '<STR_LIT>' <EOL> ctx = context . Context ( '<STR_LIT>' , tenant_id , is_admin = True ) <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> plugging_driver . extend_hosting_port_info ( ctx , fake_port_db_obj , <EOL> hosting_device , hosting_info ) <EOL> self . assertEqual ( hosting_info [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( hosting_info [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> def _update_provider_net_info ( self , res_list , fields ) : <EOL> for res in res_list : <EOL> pv_info = self . _pv_info [ '<STR_LIT>' ] . get ( res [ '<STR_LIT:id>' ] ) <EOL> if pv_info is None : <EOL> pv_info = self . _pv_info [ '<STR_LIT>' ] . get ( res [ '<STR_LIT:id>' ] ) <EOL> if pv_info is None : <EOL> nw_type = self . _network_type <EOL> if not self . _pv_info [ nw_type ] : <EOL> tag = { '<STR_LIT>' : <NUM_LIT:50> , '<STR_LIT>' : <NUM_LIT> } [ nw_type ] <EOL> pv_info = { '<STR_LIT>' : nw_type , '<STR_LIT>' : tag } <EOL> self . _pv_info [ nw_type ] [ res [ '<STR_LIT:id>' ] ] = pv_info <EOL> if pv_info is None : <EOL> tag = max ( [ i [ '<STR_LIT>' ] <EOL> for i in self . _pv_info [ nw_type ] . values ( ) ] ) + <NUM_LIT:1> <EOL> pv_info = { '<STR_LIT>' : nw_type , '<STR_LIT>' : tag } <EOL> self . _pv_info [ nw_type ] [ res [ '<STR_LIT:id>' ] ] = pv_info <EOL> res [ pr_net . NETWORK_TYPE ] = pv_info [ '<STR_LIT>' ] <EOL> res [ pr_net . SEGMENTATION_ID ] = pv_info [ '<STR_LIT>' ] <EOL> if fields is not None : <EOL> for attr in list ( res . keys ( ) ) : <EOL> if attr not in fields : <EOL> del res [ attr ] <EOL> def _mocked_get_network ( self , context , id , fields = None ) : <EOL> res = self . real_get_network ( context , id ) <EOL> self . _update_provider_net_info ( [ res ] , fields ) <EOL> return res <EOL> def _mocked_get_networks ( self , * args , ** kwargs ) : <EOL> if len ( args ) >= <NUM_LIT:3> : <EOL> fields = args [ <NUM_LIT:2> ] <EOL> list_args = [ i for i in args ] <EOL> list_args [ <NUM_LIT:2> ] = None <EOL> args = list_args <EOL> else : <EOL> fields = kwargs . pop ( '<STR_LIT>' , None ) <EOL> res_list = self . real_get_networks ( * args , ** kwargs ) <EOL> self . _update_provider_net_info ( res_list , fields ) <EOL> return res_list <EOL> def _test_allocate_hosting_port ( self , test_info1 ) : <EOL> def _validate_allocation ( plugin , ctx , r , port_db , test_info , <EOL> i , hd , plugging_driver ) : <EOL> binding_db = plugin . _allocate_hosting_port ( <EOL> ctx , r [ '<STR_LIT:id>' ] , port_db , hd [ '<STR_LIT:id>' ] , plugging_driver ) <EOL> self . assertIsNotNone ( binding_db . hosting_port_id ) <EOL> self . assertEqual ( binding_db . hosting_port_id , <EOL> port_db . id ) <EOL> self . assertEqual ( binding_db . segmentation_id , <EOL> test_info [ '<STR_LIT>' ] [ i ] ) <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> with self . subnet ( ) as subnet1 : <EOL> sn1 = subnet1 [ '<STR_LIT>' ] <EOL> ext_net_id = sn1 [ '<STR_LIT>' ] <EOL> self . _set_net_external ( ext_net_id ) <EOL> gw_info = { '<STR_LIT>' : ext_net_id } <EOL> with self . router ( external_gateway_info = gw_info , <EOL> tenant_id = sn1 [ '<STR_LIT>' ] ) as router1 : <EOL> r1 = router1 [ '<STR_LIT>' ] <EOL> hds = self . _list ( '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> hd = hds [ <NUM_LIT:0> ] <EOL> self . _pv_info = { '<STR_LIT>' : { } , '<STR_LIT>' : { } } <EOL> self . _network_type = test_info1 [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . real_get_network = self . core_plugin . get_network <EOL> self . real_get_networks = self . core_plugin . get_networks <EOL> with mock . patch . object ( self . core_plugin , '<STR_LIT>' ) as m1 , mock . patch . object ( self . core_plugin , <EOL> '<STR_LIT>' ) as m2 : <EOL> m1 . side_effect = self . _mocked_get_network <EOL> m2 . side_effect = self . _mocked_get_networks <EOL> u1_ctx = context . Context ( '<STR_LIT>' , r1 [ '<STR_LIT>' ] , <EOL> is_admin = True ) <EOL> gw_port_db = self . core_plugin . _get_ports_query ( <EOL> u1_ctx , filters = { '<STR_LIT>' : [ ext_net_id ] } ) . one ( ) <EOL> _validate_allocation ( <EOL> self . plugin , u1_ctx , r1 , gw_port_db , <EOL> test_info1 , <NUM_LIT:0> , hd , plugging_driver ) <EOL> for i in range ( <NUM_LIT:1> , len ( test_info1 [ '<STR_LIT>' ] ) ) : <EOL> cidr = '<STR_LIT>' + str ( i ) + '<STR_LIT>' <EOL> with self . subnet ( cidr = cidr ) as subnet2 : <EOL> sn2 = subnet2 [ '<STR_LIT>' ] <EOL> itfc_info = self . _router_interface_action ( <EOL> '<STR_LIT>' , r1 [ '<STR_LIT:id>' ] , sn2 [ '<STR_LIT:id>' ] , None ) <EOL> self . _network_type = test_info1 [ '<STR_LIT>' ] [ i ] <EOL> port_db = self . core_plugin . _get_port ( <EOL> u1_ctx , itfc_info [ '<STR_LIT>' ] ) <EOL> _validate_allocation ( <EOL> self . plugin , u1_ctx , r1 , port_db , test_info1 , <EOL> i , hd , plugging_driver ) <EOL> def test_allocate_hosting_port_vlan_network_all_unused ( self ) : <EOL> self . _test_allocate_hosting_port ( { '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:50> ] } ) <EOL> def test_allocate_hosting_port_vlan_network_vlan_already_allocated ( self ) : <EOL> self . _test_allocate_hosting_port ( { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:50> , <NUM_LIT> ] } ) <EOL> def test_allocate_hosting_port_vlan_network_not_found_failure ( self ) : <EOL> with self . subnet ( ) as subnet1 : <EOL> sn1 = subnet1 [ '<STR_LIT>' ] <EOL> ext_net_id = sn1 [ '<STR_LIT>' ] <EOL> self . _set_net_external ( ext_net_id ) <EOL> gw_info = { '<STR_LIT>' : ext_net_id } <EOL> with self . router ( external_gateway_info = gw_info , <EOL> tenant_id = sn1 [ '<STR_LIT>' ] ) as router1 : <EOL> r1 = router1 [ '<STR_LIT>' ] <EOL> plugging_driver = HwVLANTrunkingPlugDriver ( ) <EOL> u_ctx = context . Context ( '<STR_LIT>' , r1 [ '<STR_LIT>' ] , is_admin = True ) <EOL> gw_port_db = self . core_plugin . _get_ports_query ( <EOL> u_ctx , filters = { '<STR_LIT>' : [ ext_net_id ] } ) . one ( ) <EOL> allocations = plugging_driver . allocate_hosting_port ( <EOL> u_ctx , r1 [ '<STR_LIT:id>' ] , gw_port_db , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertIsNone ( allocations ) </s>
