<s> import copy <EOL> import re <EOL> import sys <EOL> import mock <EOL> import netaddr <EOL> import webob . exc <EOL> from apic_ml2 . neutron . db import port_ha_ipaddress_binding as ha_ip_db <EOL> from apic_ml2 . neutron . tests . unit . ml2 . drivers . cisco . apic import ( <EOL> test_cisco_apic_common as mocked ) <EOL> from apicapi import apic_mapper <EOL> from neutron . agent import securitygroups_rpc as sg_cfg <EOL> from neutron . common import rpc as n_rpc <EOL> from neutron import context <EOL> from neutron . db import api as db_api <EOL> from neutron . db import db_base_plugin_v2 as n_db <EOL> from neutron . db import model_base <EOL> from neutron . extensions import portbindings <EOL> from neutron import manager <EOL> from opflexagent import constants as ocst <EOL> from oslo_config import cfg <EOL> sys . modules [ "<STR_LIT>" ] = mock . Mock ( ) <EOL> from gbpservice . neutron . plugins . ml2 . drivers . grouppolicy . apic import driver <EOL> from gbpservice . neutron . services . grouppolicy import ( <EOL> group_policy_context as p_context ) <EOL> from gbpservice . neutron . services . grouppolicy import config <EOL> from gbpservice . neutron . services . grouppolicy . drivers . cisco . apic import ( <EOL> apic_mapping as amap ) <EOL> from gbpservice . neutron . tests . unit . services . grouppolicy import ( <EOL> test_resource_mapping as test_rmd ) <EOL> APIC_L2_POLICY = '<STR_LIT>' <EOL> APIC_L3_POLICY = '<STR_LIT>' <EOL> APIC_POLICY_RULE_SET = '<STR_LIT>' <EOL> APIC_POLICY_TARGET_GROUP = '<STR_LIT>' <EOL> APIC_POLICY_RULE = '<STR_LIT>' <EOL> APIC_EXTERNAL_RID = '<STR_LIT>' <EOL> APIC_EXTERNAL_EPG = '<STR_LIT>' <EOL> APIC_PRE_L3OUT_TENANT = '<STR_LIT>' <EOL> APIC_PRE_VRF_TENANT = APIC_PRE_L3OUT_TENANT <EOL> APIC_PRE_VRF = '<STR_LIT>' <EOL> AGENT_TYPE = ocst . AGENT_TYPE_OPFLEX_OVS <EOL> AGENT_CONF = { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : AGENT_TYPE , <EOL> '<STR_LIT>' : { '<STR_LIT>' : None , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } } <EOL> AGENT_TYPE_DVS = driver . AGENT_TYPE_DVS <EOL> AGENT_CONF_DVS = { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : AGENT_TYPE_DVS , <EOL> '<STR_LIT>' : { '<STR_LIT>' : None } } <EOL> BOOKED_PORT_VALUE = '<STR_LIT>' <EOL> def echo ( context , string , prefix = '<STR_LIT>' ) : <EOL> return prefix + string <EOL> class MockCallRecorder ( mock . Mock ) : <EOL> recorded_call_set = set ( ) <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> self . recorded_call_set . add ( self . generate_entry ( * args , ** kwargs ) ) <EOL> return mock . Mock ( ) <EOL> def call_happened_with ( self , * args , ** kwargs ) : <EOL> return self . generate_entry ( * args , ** kwargs ) in self . recorded_call_set <EOL> def generate_entry ( self , * args , ** kwargs ) : <EOL> return args , tuple ( ( x , kwargs [ x ] ) for x in sorted ( kwargs . keys ( ) ) ) <EOL> class ApicMappingTestCase ( <EOL> test_rmd . ResourceMappingTestCase , <EOL> mocked . ControllerMixin , mocked . ConfigMixin ) : <EOL> def setUp ( self , sc_plugin = None , nat_enabled = True , <EOL> pre_existing_l3out = False ) : <EOL> self . agent_conf = AGENT_CONF <EOL> cfg . CONF . register_opts ( sg_cfg . security_group_opts , '<STR_LIT>' ) <EOL> config . cfg . CONF . set_override ( '<STR_LIT>' , False , <EOL> group = '<STR_LIT>' ) <EOL> n_rpc . create_connection = mock . Mock ( ) <EOL> amap . ApicMappingDriver . get_apic_manager = mock . MagicMock ( ) <EOL> self . set_up_mocks ( ) <EOL> ml2_opts = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] <EOL> } <EOL> mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) . start ( ) <EOL> nova_client = mock . patch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . start ( ) <EOL> vm = mock . Mock ( ) <EOL> vm . name = '<STR_LIT>' <EOL> nova_client . return_value = vm <EOL> super ( ApicMappingTestCase , self ) . setUp ( <EOL> policy_drivers = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> ml2_options = ml2_opts , sc_plugin = sc_plugin ) <EOL> engine = db_api . get_engine ( ) <EOL> model_base . BASEV2 . metadata . create_all ( engine ) <EOL> plugin = manager . NeutronManager . get_plugin ( ) <EOL> plugin . remove_networks_from_down_agents = mock . Mock ( ) <EOL> plugin . is_agent_down = mock . Mock ( return_value = False ) <EOL> self . driver = manager . NeutronManager . get_service_plugins ( ) [ <EOL> '<STR_LIT>' ] . policy_driver_manager . policy_drivers [ '<STR_LIT>' ] . obj <EOL> amap . ApicMappingDriver . get_base_synchronizer = mock . Mock ( ) <EOL> self . driver . name_mapper . name_mapper = mock . Mock ( ) <EOL> self . driver . name_mapper . name_mapper . tenant = echo <EOL> self . driver . name_mapper . name_mapper . l2_policy = echo <EOL> self . driver . name_mapper . name_mapper . l3_policy = echo <EOL> self . driver . name_mapper . name_mapper . policy_rule_set = echo <EOL> self . driver . name_mapper . name_mapper . policy_rule = echo <EOL> self . driver . name_mapper . name_mapper . app_profile . return_value = ( <EOL> mocked . APIC_AP ) <EOL> self . driver . name_mapper . name_mapper . policy_target_group = echo <EOL> self . driver . name_mapper . name_mapper . external_policy = echo <EOL> self . driver . name_mapper . name_mapper . external_segment = echo <EOL> self . driver . name_mapper . name_mapper . pre_existing = echo <EOL> self . driver . apic_manager = mock . Mock ( name_mapper = mock . Mock ( ) , <EOL> ext_net_dict = { } ) <EOL> self . driver . apic_manager . apic . transaction = self . fake_transaction <EOL> self . driver . notifier = mock . Mock ( ) <EOL> self . driver . apic_manager . ext_net_dict = { } <EOL> amap . apic_manager . TENANT_COMMON = '<STR_LIT>' <EOL> amap . apic_manager . CP_ENTRY = '<STR_LIT>' <EOL> self . common_tenant = amap . apic_manager . TENANT_COMMON <EOL> self . nat_enabled = nat_enabled <EOL> self . driver . l3out_vlan_alloc = mock . Mock ( ) <EOL> self . pre_l3out = pre_existing_l3out <EOL> def echo2 ( string ) : <EOL> return string <EOL> if self . pre_l3out : <EOL> self . orig_query_l3out_info = self . driver . _query_l3out_info <EOL> self . driver . _query_l3out_info = mock . Mock ( ) <EOL> self . driver . _query_l3out_info . return_value = { <EOL> '<STR_LIT>' : apic_mapper . ApicName ( APIC_PRE_L3OUT_TENANT ) , <EOL> '<STR_LIT>' : APIC_PRE_VRF , <EOL> '<STR_LIT>' : APIC_PRE_VRF_TENANT , <EOL> '<STR_LIT>' : ( [ { u'<STR_LIT>' : { } } , <EOL> { u'<STR_LIT>' : { } } , <EOL> { u'<STR_LIT>' : { } } , <EOL> { u'<STR_LIT>' : { } } , <EOL> { u'<STR_LIT>' : { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' } } , <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' } } , <EOL> { u'<STR_LIT>' : { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' } } , <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' } } , <EOL> { u'<STR_LIT>' : { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' } } , <EOL> { u'<STR_LIT>' : { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' } } , <EOL> { u'<STR_LIT>' : { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' } } , <EOL> { u'<STR_LIT>' : { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT:name>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT:type>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT:state>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT>' } , <EOL> u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } } } ] } } <EOL> ] } } , <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , u'<STR_LIT>' : u'<STR_LIT:default>' } } } ] ) } <EOL> self . trimmed_l3out = u'<STR_LIT>' <EOL> self . driver . apic_manager . apic . fvTenant . rn = echo2 <EOL> self . driver . apic_manager . apic . l3extOut . rn = echo2 <EOL> self . driver . l3out_vlan_alloc . reserve_vlan . return_value = <NUM_LIT> <EOL> self . driver . apic_manager . apic . fvTenant . name = echo2 <EOL> self . driver . apic_manager . apic . fvCtx . name = echo2 <EOL> self . _db_plugin = n_db . NeutronDbPluginV2 ( ) <EOL> def _build_external_dict ( self , name , cidr_exposed , is_edge_nat = False ) : <EOL> ext_info = { <EOL> '<STR_LIT>' : '<STR_LIT:True>' if self . nat_enabled else '<STR_LIT:False>' <EOL> } <EOL> if self . pre_l3out : <EOL> ext_info [ '<STR_LIT>' ] = '<STR_LIT:True>' <EOL> ext_info [ '<STR_LIT>' ] = APIC_EXTERNAL_EPG <EOL> else : <EOL> ext_info . update ( { <EOL> '<STR_LIT>' : mocked . APIC_EXT_SWITCH , <EOL> '<STR_LIT:port>' : mocked . APIC_EXT_MODULE + '<STR_LIT:/>' + mocked . APIC_EXT_PORT , <EOL> '<STR_LIT>' : mocked . APIC_EXT_ENCAP , <EOL> '<STR_LIT>' : APIC_EXTERNAL_RID , <EOL> '<STR_LIT>' : str ( netaddr . IPNetwork ( cidr_exposed ) [ <NUM_LIT:1> ] ) , <EOL> '<STR_LIT>' : cidr_exposed } ) <EOL> if is_edge_nat : <EOL> ext_info [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> ext_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return { name : ext_info } <EOL> def _mock_external_dict ( self , data , is_edge_nat = False ) : <EOL> self . driver . apic_manager . ext_net_dict = { } <EOL> for x in data : <EOL> self . driver . apic_manager . ext_net_dict . update ( <EOL> self . _build_external_dict ( x [ <NUM_LIT:0> ] , x [ <NUM_LIT:1> ] , is_edge_nat = is_edge_nat ) ) <EOL> def _create_simple_policy_rule ( self , direction = '<STR_LIT>' , protocol = '<STR_LIT>' , <EOL> port_range = <NUM_LIT> , shared = False , <EOL> action_type = '<STR_LIT>' , action_value = None ) : <EOL> cls = self . create_policy_classifier ( <EOL> direction = direction , protocol = protocol , <EOL> port_range = port_range , shared = shared ) [ '<STR_LIT>' ] <EOL> action = self . create_policy_action ( <EOL> action_type = action_type , shared = shared , <EOL> action_value = action_value ) [ '<STR_LIT>' ] <EOL> return self . create_policy_rule ( <EOL> policy_classifier_id = cls [ '<STR_LIT:id>' ] , policy_actions = [ action [ '<STR_LIT:id>' ] ] , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> def _bind_port_to_host ( self , port_id , host ) : <EOL> data = { '<STR_LIT:port>' : { '<STR_LIT>' : host , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> return super ( ApicMappingTestCase , self ) . _bind_port_to_host ( <EOL> port_id , host , data = data ) <EOL> def _bind_dhcp_port_to_host ( self , port_id , host ) : <EOL> data = { '<STR_LIT:port>' : { '<STR_LIT>' : host , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> return super ( ApicMappingTestCase , self ) . _bind_port_to_host ( <EOL> port_id , host , data = data ) <EOL> class TestPolicyTarget ( ApicMappingTestCase ) : <EOL> def test_policy_target_port_deleted_on_apic ( self ) : <EOL> ptg = self . create_policy_target_group ( ) [ '<STR_LIT>' ] <EOL> subnet = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , self . api ) <EOL> with self . port ( subnet = subnet ) as port : <EOL> self . _bind_port_to_host ( port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> pt = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] , port_id = port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> self . delete_policy_target ( pt [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . assertTrue ( self . driver . notifier . port_update . called ) <EOL> def test_policy_target_delete_no_port ( self ) : <EOL> ptg = self . create_policy_target_group ( ) [ '<STR_LIT>' ] <EOL> subnet = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , self . api ) <EOL> with self . port ( subnet = subnet ) as port : <EOL> self . _bind_port_to_host ( port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> pt = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] , port_id = port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> res = self . new_delete_request ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , <EOL> self . fmt ) . get_response ( self . api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> self . delete_policy_target ( pt [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> expected_res_status = <NUM_LIT> ) <EOL> def test_delete_policy_target_notification_no_apic_network ( self ) : <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . delete_policy_target ( pt1 [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT> ) <EOL> self . assertFalse ( self . driver . notifier . port_update . called ) <EOL> self . driver . notifier . port_update . reset_mock ( ) <EOL> subnet = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , self . api ) <EOL> with self . port ( subnet = subnet ) as port : <EOL> port = self . _bind_port_to_host ( port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] , port_id = port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> self . delete_policy_target ( pt1 [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertTrue ( self . driver . notifier . port_update . called ) <EOL> def test_get_vrf_details ( self ) : <EOL> l3p = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> details = self . driver . get_vrf_details ( <EOL> context . get_admin_context ( ) , <EOL> vrf_id = l3p [ '<STR_LIT:id>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( l3p [ '<STR_LIT:id>' ] , details [ '<STR_LIT>' ] ) <EOL> pool = set ( [ l3p [ '<STR_LIT>' ] ] ) <EOL> if '<STR_LIT>' in l3p : <EOL> pool . add ( l3p [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( pool , set ( details [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( l3p [ '<STR_LIT>' ] , details [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( l3p [ '<STR_LIT:id>' ] , details [ '<STR_LIT>' ] ) <EOL> def test_get_gbp_details ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> self . driver . apic_manager . ext_net_dict [ <EOL> '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> es = self . create_external_segment ( name = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> , shared = False ) [ '<STR_LIT>' ] <EOL> self . create_nat_pool ( external_segment_id = es [ '<STR_LIT:id>' ] , <EOL> ip_pool = '<STR_LIT>' ) <EOL> l3p = self . create_l3_policy ( name = '<STR_LIT>' , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( name = '<STR_LIT>' , <EOL> l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> nsp = self . create_network_service_policy ( <EOL> network_service_params = [ <EOL> { "<STR_LIT:type>" : "<STR_LIT>" , "<STR_LIT:value>" : "<STR_LIT>" , "<STR_LIT:name>" : "<STR_LIT:test>" } ] ) [ <EOL> '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p [ '<STR_LIT:id>' ] , <EOL> network_service_policy_id = nsp [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> mapping = self . driver . get_gbp_details ( context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt1 [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( pt1 [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( ptg [ '<STR_LIT:id>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , mapping [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mapping [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mapping [ '<STR_LIT>' ] ) ) <EOL> fip = mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( pt1 [ '<STR_LIT>' ] , fip [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] , fip [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( es [ '<STR_LIT>' ] , fip [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( l3p [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( l3p [ '<STR_LIT:id>' ] , mapping [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in l3p : <EOL> self . assertEqual ( [ l3p [ '<STR_LIT>' ] , l3p [ '<STR_LIT>' ] ] , <EOL> mapping [ '<STR_LIT>' ] ) <EOL> else : <EOL> self . assertEqual ( [ l3p [ '<STR_LIT>' ] ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mapping [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( es [ '<STR_LIT:name>' ] , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT> , mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> pt2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> mapping = self . driver . get_gbp_details ( context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt2 [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( pt2 [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mapping [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( es [ '<STR_LIT:name>' ] , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT> , mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> def test_snat_pool_subnet_deletion ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> self . driver . apic_manager . ext_net_dict [ <EOL> '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> es = self . create_external_segment ( name = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> , shared = False ) [ '<STR_LIT>' ] <EOL> admin_ctx = context . get_admin_context ( ) <EOL> ext_net_id = self . _db_plugin . get_subnet ( <EOL> admin_ctx , es [ '<STR_LIT>' ] ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( name = '<STR_LIT>' , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( name = '<STR_LIT>' , <EOL> l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> mapping = self . driver . get_gbp_details ( context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt1 [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( pt1 [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mapping [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( es [ '<STR_LIT:name>' ] , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT> , mapping [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . update_l3_policy ( l3p [ '<STR_LIT:id>' ] , external_segments = { } , <EOL> expected_res_status = <NUM_LIT:200> ) <EOL> subnet_filter = { '<STR_LIT:name>' : [ amap . HOST_SNAT_POOL ] , <EOL> '<STR_LIT>' : [ ext_net_id ] } <EOL> internal_subnets = self . _db_plugin . get_subnets ( <EOL> admin_ctx , filters = subnet_filter ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( internal_subnets ) ) <EOL> self . delete_external_segment ( es [ '<STR_LIT:id>' ] , <EOL> expected_res_status = webob . exc . HTTPNoContent . code ) <EOL> internal_subnets = self . _db_plugin . get_subnets ( <EOL> admin_ctx , filters = subnet_filter ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( internal_subnets ) ) <EOL> def test_ip_address_owner_update ( self ) : <EOL> l3p = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( name = '<STR_LIT>' , <EOL> l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . _bind_port_to_host ( pt2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> ip_owner_info = { '<STR_LIT:port>' : pt1 [ '<STR_LIT>' ] , '<STR_LIT>' : '<STR_LIT>' } <EOL> self . driver . _notify_port_update = mock . Mock ( ) <EOL> self . driver . ip_address_owner_update ( context . get_admin_context ( ) , <EOL> ip_owner_info = ip_owner_info , host = '<STR_LIT>' ) <EOL> obj = self . driver . ha_ip_handler . get_port_for_ha_ipaddress ( <EOL> '<STR_LIT>' , l2p [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( pt1 [ '<STR_LIT>' ] , obj [ '<STR_LIT>' ] ) <EOL> self . driver . _notify_port_update . assert_called_with ( mock . ANY , <EOL> pt1 [ '<STR_LIT>' ] ) <EOL> self . driver . _notify_port_update . reset_mock ( ) <EOL> ip_owner_info [ '<STR_LIT:port>' ] = pt2 [ '<STR_LIT>' ] <EOL> self . driver . ip_address_owner_update ( context . get_admin_context ( ) , <EOL> ip_owner_info = ip_owner_info , host = '<STR_LIT>' ) <EOL> obj = self . driver . ha_ip_handler . get_port_for_ha_ipaddress ( <EOL> '<STR_LIT>' , l2p [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( pt2 [ '<STR_LIT>' ] , obj [ '<STR_LIT>' ] ) <EOL> exp_calls = [ <EOL> mock . call ( mock . ANY , pt1 [ '<STR_LIT>' ] ) , <EOL> mock . call ( mock . ANY , pt2 [ '<STR_LIT>' ] ) ] <EOL> self . _check_call_list ( exp_calls , <EOL> self . driver . _notify_port_update . call_args_list ) <EOL> def test_enhanced_subnet_options ( self ) : <EOL> self . driver . enable_metadata_opt = False <EOL> l3p = self . create_l3_policy ( name = '<STR_LIT>' , <EOL> ip_pool = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( name = '<STR_LIT>' , <EOL> l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> sub = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , <EOL> self . api ) <EOL> with self . port ( subnet = sub , device_owner = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' ) as dhcp : <EOL> dhcp = dhcp [ '<STR_LIT:port>' ] <EOL> details = self . driver . get_gbp_details ( <EOL> context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt1 [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( details [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( [ dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( [ dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> def test_update_l2p_inject_default_route_false ( self ) : <EOL> self . driver . enable_metadata_opt = False <EOL> l3p = self . create_l3_policy ( name = '<STR_LIT>' , <EOL> ip_pool = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( name = '<STR_LIT>' , <EOL> l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> sub = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , <EOL> self . api ) <EOL> more_host_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> data = { '<STR_LIT>' : { '<STR_LIT>' : more_host_routes } } <EOL> req = self . new_update_request ( '<STR_LIT>' , data , sub [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = self . deserialize ( self . fmt , req . get_response ( self . api ) ) <EOL> self . assertEqual ( sorted ( res [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> sorted ( more_host_routes ) ) <EOL> with self . port ( subnet = sub , device_owner = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' ) as dhcp : <EOL> dhcp = dhcp [ '<STR_LIT:port>' ] <EOL> details = self . driver . get_gbp_details ( <EOL> context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt1 [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( details [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( [ dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( [ dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' in details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> data = { '<STR_LIT>' : { '<STR_LIT>' : False } } <EOL> res = self . new_update_request ( '<STR_LIT>' , data , l2p [ '<STR_LIT:id>' ] , <EOL> self . fmt ) . get_response ( self . ext_api ) <EOL> pt2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> with self . port ( subnet = sub , tenant_id = '<STR_LIT>' ) : <EOL> details = self . driver . get_gbp_details ( <EOL> context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt2 [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( details [ '<STR_LIT>' ] ) ) <EOL> self . assertFalse ( { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertTrue ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( [ dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( '<STR_LIT>' in details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> def test_create_l2p_inject_default_route_false ( self ) : <EOL> self . driver . enable_metadata_opt = False <EOL> l3p = self . create_l3_policy ( name = '<STR_LIT>' , <EOL> ip_pool = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( name = '<STR_LIT>' , <EOL> l3_policy_id = l3p [ '<STR_LIT:id>' ] , <EOL> inject_default_route = False ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> sub = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , <EOL> self . api ) <EOL> more_host_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> data = { '<STR_LIT>' : { '<STR_LIT>' : more_host_routes } } <EOL> req = self . new_update_request ( '<STR_LIT>' , data , sub [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> res = self . deserialize ( self . fmt , req . get_response ( self . api ) ) <EOL> self . assertEqual ( sorted ( res [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> sorted ( more_host_routes ) ) <EOL> with self . port ( subnet = sub , device_owner = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' ) as dhcp : <EOL> dhcp = dhcp [ '<STR_LIT:port>' ] <EOL> details = self . driver . get_gbp_details ( <EOL> context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt1 [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( details [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( [ dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertTrue ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } in <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( [ dhcp [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( '<STR_LIT>' in details [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> def test_get_gbp_details_error ( self ) : <EOL> details = self . driver . get_gbp_details ( <EOL> context . get_admin_context ( ) , device = '<STR_LIT>' % '<STR_LIT>' , <EOL> host = '<STR_LIT>' ) <EOL> self . assertEqual ( None , details ) <EOL> ptg = self . create_policy_target_group ( ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . driver . _get_owned_addresses = mock . Mock ( side_effect = Exception ) <EOL> details = self . driver . get_gbp_details ( <EOL> context . get_admin_context ( ) , device = '<STR_LIT>' % pt1 [ '<STR_LIT>' ] , <EOL> host = '<STR_LIT>' ) <EOL> self . assertEqual ( { '<STR_LIT>' : '<STR_LIT>' % pt1 [ '<STR_LIT>' ] } , details ) <EOL> def test_get_gbp_proxy_details ( self ) : <EOL> l3p_fake = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_fake = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg_fake = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_bound = self . create_policy_target ( <EOL> policy_target_group_id = ptg_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> l3p_real = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_real = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_real [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg_real = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_real [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_unbound = self . create_policy_target ( <EOL> policy_target_group_id = ptg_real [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . update_policy_target ( <EOL> pt_bound [ '<STR_LIT:id>' ] , <EOL> description = amap . PROXY_PORT_PREFIX + pt_unbound [ '<STR_LIT>' ] ) <EOL> port_unbound = self . _get_object ( '<STR_LIT>' , pt_unbound [ '<STR_LIT>' ] , <EOL> self . api ) [ '<STR_LIT:port>' ] <EOL> self . _bind_port_to_host ( pt_bound [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> mapping = self . driver . get_gbp_details ( context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % pt_bound [ '<STR_LIT>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( pt_bound [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' % pt_bound [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( ptg_real [ '<STR_LIT:id>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( l3p_real [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( l3p_real [ '<STR_LIT:id>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( port_unbound [ '<STR_LIT>' ] , mapping [ '<STR_LIT>' ] ) <EOL> def test_get_gbp_details_shadow ( self ) : <EOL> l2p = self . create_l2_policy ( ) [ '<STR_LIT>' ] <EOL> network = self . _get_object ( '<STR_LIT>' , l2p [ '<STR_LIT>' ] , self . api ) <EOL> with self . subnet ( network = network ) as sub : <EOL> with self . port ( subnet = sub ) as port : <EOL> self . _bind_port_to_host ( port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> mapping = self . driver . get_gbp_details ( <EOL> context . get_admin_context ( ) , <EOL> device = '<STR_LIT>' % port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , host = '<STR_LIT>' ) <EOL> self . assertEqual ( port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , mapping [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( amap . SHADOW_PREFIX + l2p [ '<STR_LIT:id>' ] , <EOL> mapping [ '<STR_LIT>' ] ) <EOL> def test_explicit_port ( self ) : <EOL> with self . network ( ) as net : <EOL> with self . subnet ( network = net ) as sub : <EOL> with self . port ( subnet = sub ) as port : <EOL> self . _bind_port_to_host ( port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> l2p = self . create_l2_policy ( <EOL> network_id = net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . create_policy_target ( <EOL> port_id = port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) <EOL> self . assertTrue ( self . driver . notifier . port_update . called ) <EOL> def test_port_notified_on_changed_ptg ( self ) : <EOL> ptg = self . create_policy_target_group ( ) [ '<STR_LIT>' ] <EOL> ptg2 = self . create_policy_target_group ( <EOL> l2_policy_id = ptg [ '<STR_LIT>' ] ) [ '<STR_LIT>' ] <EOL> pt = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . driver . notifier . port_update . reset_mock ( ) <EOL> self . update_policy_target ( pt [ '<STR_LIT:id>' ] , policy_target_group_id = ptg2 [ '<STR_LIT:id>' ] ) <EOL> self . assertTrue ( self . driver . notifier . port_update . called ) <EOL> def test_update_ptg_failed ( self ) : <EOL> ptg = self . create_policy_target_group ( ) [ '<STR_LIT>' ] <EOL> ptg2 = self . create_policy_target_group ( ) [ '<STR_LIT>' ] <EOL> pt = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> res = self . update_policy_target ( <EOL> pt [ '<STR_LIT:id>' ] , policy_target_group_id = ptg2 [ '<STR_LIT:id>' ] , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_port_notified_on_subnet_change ( self ) : <EOL> ptg = self . create_policy_target_group ( ) [ '<STR_LIT>' ] <EOL> pt = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . _bind_port_to_host ( pt [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> subnet = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , self . api ) <EOL> subnet2 = copy . deepcopy ( subnet ) <EOL> subnet2 [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . driver . apic_manager . reset_mock ( ) <EOL> self . driver . notifier . port_update . reset_mock ( ) <EOL> self . driver . process_subnet_changed ( context . get_admin_context ( ) , <EOL> subnet [ '<STR_LIT>' ] , subnet2 [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( self . driver . notifier . port_update . called ) <EOL> def test_get_gbp_proxy_address_ownership ( self ) : <EOL> l3p_fake = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_fake = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg_fake = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_bound_1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_bound_2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_bound_3 = self . create_policy_target ( <EOL> policy_target_group_id = ptg_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> l3p_real = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_real_1 = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_real [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> l2p_real_2 = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_real [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg_real_1 = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_real_1 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg_real_2 = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_real_2 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_unbound_1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg_real_1 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_unbound_2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg_real_2 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt_unbound_2_1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg_real_2 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . update_policy_target ( <EOL> pt_bound_1 [ '<STR_LIT:id>' ] , <EOL> description = amap . PROXY_PORT_PREFIX + pt_unbound_1 [ '<STR_LIT>' ] ) <EOL> self . update_policy_target ( <EOL> pt_bound_2 [ '<STR_LIT:id>' ] , <EOL> description = amap . PROXY_PORT_PREFIX + pt_unbound_2 [ '<STR_LIT>' ] ) <EOL> self . update_policy_target ( <EOL> pt_bound_3 [ '<STR_LIT:id>' ] , <EOL> description = amap . PROXY_PORT_PREFIX + pt_unbound_2_1 [ '<STR_LIT>' ] ) <EOL> self . driver . update_ip_owner ( { '<STR_LIT:port>' : pt_bound_1 [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . driver . update_ip_owner ( { '<STR_LIT:port>' : pt_bound_2 [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> entries = self . driver . ha_ip_handler . session . query ( <EOL> ha_ip_db . HAIPAddressToPortAssocation ) . all ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( entries ) ) <EOL> self . assertEqual ( '<STR_LIT>' , entries [ <NUM_LIT:0> ] . ha_ip_address ) <EOL> self . assertEqual ( '<STR_LIT>' , entries [ <NUM_LIT:1> ] . ha_ip_address ) <EOL> self . driver . update_ip_owner ( { '<STR_LIT:port>' : pt_bound_3 [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> entries = self . driver . ha_ip_handler . session . query ( <EOL> ha_ip_db . HAIPAddressToPortAssocation ) . all ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( entries ) ) <EOL> self . assertEqual ( '<STR_LIT>' , entries [ <NUM_LIT:0> ] . ha_ip_address ) <EOL> self . assertEqual ( '<STR_LIT>' , entries [ <NUM_LIT:1> ] . ha_ip_address ) <EOL> class FakeNetworkContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , network , segments ) : <EOL> self . _network = network <EOL> self . _segments = segments <EOL> self . _plugin_context = mock . Mock ( ) <EOL> @ property <EOL> def current ( self ) : <EOL> return self . _network <EOL> @ property <EOL> def network_segments ( self ) : <EOL> return self . _segments <EOL> class FakePortContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , port , network ) : <EOL> self . _port = port <EOL> self . _network = network <EOL> self . _plugin = mock . Mock ( ) <EOL> self . _plugin_context = mock . Mock ( ) <EOL> self . _plugin . get_ports . return_value = [ ] <EOL> if network . network_segments : <EOL> self . _bound_segment = network . network_segments [ <NUM_LIT:0> ] <EOL> else : <EOL> self . _bound_segment = None <EOL> self . current = self . _port <EOL> self . original = self . _port <EOL> self . network = self . _network <EOL> self . top_bound_segment = self . _bound_segment <EOL> self . host = self . _port . get ( portbindings . HOST_ID ) <EOL> self . original_host = None <EOL> self . _binding = mock . Mock ( ) <EOL> self . _binding . segment = self . _bound_segment <EOL> def set_binding ( self , segment_id , vif_type , cap_port_filter ) : <EOL> pass <EOL> class TestPolicyTargetDvs ( ApicMappingTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestPolicyTargetDvs , self ) . setUp ( ) <EOL> self . driver . apic_manager . app_profile_name = mocked . APIC_AP <EOL> plugin = manager . NeutronManager . get_plugin ( ) <EOL> self . ml2 = plugin . mechanism_manager . mech_drivers [ '<STR_LIT>' ] . obj <EOL> self . ml2 . _dvs_notifier = mock . MagicMock ( ) <EOL> self . ml2 . dvs_notifier . bind_port_call = mock . Mock ( <EOL> return_value = BOOKED_PORT_VALUE ) <EOL> mapper = self . driver . name_mapper <EOL> mapper . name_mapper . policy_taget_group . return_value = '<STR_LIT>' <EOL> def _verify_dvs_notifier ( self , notifier , port , host ) : <EOL> try : <EOL> dvs_mock = eval ( '<STR_LIT>' + notifier ) <EOL> except Exception : <EOL> self . assertTrue ( False , <EOL> "<STR_LIT>" + notifier + "<STR_LIT>" ) <EOL> return <EOL> self . assertTrue ( dvs_mock . called ) <EOL> a1 , a2 , a3 , a4 = dvs_mock . call_args [ <NUM_LIT:0> ] <EOL> self . assertEqual ( a1 [ '<STR_LIT:id>' ] , port [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( a2 [ '<STR_LIT:id>' ] , port [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( a4 , host ) <EOL> def _pg_name ( self , project , profile , network ) : <EOL> return ( str ( project ) + '<STR_LIT:|>' + str ( profile ) + '<STR_LIT:|>' + network ) <EOL> def test_bind_port_dvs ( self ) : <EOL> self . agent_conf = AGENT_CONF_DVS <EOL> l3p_fake = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_fake = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> newp1 = self . _bind_port_to_host ( pt [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> vif_details = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNotNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> pg = self . _pg_name ( ptg [ '<STR_LIT>' ] , mocked . APIC_AP , ptg [ '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( pg , vif_details . get ( '<STR_LIT>' ) ) <EOL> port_key = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( port_key ) <EOL> self . assertEqual ( port_key , BOOKED_PORT_VALUE ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , <EOL> newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> net_ctx = FakeNetworkContext ( mock . Mock ( ) , [ mock . Mock ( ) ] ) <EOL> port_ctx = FakePortContext ( newp1 [ '<STR_LIT:port>' ] , net_ctx ) <EOL> self . ml2 . delete_port_postcommit ( port_ctx ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> def test_bind_port_dvs_with_opflex_different_hosts ( self ) : <EOL> l3p_fake = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_fake = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . agent_conf = AGENT_CONF <EOL> pt2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> newp2 = self . _bind_port_to_host ( pt2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> vif_details = newp2 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . agent_conf = AGENT_CONF_DVS <EOL> self . ml2 . _dvs_notifier . reset_mock ( ) <EOL> newp1 = self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> port_key = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( port_key ) <EOL> self . assertEqual ( port_key , BOOKED_PORT_VALUE ) <EOL> vif_details = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNotNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> pg = self . _pg_name ( ptg [ '<STR_LIT>' ] , mocked . APIC_AP , ptg [ '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( pg , vif_details . get ( '<STR_LIT>' ) ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , <EOL> newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> net_ctx = FakeNetworkContext ( mock . Mock ( ) , [ mock . Mock ( ) ] ) <EOL> port_ctx = FakePortContext ( newp1 [ '<STR_LIT:port>' ] , net_ctx ) <EOL> self . ml2 . delete_port_postcommit ( port_ctx ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> def test_bind_ports_opflex_same_host ( self ) : <EOL> l3p_fake = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_fake = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> newp1 = self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> vif_details = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> port_key = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNone ( port_key ) <EOL> dvs_mock = self . ml2 . dvs_notifier . update_postcommit_port_call <EOL> dvs_mock . assert_not_called ( ) <EOL> net_ctx = FakeNetworkContext ( mock . Mock ( ) , [ mock . Mock ( ) ] ) <EOL> port_ctx = FakePortContext ( newp1 [ '<STR_LIT:port>' ] , net_ctx ) <EOL> self . ml2 . delete_port_postcommit ( port_ctx ) <EOL> dvs_mock = self . ml2 . dvs_notifier . delete_port_call <EOL> dvs_mock . assert_not_called ( ) <EOL> self . ml2 . dvs_notifier . reset_mock ( ) <EOL> pt2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> newp2 = self . _bind_port_to_host ( pt2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> vif_details = newp2 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> port_key = newp2 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNone ( port_key ) <EOL> dvs_mock . assert_not_called ( ) <EOL> net_ctx = FakeNetworkContext ( mock . Mock ( ) , [ mock . Mock ( ) ] ) <EOL> port_ctx = FakePortContext ( newp2 [ '<STR_LIT:port>' ] , net_ctx ) <EOL> self . ml2 . delete_port_postcommit ( port_ctx ) <EOL> dvs_mock = self . ml2 . dvs_notifier . delete_port_call <EOL> dvs_mock . assert_not_called ( ) <EOL> def test_bind_ports_dvs_with_opflex_same_host ( self ) : <EOL> self . agent_conf = AGENT_CONF_DVS <EOL> l3p_fake = self . create_l3_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l2p_fake = self . create_l2_policy ( <EOL> name = '<STR_LIT>' , l3_policy_id = l3p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_fake [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> pt1 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> newp1 = self . _bind_port_to_host ( pt1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> vif_details = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNotNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> port_key = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( port_key ) <EOL> self . assertEqual ( port_key , BOOKED_PORT_VALUE ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , <EOL> newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> net_ctx = FakeNetworkContext ( mock . Mock ( ) , [ mock . Mock ( ) ] ) <EOL> port_ctx = FakePortContext ( newp1 [ '<STR_LIT:port>' ] , net_ctx ) <EOL> self . ml2 . delete_port_postcommit ( port_ctx ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> self . ml2 . dvs_notifier . reset_mock ( ) <EOL> self . agent_conf = AGENT_CONF <EOL> pt2 = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> newp2 = self . _bind_dhcp_port_to_host ( pt2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> vif_details = newp2 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> port_key = newp2 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNone ( port_key ) <EOL> dvs_mock = self . ml2 . dvs_notifier . update_postcommit_port_call <EOL> dvs_mock . assert_not_called ( ) <EOL> net_ctx = FakeNetworkContext ( mock . Mock ( ) , [ mock . Mock ( ) ] ) <EOL> port_ctx = FakePortContext ( newp2 [ '<STR_LIT:port>' ] , net_ctx ) <EOL> self . ml2 . delete_port_postcommit ( port_ctx ) <EOL> dvs_mock = self . ml2 . dvs_notifier . delete_port_call <EOL> dvs_mock . assert_not_called ( ) <EOL> def test_bind_port_dvs_shared ( self ) : <EOL> self . agent_conf = AGENT_CONF_DVS <EOL> ptg = self . create_policy_target_group ( shared = True , <EOL> name = "<STR_LIT>" ) [ '<STR_LIT>' ] <EOL> pt = self . create_policy_target ( <EOL> policy_target_group_id = ptg [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> newp1 = self . _bind_port_to_host ( pt [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> vif_details = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> self . assertIsNotNone ( vif_details . get ( '<STR_LIT>' ) ) <EOL> pg = self . _pg_name ( amap . apic_manager . TENANT_COMMON , <EOL> mocked . APIC_AP , ptg [ '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( pg , vif_details . get ( '<STR_LIT>' ) ) <EOL> port_key = newp1 [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( port_key ) <EOL> self . assertEqual ( port_key , BOOKED_PORT_VALUE ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , <EOL> newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> net_ctx = FakeNetworkContext ( mock . Mock ( ) , [ mock . Mock ( ) ] ) <EOL> port_ctx = FakePortContext ( newp1 [ '<STR_LIT:port>' ] , net_ctx ) <EOL> self . ml2 . delete_port_postcommit ( port_ctx ) <EOL> self . _verify_dvs_notifier ( '<STR_LIT>' , newp1 [ '<STR_LIT:port>' ] , '<STR_LIT>' ) <EOL> class TestPolicyTargetGroup ( ApicMappingTestCase ) : <EOL> def _test_policy_target_group_created_on_apic ( self , shared = False ) : <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , shared = shared ) [ '<STR_LIT>' ] <EOL> tenant = self . common_tenant if shared else ptg [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> expected_calls = [ <EOL> mock . call ( tenant , ptg [ '<STR_LIT:id>' ] , bd_name = ptg [ '<STR_LIT>' ] , <EOL> bd_owner = tenant ) , <EOL> mock . call ( tenant , amap . SHADOW_PREFIX + ptg [ '<STR_LIT>' ] , <EOL> bd_name = ptg [ '<STR_LIT>' ] , bd_owner = tenant , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . ensure_epg_created . call_args_list ) <EOL> def test_policy_target_group_created_on_apic ( self ) : <EOL> self . _test_policy_target_group_created_on_apic ( ) <EOL> def test_policy_target_group_created_on_apic_shared ( self ) : <EOL> self . _test_policy_target_group_created_on_apic ( shared = True ) <EOL> def _test_ptg_policy_rule_set_created ( self , provider = True , shared = False ) : <EOL> cntr = self . create_policy_rule_set ( name = '<STR_LIT:c>' , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . set_contract_for_epg . reset_mock ( ) <EOL> if provider : <EOL> ptg = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] , <EOL> provided_policy_rule_sets = { cntr [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) [ <EOL> '<STR_LIT>' ] <EOL> else : <EOL> ptg = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] , <EOL> consumed_policy_rule_sets = { cntr [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) [ <EOL> '<STR_LIT>' ] <EOL> ct_owner = self . common_tenant if shared else cntr [ '<STR_LIT>' ] <EOL> expected_calls = [ <EOL> mock . call ( <EOL> ptg [ '<STR_LIT>' ] , ptg [ '<STR_LIT:id>' ] , cntr [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY , contract_owner = ct_owner , <EOL> provider = provider ) , <EOL> mock . call ( <EOL> ptg [ '<STR_LIT>' ] , ptg [ '<STR_LIT:id>' ] , <EOL> amap . SERVICE_PREFIX + ptg [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY , contract_owner = ptg [ '<STR_LIT>' ] , <EOL> provider = False ) , <EOL> mock . call ( <EOL> ptg [ '<STR_LIT>' ] , ptg [ '<STR_LIT:id>' ] , <EOL> amap . IMPLICIT_PREFIX + ptg [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY , contract_owner = ptg [ '<STR_LIT>' ] , <EOL> provider = True ) , <EOL> mock . call ( <EOL> ptg [ '<STR_LIT>' ] , ptg [ '<STR_LIT:id>' ] , <EOL> amap . IMPLICIT_PREFIX + ptg [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY , contract_owner = ptg [ '<STR_LIT>' ] , <EOL> provider = False ) ] <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . set_contract_for_epg . call_args_list ) <EOL> def _test_ptg_policy_rule_set_updated ( self , provider = True , shared = False ) : <EOL> p_or_c = { True : '<STR_LIT>' , <EOL> False : '<STR_LIT>' } <EOL> cntr = self . create_policy_rule_set ( <EOL> name = '<STR_LIT>' , shared = shared ) [ '<STR_LIT>' ] <EOL> new_cntr = self . create_policy_rule_set ( <EOL> name = '<STR_LIT>' , shared = shared ) [ '<STR_LIT>' ] <EOL> if provider : <EOL> ptg = self . create_policy_target_group ( <EOL> provided_policy_rule_sets = { cntr [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) <EOL> else : <EOL> ptg = self . create_policy_target_group ( <EOL> consumed_policy_rule_sets = { cntr [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) <EOL> data = { '<STR_LIT>' : { p_or_c [ provider ] : <EOL> { new_cntr [ '<STR_LIT:id>' ] : '<STR_LIT>' } } } <EOL> req = self . new_update_request ( '<STR_LIT>' , data , <EOL> ptg [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> self . fmt ) <EOL> ptg = self . deserialize ( self . fmt , req . get_response ( self . ext_api ) ) <EOL> ptg = ptg [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> ct_owner = self . common_tenant if shared else cntr [ '<STR_LIT>' ] <EOL> mgr . set_contract_for_epg . assert_called_with ( <EOL> ptg [ '<STR_LIT>' ] , ptg [ '<STR_LIT:id>' ] , new_cntr [ '<STR_LIT:id>' ] , <EOL> contract_owner = ct_owner , transaction = mock . ANY , <EOL> provider = provider ) <EOL> mgr . unset_contract_for_epg . assert_called_with ( <EOL> ptg [ '<STR_LIT>' ] , ptg [ '<STR_LIT:id>' ] , cntr [ '<STR_LIT:id>' ] , <EOL> contract_owner = ct_owner , <EOL> transaction = mock . ANY , provider = provider ) <EOL> def test_ptg_policy_rule_set_provider_created ( self ) : <EOL> self . _test_ptg_policy_rule_set_created ( ) <EOL> def test_ptg_policy_rule_set_provider_updated ( self ) : <EOL> self . _test_ptg_policy_rule_set_updated ( ) <EOL> def test_ptg_policy_rule_set_consumer_created ( self ) : <EOL> self . _test_ptg_policy_rule_set_created ( False ) <EOL> def test_ptg_policy_rule_set_consumer_updated ( self ) : <EOL> self . _test_ptg_policy_rule_set_updated ( False ) <EOL> def test_ptg_policy_rule_set_provider_created_shared ( self ) : <EOL> self . _test_ptg_policy_rule_set_created ( shared = True ) <EOL> def test_ptg_policy_rule_set_provider_updated_shared ( self ) : <EOL> self . _test_ptg_policy_rule_set_updated ( shared = True ) <EOL> def test_ptg_policy_rule_set_consumer_created_shared ( self ) : <EOL> self . _test_ptg_policy_rule_set_created ( False , shared = True ) <EOL> def test_ptg_policy_rule_set_consumer_updated_shared ( self ) : <EOL> self . _test_ptg_policy_rule_set_updated ( False , shared = True ) <EOL> def _test_policy_target_group_deleted_on_apic ( self , shared = False ) : <EOL> ptg = self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , shared = shared ) [ '<STR_LIT>' ] <EOL> req = self . new_delete_request ( '<STR_LIT>' , <EOL> ptg [ '<STR_LIT:id>' ] , self . fmt ) <EOL> req . get_response ( self . ext_api ) <EOL> mgr = self . driver . apic_manager <EOL> tenant = self . common_tenant if shared else ptg [ '<STR_LIT>' ] <EOL> expected_calls = [ <EOL> mock . call ( tenant , ptg [ '<STR_LIT:id>' ] ) , <EOL> mock . call ( tenant , amap . SHADOW_PREFIX + ptg [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . delete_epg_for_network . call_args_list ) <EOL> def test_policy_target_group_deleted_on_apic ( self ) : <EOL> self . _test_policy_target_group_deleted_on_apic ( ) <EOL> def test_policy_target_group_deleted_on_apic_shared ( self ) : <EOL> self . _test_policy_target_group_deleted_on_apic ( shared = True ) <EOL> def _test_policy_target_group_subnet_created_on_apic ( self , shared = False ) : <EOL> ptg = self . _create_explicit_subnet_ptg ( '<STR_LIT>' , shared = shared ) <EOL> mgr = self . driver . apic_manager <EOL> tenant = self . common_tenant if shared else ptg [ '<STR_LIT>' ] <EOL> mgr . ensure_subnet_created_on_apic . assert_called_once_with ( <EOL> tenant , ptg [ '<STR_LIT>' ] , '<STR_LIT>' , <EOL> transaction = mock . ANY ) <EOL> def test_policy_target_group_subnet_created_on_apic ( self ) : <EOL> self . _test_policy_target_group_subnet_created_on_apic ( ) <EOL> def test_policy_target_group_subnet_created_on_apic_shared ( self ) : <EOL> self . _test_policy_target_group_subnet_created_on_apic ( shared = True ) <EOL> def _test_policy_target_group_subnet_added ( self , shared = False ) : <EOL> ptg = self . _create_explicit_subnet_ptg ( '<STR_LIT>' , shared = shared ) <EOL> l2p = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] , <EOL> self . ext_api ) <EOL> network = self . _get_object ( '<STR_LIT>' , l2p [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> self . api ) <EOL> with self . subnet ( network = network , cidr = '<STR_LIT>' ) as subnet : <EOL> data = { '<STR_LIT>' : <EOL> { '<STR_LIT>' : ptg [ '<STR_LIT>' ] + [ subnet [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ] } } <EOL> mgr = self . driver . apic_manager <EOL> self . new_update_request ( '<STR_LIT>' , data , ptg [ '<STR_LIT:id>' ] , <EOL> self . fmt ) . get_response ( self . ext_api ) <EOL> tenant = self . common_tenant if shared else ptg [ '<STR_LIT>' ] <EOL> mgr . ensure_subnet_created_on_apic . assert_called_with ( <EOL> tenant , ptg [ '<STR_LIT>' ] , '<STR_LIT>' , <EOL> transaction = mock . ANY ) <EOL> def test_policy_target_group_subnet_added ( self ) : <EOL> self . _test_policy_target_group_subnet_added ( ) <EOL> def test_policy_target_group_subnet_added_shared ( self ) : <EOL> self . _test_policy_target_group_subnet_added ( shared = True ) <EOL> def _test_process_subnet_update ( self , shared = False ) : <EOL> ptg = self . _create_explicit_subnet_ptg ( '<STR_LIT>' , shared = shared ) <EOL> subnet = self . _get_object ( '<STR_LIT>' , ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , self . api ) <EOL> subnet2 = copy . deepcopy ( subnet ) <EOL> subnet2 [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> mgr = self . driver . apic_manager <EOL> mgr . reset_mock ( ) <EOL> self . driver . process_subnet_changed ( context . get_admin_context ( ) , <EOL> subnet [ '<STR_LIT>' ] , subnet2 [ '<STR_LIT>' ] ) <EOL> tenant = self . common_tenant if shared else ptg [ '<STR_LIT>' ] <EOL> mgr . ensure_subnet_created_on_apic . assert_called_once_with ( <EOL> tenant , ptg [ '<STR_LIT>' ] , '<STR_LIT>' , <EOL> transaction = mock . ANY ) <EOL> mgr . ensure_subnet_deleted_on_apic . assert_called_with ( <EOL> tenant , ptg [ '<STR_LIT>' ] , '<STR_LIT>' , <EOL> transaction = mock . ANY ) <EOL> def test_process_subnet_update ( self ) : <EOL> self . _test_process_subnet_update ( ) <EOL> def test_process_subnet_update_shared ( self ) : <EOL> self . _test_process_subnet_update ( shared = True ) <EOL> def test_multiple_ptg_per_l2p ( self ) : <EOL> l2p = self . create_l2_policy ( ) [ '<STR_LIT>' ] <EOL> ptg1 = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg2 = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( ptg1 [ '<STR_LIT>' ] , ptg2 [ '<STR_LIT>' ] ) <EOL> def test_force_add_subnet ( self ) : <EOL> l2p = self . create_l2_policy ( ) [ '<STR_LIT>' ] <EOL> ptg1 = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg2 = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ctx = p_context . PolicyTargetGroupContext ( <EOL> self . driver . gbp_plugin , context . get_admin_context ( ) , ptg2 ) <EOL> self . driver . _use_implicit_subnet ( ctx , force_add = True ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ctx . current [ '<STR_LIT>' ] ) ) <EOL> ptg1 = self . show_policy_target_group ( ptg1 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ptg1 [ '<STR_LIT>' ] ) ) <EOL> ptg2 = self . show_policy_target_group ( ptg2 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ptg2 [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( set ( ptg1 [ '<STR_LIT>' ] ) , set ( ptg2 [ '<STR_LIT>' ] ) ) <EOL> self . assertNotEqual ( ptg2 [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , ptg2 [ '<STR_LIT>' ] [ <NUM_LIT:1> ] ) <EOL> def test_subnets_unique_per_l3p ( self ) : <EOL> l3p = self . create_l3_policy ( shared = True , tenant_id = '<STR_LIT>' , <EOL> is_admin_context = True ) [ '<STR_LIT>' ] <EOL> l2p1 = self . create_l2_policy ( <EOL> tenant_id = '<STR_LIT>' , l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> l2p2 = self . create_l2_policy ( <EOL> tenant_id = '<STR_LIT>' , l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg1 = self . create_policy_target_group ( <EOL> tenant_id = '<STR_LIT>' , l2_policy_id = l2p1 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg2 = self . create_policy_target_group ( <EOL> tenant_id = '<STR_LIT>' , l2_policy_id = l2p2 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> sub_ptg_1 = set ( self . _get_object ( '<STR_LIT>' , <EOL> x , self . api ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for x in ptg1 [ '<STR_LIT>' ] ) <EOL> sub_ptg_2 = set ( self . _get_object ( '<STR_LIT>' , <EOL> x , self . api ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for x in ptg2 [ '<STR_LIT>' ] ) <EOL> self . assertNotEqual ( sub_ptg_1 , sub_ptg_2 ) <EOL> self . assertFalse ( sub_ptg_1 & sub_ptg_2 ) <EOL> def test_preexisting_l2p_no_service_contracts ( self ) : <EOL> self . driver . name_mapper . has_valid_name = ( <EOL> self . driver . name_mapper . _is_apic_reference ) <EOL> self . driver . name_mapper . tenant = mock . Mock ( <EOL> return_value = self . _tenant_id ) <EOL> self . driver . name_mapper . dn_manager . decompose_bridge_domain = mock . Mock ( <EOL> return_value = [ '<STR_LIT>' ] ) <EOL> self . driver . _configure_epg_service_contract = mock . Mock ( ) <EOL> self . driver . _configure_epg_implicit_contract = mock . Mock ( ) <EOL> l2p = self . create_l2_policy ( name = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . create_policy_target_group ( l2_policy_id = l2p [ '<STR_LIT:id>' ] ) <EOL> self . assertFalse ( self . driver . _configure_epg_service_contract . called ) <EOL> self . assertFalse ( self . driver . _configure_epg_implicit_contract . called ) <EOL> self . create_policy_target_group ( ) <EOL> self . assertTrue ( self . driver . _configure_epg_service_contract . called ) <EOL> self . assertTrue ( self . driver . _configure_epg_implicit_contract . called ) <EOL> def _create_explicit_subnet_ptg ( self , cidr , shared = False ) : <EOL> l2p = self . create_l2_policy ( name = "<STR_LIT>" , shared = shared ) <EOL> l2p_id = l2p [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> network_id = l2p [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> network = self . _get_object ( '<STR_LIT>' , network_id , self . api ) <EOL> with self . subnet ( network = network , cidr = cidr ) : <EOL> return self . create_policy_target_group ( <EOL> name = "<STR_LIT>" , l2_policy_id = l2p_id , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> class TestL2Policy ( ApicMappingTestCase ) : <EOL> def _test_l2_policy_created_on_apic ( self , shared = False ) : <EOL> l2p = self . create_l2_policy ( name = "<STR_LIT>" , shared = shared ) [ '<STR_LIT>' ] <EOL> tenant = self . common_tenant if shared else l2p [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . ensure_bd_created_on_apic . assert_called_once_with ( <EOL> tenant , l2p [ '<STR_LIT:id>' ] , ctx_owner = tenant , ctx_name = l2p [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY ) <EOL> mgr . ensure_epg_created . assert_called_once_with ( <EOL> tenant , amap . SHADOW_PREFIX + l2p [ '<STR_LIT:id>' ] , bd_owner = tenant , <EOL> bd_name = l2p [ '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> def test_l2_policy_created_on_apic ( self ) : <EOL> self . _test_l2_policy_created_on_apic ( ) <EOL> def test_l2_policy_created_on_apic_shared ( self ) : <EOL> self . _test_l2_policy_created_on_apic ( shared = True ) <EOL> def _test_l2_policy_deleted_on_apic ( self , shared = False ) : <EOL> l2p = self . create_l2_policy ( name = "<STR_LIT>" , shared = shared ) [ '<STR_LIT>' ] <EOL> req = self . new_delete_request ( '<STR_LIT>' , l2p [ '<STR_LIT:id>' ] , self . fmt ) <EOL> req . get_response ( self . ext_api ) <EOL> tenant = self . common_tenant if shared else l2p [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . delete_bd_on_apic . assert_called_once_with ( <EOL> tenant , l2p [ '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> mgr . delete_epg_for_network . assert_called_once_with ( <EOL> tenant , amap . SHADOW_PREFIX + l2p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) <EOL> expected_calls = [ <EOL> mock . call ( amap . IMPLICIT_PREFIX + l2p [ '<STR_LIT:id>' ] , owner = tenant , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( amap . SERVICE_PREFIX + l2p [ '<STR_LIT:id>' ] , owner = tenant , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . delete_contract . call_args_list ) <EOL> def test_l2_policy_deleted_on_apic ( self ) : <EOL> self . _test_l2_policy_deleted_on_apic ( ) <EOL> def test_l2_policy_deleted_on_apic_shared ( self ) : <EOL> self . _test_l2_policy_deleted_on_apic ( shared = True ) <EOL> def test_pre_existing_subnets_added ( self ) : <EOL> with self . network ( ) as net : <EOL> with self . subnet ( network = net ) as sub : <EOL> sub = sub [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( <EOL> network_id = net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . ensure_subnet_created_on_apic . assert_called_with ( <EOL> l2p [ '<STR_LIT>' ] , l2p [ '<STR_LIT:id>' ] , <EOL> sub [ '<STR_LIT>' ] + '<STR_LIT:/>' + sub [ '<STR_LIT>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] , <EOL> transaction = mock . ANY ) <EOL> ptg = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( ptg [ '<STR_LIT>' ] , [ sub [ '<STR_LIT:id>' ] ] ) <EOL> def test_reject_l3p_update ( self ) : <EOL> l2p = self . create_l2_policy ( ) [ '<STR_LIT>' ] <EOL> new_l3p = self . create_l3_policy ( ) [ '<STR_LIT>' ] <EOL> res = self . update_l2_policy ( l2p [ '<STR_LIT:id>' ] , l3_policy_id = new_l3p [ '<STR_LIT:id>' ] , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_subnet_deallocated ( self ) : <EOL> l2p = self . create_l2_policy ( ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> subnet = netaddr . IPSet ( <EOL> [ self . _show_subnet ( x ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] for x in ptg [ '<STR_LIT>' ] ] ) <EOL> self . delete_policy_target_group ( ptg [ '<STR_LIT:id>' ] ) <EOL> l2p2 = self . create_l2_policy ( ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> l2_policy_id = l2p2 [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> subnet2 = netaddr . IPSet ( <EOL> [ self . _show_subnet ( x ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] for x in ptg [ '<STR_LIT>' ] ] ) <EOL> self . assertFalse ( subnet & subnet2 ) <EOL> class TestL3Policy ( ApicMappingTestCase ) : <EOL> def _test_l3_policy_created_on_apic ( self , shared = False ) : <EOL> l3p = self . create_l3_policy ( name = "<STR_LIT>" , shared = shared ) [ '<STR_LIT>' ] <EOL> tenant = self . common_tenant if shared else l3p [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . ensure_context_enforced . assert_called_once_with ( <EOL> tenant , l3p [ '<STR_LIT:id>' ] ) <EOL> def test_l3_policy_created_on_apic ( self ) : <EOL> self . _test_l3_policy_created_on_apic ( ) <EOL> def test_l3_policy_created_on_apic_shared ( self ) : <EOL> self . _test_l3_policy_created_on_apic ( shared = True ) <EOL> def _test_l3_policy_deleted_on_apic ( self , shared = False ) : <EOL> l3p = self . create_l3_policy ( name = "<STR_LIT>" , shared = shared ) [ '<STR_LIT>' ] <EOL> req = self . new_delete_request ( '<STR_LIT>' , l3p [ '<STR_LIT:id>' ] , self . fmt ) <EOL> req . get_response ( self . ext_api ) <EOL> tenant = self . common_tenant if shared else l3p [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . ensure_context_deleted . assert_called_once_with ( <EOL> tenant , l3p [ '<STR_LIT:id>' ] ) <EOL> def test_l3_policy_deleted_on_apic ( self ) : <EOL> self . _test_l3_policy_deleted_on_apic ( ) <EOL> def test_l3_policy_deleted_on_apic_shared ( self ) : <EOL> self . _test_l3_policy_deleted_on_apic ( shared = True ) <EOL> def _test_multiple_l3_policy_per_es ( self , shared_es = False ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( name = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , shared = shared_es ) [ '<STR_LIT>' ] <EOL> self . create_l3_policy ( external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> res = self . create_l3_policy ( <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> if self . nat_enabled else <NUM_LIT> ) <EOL> if self . nat_enabled : <EOL> es = self . show_external_segment ( es [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:2> , len ( es [ '<STR_LIT>' ] ) ) <EOL> else : <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> sneaky_l3p = self . create_l3_policy ( ) [ '<STR_LIT>' ] <EOL> self . update_l3_policy ( <EOL> sneaky_l3p [ '<STR_LIT:id>' ] , <EOL> expected_res_status = <NUM_LIT:200> if self . nat_enabled else <NUM_LIT> , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } ) <EOL> if self . nat_enabled : <EOL> es = self . show_external_segment ( es [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:3> , len ( es [ '<STR_LIT>' ] ) ) <EOL> else : <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_multiple_l3_policy_per_es ( self ) : <EOL> self . _test_multiple_l3_policy_per_es ( shared_es = False ) <EOL> def test_multiple_l3_policy_per_es_shared ( self ) : <EOL> self . _test_multiple_l3_policy_per_es ( shared_es = True ) <EOL> def test_one_l3_policy_ip_on_es ( self ) : <EOL> es = self . create_external_segment ( <EOL> cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> res = self . create_l3_policy ( <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' , '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> sneaky_l3p = self . create_l3_policy ( <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> res = self . update_l3_policy ( <EOL> sneaky_l3p [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT> , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' , '<STR_LIT>' ] } ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_router_interface_no_gateway ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> l3p = self . show_l3_policy ( l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( l3p [ '<STR_LIT>' ] ) ) <EOL> subnet = self . _show_subnet ( ptg [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) [ '<STR_LIT>' ] <EOL> router_ports = self . _list ( <EOL> '<STR_LIT>' , <EOL> query_params = '<STR_LIT>' % l3p [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:2> , len ( router_ports ) ) <EOL> for port in router_ports : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( port [ '<STR_LIT>' ] ) ) <EOL> self . assertNotEqual ( subnet [ '<STR_LIT>' ] , <EOL> port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertNotEqual ( router_ports [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> router_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( <EOL> router_ports [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == subnet [ '<STR_LIT:id>' ] or <EOL> router_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == subnet [ '<STR_LIT:id>' ] ) <EOL> def _test_l3p_plugged_to_es_at_creation ( self , shared_es , <EOL> shared_l3p , is_edge_nat = False ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> shared = shared_es , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( <EOL> name = '<STR_LIT>' , <EOL> shared = shared_l3p , <EOL> tenant_id = es [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( l3p [ '<STR_LIT>' ] [ es [ '<STR_LIT:id>' ] ] ) ) <EOL> self . assertEqual ( '<STR_LIT>' , l3p [ '<STR_LIT>' ] [ es [ '<STR_LIT:id>' ] ] [ <NUM_LIT:0> ] ) <EOL> owner = self . common_tenant if shared_es else es [ '<STR_LIT>' ] <EOL> l3p_owner = self . common_tenant if shared_l3p else l3p [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> call_name = mgr . ensure_external_routed_network_created <EOL> l3out_str = "<STR_LIT>" <EOL> if is_edge_nat : <EOL> l3out_str = "<STR_LIT>" <EOL> if self . nat_enabled : <EOL> expected_l3out_calls = [ ] <EOL> if not is_edge_nat or not self . pre_l3out : <EOL> expected_l3out_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , context = l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ) <EOL> if not self . pre_l3out : <EOL> expected_l3out_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , <EOL> context = "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ) <EOL> elif not self . pre_l3out : <EOL> expected_l3out_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , context = l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ] <EOL> else : <EOL> call_name = mgr . set_context_for_external_routed_network <EOL> expected_l3out_calls = [ <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , es [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( expected_l3out_calls , call_name . call_args_list ) <EOL> if is_edge_nat and self . nat_enabled : <EOL> ( self . driver . l3out_vlan_alloc . <EOL> reserve_vlan . assert_called_once_with ( <EOL> es [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) ) <EOL> if not self . pre_l3out : <EOL> expected_set_domain_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) ] <EOL> expected_logic_node_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , <EOL> mocked . APIC_EXT_MODULE , mocked . APIC_EXT_PORT , <EOL> mocked . APIC_EXT_ENCAP , '<STR_LIT>' , <EOL> owner = owner , router_id = APIC_EXTERNAL_RID , <EOL> transaction = mock . ANY ) ] <EOL> expected_route_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ] <EOL> if is_edge_nat and self . nat_enabled : <EOL> expected_set_domain_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> owner = owner , transaction = mock . ANY ) ) <EOL> expected_logic_node_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> mocked . APIC_EXT_SWITCH , mocked . APIC_EXT_MODULE , <EOL> mocked . APIC_EXT_PORT , mock . ANY , '<STR_LIT>' , <EOL> owner = owner , router_id = APIC_EXTERNAL_RID , <EOL> transaction = mock . ANY ) ) <EOL> expected_route_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ) <EOL> expected_route_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_set_domain_calls , <EOL> mgr . set_domain_for_external_routed_network . call_args_list ) <EOL> self . _check_call_list ( expected_logic_node_calls , <EOL> mgr . ensure_logical_node_profile_created . call_args_list ) <EOL> self . _check_call_list ( expected_route_calls , <EOL> mgr . ensure_static_route_created . call_args_list ) <EOL> else : <EOL> if is_edge_nat and self . nat_enabled : <EOL> final_req = re . sub ( '<STR_LIT>' , <EOL> l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , self . trimmed_l3out ) <EOL> mgr . apic . post_body . assert_called_once_with ( <EOL> mgr . apic . l3extOut . mo , final_req , l3p_owner , <EOL> l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) ) <EOL> self . assertFalse ( mgr . set_domain_for_external_routed_network . called ) <EOL> self . assertFalse ( mgr . ensure_logical_node_profile_created . called ) <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> if self . nat_enabled : <EOL> mgr . set_l3out_for_bd . assert_called_once_with ( owner , <EOL> "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> def test_l3p_plugged_to_es_at_creation_1 ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_creation ( shared_es = True , <EOL> shared_l3p = False ) <EOL> def test_l3p_plugged_to_es_at_creation_2 ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_creation ( shared_es = True , <EOL> shared_l3p = True ) <EOL> def test_l3p_plugged_to_es_at_creation_3 ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_creation ( shared_es = False , <EOL> shared_l3p = False ) <EOL> def test_l3p_plugged_to_es_at_creation_edge_nat_mode ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_creation ( shared_es = False , <EOL> shared_l3p = False , <EOL> is_edge_nat = True ) <EOL> def _test_l3p_plugged_to_es_at_update ( self , shared_es , <EOL> shared_l3p , is_edge_nat = False ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> shared = shared_es , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( <EOL> name = '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> , <EOL> tenant_id = es [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> shared = shared_l3p , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ ] } ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( l3p [ '<STR_LIT>' ] [ es [ '<STR_LIT:id>' ] ] ) ) <EOL> self . assertEqual ( '<STR_LIT>' , l3p [ '<STR_LIT>' ] [ es [ '<STR_LIT:id>' ] ] [ <NUM_LIT:0> ] ) <EOL> mgr = self . driver . apic_manager <EOL> owner = self . common_tenant if shared_es else es [ '<STR_LIT>' ] <EOL> l3p_owner = self . common_tenant if shared_l3p else l3p [ '<STR_LIT>' ] <EOL> l3out_str = "<STR_LIT>" <EOL> if is_edge_nat : <EOL> l3out_str = "<STR_LIT>" <EOL> expected_l3out_calls = [ ] <EOL> call_name = mgr . ensure_external_routed_network_created <EOL> if self . nat_enabled : <EOL> if not is_edge_nat or not self . pre_l3out : <EOL> expected_l3out_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , context = l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ) <EOL> if not self . pre_l3out : <EOL> expected_l3out_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , <EOL> context = "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ) <EOL> elif not self . pre_l3out : <EOL> expected_l3out_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , context = l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ] <EOL> else : <EOL> call_name = mgr . set_context_for_external_routed_network <EOL> expected_l3out_calls = [ <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , es [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( expected_l3out_calls , call_name . call_args_list ) <EOL> if is_edge_nat and self . nat_enabled : <EOL> ( self . driver . l3out_vlan_alloc . <EOL> reserve_vlan . assert_called_once_with ( <EOL> es [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) ) <EOL> if not self . pre_l3out : <EOL> expected_set_domain_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) ] <EOL> expected_logic_node_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , <EOL> mocked . APIC_EXT_MODULE , mocked . APIC_EXT_PORT , <EOL> mocked . APIC_EXT_ENCAP , '<STR_LIT>' , <EOL> owner = owner , router_id = APIC_EXTERNAL_RID , <EOL> transaction = mock . ANY ) ] <EOL> expected_route_calls = [ <EOL> mock . call ( es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , <EOL> '<STR_LIT>' , owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ] <EOL> if is_edge_nat and self . nat_enabled : <EOL> expected_set_domain_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> owner = owner , transaction = mock . ANY ) ) <EOL> expected_logic_node_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> mocked . APIC_EXT_SWITCH , mocked . APIC_EXT_MODULE , <EOL> mocked . APIC_EXT_PORT , mock . ANY , '<STR_LIT>' , <EOL> owner = owner , router_id = APIC_EXTERNAL_RID , <EOL> transaction = mock . ANY ) ) <EOL> expected_route_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ) <EOL> expected_route_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , subnet = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_set_domain_calls , <EOL> mgr . set_domain_for_external_routed_network . call_args_list ) <EOL> self . _check_call_list ( expected_logic_node_calls , <EOL> mgr . ensure_logical_node_profile_created . call_args_list ) <EOL> self . _check_call_list ( expected_route_calls , <EOL> mgr . ensure_static_route_created . call_args_list ) <EOL> else : <EOL> if is_edge_nat and self . nat_enabled : <EOL> final_req = re . sub ( '<STR_LIT>' , <EOL> l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , self . trimmed_l3out ) <EOL> mgr . apic . post_body . assert_called_once_with ( <EOL> mgr . apic . l3extOut . mo , final_req , l3p_owner , <EOL> l3out_str % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) ) <EOL> self . assertFalse ( mgr . set_domain_for_external_routed_network . called ) <EOL> self . assertFalse ( mgr . ensure_logical_node_profile_created . called ) <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> if self . nat_enabled : <EOL> mgr . set_l3out_for_bd . assert_called_once_with ( owner , <EOL> "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> def test_l3p_plugged_to_es_at_update_1 ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_update ( shared_es = True , <EOL> shared_l3p = False ) <EOL> def test_l3p_plugged_to_es_at_update_2 ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_update ( shared_es = True , <EOL> shared_l3p = True ) <EOL> def test_l3p_plugged_to_es_at_update_3 ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_update ( shared_es = False , <EOL> shared_l3p = False ) <EOL> def test_l3p_plugged_to_es_at_update_edge_nat_mode ( self ) : <EOL> self . _test_l3p_plugged_to_es_at_update ( shared_es = False , <EOL> shared_l3p = False , <EOL> is_edge_nat = True ) <EOL> def _test_l3p_unplugged_from_es_on_delete ( self , shared_es , <EOL> shared_l3p , is_edge_nat = False ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat ) <EOL> es1 = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] ) [ '<STR_LIT>' ] <EOL> es2 = self . create_external_segment ( <EOL> shared = shared_es , name = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( shared = shared_l3p , <EOL> tenant_id = es1 [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . set_context_for_external_routed_network . reset_mock ( ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , l3p [ '<STR_LIT:id>' ] , self . fmt ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> owner = self . common_tenant if shared_es else es1 [ '<STR_LIT>' ] <EOL> l3p_owner = self . common_tenant if shared_l3p else l3p [ '<STR_LIT>' ] <EOL> expected_delete_calls = [ ] <EOL> if not self . pre_l3out : <EOL> expected_delete_calls . append ( <EOL> mock . call ( es1 [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) ) <EOL> if self . nat_enabled : <EOL> l3out_str = "<STR_LIT>" <EOL> if is_edge_nat : <EOL> l3out_str = "<STR_LIT>" <EOL> expected_delete_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es1 [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , transaction = mock . ANY ) ) <EOL> self . _check_call_list ( <EOL> expected_delete_calls , <EOL> mgr . delete_external_routed_network . call_args_list ) <EOL> if self . nat_enabled : <EOL> mgr . unset_l3out_for_bd . assert_called_once_with ( owner , <EOL> "<STR_LIT>" % es1 [ '<STR_LIT:id>' ] , <EOL> es1 [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> if self . pre_l3out and not self . nat_enabled : <EOL> call_name = mgr . set_context_for_external_routed_network <EOL> call_name . assert_called_once_with ( APIC_PRE_L3OUT_TENANT , <EOL> es1 [ '<STR_LIT:name>' ] , None , transaction = mock . ANY ) <EOL> if is_edge_nat and self . nat_enabled : <EOL> self . driver . l3out_vlan_alloc . release_vlan . assert_called_once_with ( <EOL> es1 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) <EOL> mgr . delete_external_routed_network . reset_mock ( ) <EOL> mgr . unset_l3out_for_bd . reset_mock ( ) <EOL> self . driver . l3out_vlan_alloc . release_vlan . reset_mock ( ) <EOL> l3p = self . create_l3_policy ( <EOL> shared = shared_l3p , <EOL> tenant_id = es1 [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] , <EOL> es2 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> mgr . set_context_for_external_routed_network . reset_mock ( ) <EOL> req = self . new_delete_request ( '<STR_LIT>' , l3p [ '<STR_LIT:id>' ] , self . fmt ) <EOL> res = req . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> expected_delete_calls = [ ] <EOL> if not self . pre_l3out : <EOL> expected_delete_calls . extend ( [ <EOL> mock . call ( es1 [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) , <EOL> mock . call ( es2 [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) ] ) <EOL> if self . nat_enabled : <EOL> l3out_str = "<STR_LIT>" <EOL> if is_edge_nat : <EOL> l3out_str = "<STR_LIT>" <EOL> expected_delete_calls . extend ( [ <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es1 [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , transaction = mock . ANY ) , <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es2 [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , transaction = mock . ANY ) ] ) <EOL> self . _check_call_list ( <EOL> expected_delete_calls , <EOL> mgr . delete_external_routed_network . call_args_list ) <EOL> if self . nat_enabled : <EOL> expected_unset_calls = [ <EOL> mock . call ( owner , "<STR_LIT>" % es1 [ '<STR_LIT:id>' ] , <EOL> es1 [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( owner , "<STR_LIT>" % es2 [ '<STR_LIT:id>' ] , <EOL> es2 [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_unset_calls , mgr . unset_l3out_for_bd . call_args_list ) <EOL> if self . pre_l3out and not self . nat_enabled : <EOL> expected_calls = [ <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , <EOL> es1 [ '<STR_LIT:name>' ] , None , transaction = mock . ANY ) , <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , <EOL> es2 [ '<STR_LIT:name>' ] , None , transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , <EOL> mgr . set_context_for_external_routed_network . call_args_list ) <EOL> if is_edge_nat and self . nat_enabled : <EOL> expected_release_vlan_calls = [ mock . call ( es1 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) , <EOL> mock . call ( es2 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) ] <EOL> self . _check_call_list ( <EOL> expected_release_vlan_calls , <EOL> self . driver . l3out_vlan_alloc . release_vlan . call_args_list ) <EOL> def test_l3p_unplugged_from_es_on_delete_1 ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_delete ( shared_es = True , <EOL> shared_l3p = False ) <EOL> def test_l3p_unplugged_from_es_on_delete_2 ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_delete ( shared_es = True , <EOL> shared_l3p = True ) <EOL> def test_l3p_unplugged_from_es_on_delete_3 ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_delete ( shared_es = False , <EOL> shared_l3p = False ) <EOL> def test_l3p_unplugged_from_es_on_delete_edge_nat_mode ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_delete ( shared_es = False , <EOL> shared_l3p = False , <EOL> is_edge_nat = True ) <EOL> def _test_l3p_unplugged_from_es_on_update ( self , shared_es , <EOL> shared_l3p , is_edge_nat = False ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat ) <EOL> es1 = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] ) [ '<STR_LIT>' ] <EOL> es2 = self . create_external_segment ( <EOL> shared = shared_es , <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( <EOL> name = '<STR_LIT>' , <EOL> tenant_id = es1 [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> shared = shared_l3p , <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> owner = self . common_tenant if shared_es else es1 [ '<STR_LIT>' ] <EOL> l3p_owner = self . common_tenant if shared_l3p else l3p [ '<STR_LIT>' ] <EOL> mgr . ensure_external_routed_network_created . reset_mock ( ) <EOL> mgr . set_domain_for_external_routed_network . reset_mock ( ) <EOL> mgr . ensure_logical_node_profile_created . reset_mock ( ) <EOL> mgr . ensure_static_route_created . reset_mock ( ) <EOL> self . driver . l3out_vlan_alloc . reserve_vlan . reset_mock ( ) <EOL> mgr . apic . post_body . reset_mock ( ) <EOL> mgr . set_context_for_external_routed_network . reset_mock ( ) <EOL> l3p = self . update_l3_policy ( <EOL> l3p [ '<STR_LIT:id>' ] , tenant_id = l3p [ '<STR_LIT>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_segments = { es2 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } ) [ '<STR_LIT>' ] <EOL> l3out_str = "<STR_LIT>" <EOL> if is_edge_nat : <EOL> l3out_str = "<STR_LIT>" <EOL> expected_delete_calls = [ ] <EOL> if not self . pre_l3out : <EOL> expected_delete_calls . append ( <EOL> mock . call ( es1 [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) ) <EOL> if self . nat_enabled : <EOL> expected_delete_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es1 [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , transaction = mock . ANY ) ) <EOL> self . _check_call_list ( <EOL> expected_delete_calls , <EOL> mgr . delete_external_routed_network . call_args_list ) <EOL> if self . pre_l3out and not self . nat_enabled : <EOL> expected_calls = [ <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , <EOL> es1 [ '<STR_LIT:name>' ] , None , transaction = mock . ANY ) , <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , <EOL> es2 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] , transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , <EOL> mgr . set_context_for_external_routed_network . call_args_list ) <EOL> if is_edge_nat and self . nat_enabled : <EOL> self . driver . l3out_vlan_alloc . release_vlan . assert_called_once_with ( <EOL> es1 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) <EOL> expected_l3out_calls = [ ] <EOL> if self . nat_enabled : <EOL> if not is_edge_nat or not self . pre_l3out : <EOL> expected_l3out_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es2 [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , context = l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ) <EOL> if not self . pre_l3out : <EOL> expected_l3out_calls . append ( <EOL> mock . call ( es2 [ '<STR_LIT:id>' ] , owner = owner , <EOL> context = "<STR_LIT>" % es2 [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ) <EOL> elif not self . pre_l3out : <EOL> expected_l3out_calls = [ <EOL> mock . call ( es2 [ '<STR_LIT:id>' ] , owner = owner , context = l3p [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( expected_l3out_calls , <EOL> mgr . ensure_external_routed_network_created . call_args_list ) <EOL> if is_edge_nat and self . nat_enabled : <EOL> ( self . driver . l3out_vlan_alloc . <EOL> reserve_vlan . assert_called_once_with ( <EOL> es2 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) ) <EOL> if not self . pre_l3out : <EOL> expected_set_domain_calls = [ <EOL> mock . call ( es2 [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) ] <EOL> expected_logic_node_calls = [ <EOL> mock . call ( es2 [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , <EOL> mocked . APIC_EXT_MODULE , mocked . APIC_EXT_PORT , <EOL> mocked . APIC_EXT_ENCAP , '<STR_LIT>' , <EOL> owner = owner , router_id = APIC_EXTERNAL_RID , <EOL> transaction = mock . ANY ) ] <EOL> if is_edge_nat and self . nat_enabled : <EOL> expected_set_domain_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es2 [ '<STR_LIT:id>' ] ) , <EOL> owner = owner , transaction = mock . ANY ) ) <EOL> expected_logic_node_calls . append ( <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es2 [ '<STR_LIT:id>' ] ) , <EOL> mocked . APIC_EXT_SWITCH , mocked . APIC_EXT_MODULE , <EOL> mocked . APIC_EXT_PORT , mock . ANY , '<STR_LIT>' , <EOL> owner = owner , router_id = APIC_EXTERNAL_RID , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_set_domain_calls , <EOL> mgr . set_domain_for_external_routed_network . call_args_list ) <EOL> self . _check_call_list ( expected_logic_node_calls , <EOL> mgr . ensure_logical_node_profile_created . call_args_list ) <EOL> else : <EOL> if is_edge_nat and self . nat_enabled : <EOL> final_req = re . sub ( '<STR_LIT>' , <EOL> l3out_str % ( l3p [ '<STR_LIT:id>' ] , es2 [ '<STR_LIT:id>' ] ) , <EOL> self . trimmed_l3out ) <EOL> mgr . apic . post_body . assert_called_once_with ( <EOL> mgr . apic . l3extOut . mo , final_req , l3p_owner , <EOL> l3out_str % ( l3p [ '<STR_LIT:id>' ] , es2 [ '<STR_LIT:id>' ] ) ) <EOL> self . assertFalse ( mgr . set_domain_for_external_routed_network . called ) <EOL> self . assertFalse ( mgr . ensure_logical_node_profile_created . called ) <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> if self . nat_enabled : <EOL> mgr . unset_l3out_for_bd . assert_called_once_with ( owner , <EOL> "<STR_LIT>" % es1 [ '<STR_LIT:id>' ] , <EOL> es1 [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> self . driver . l3out_vlan_alloc . release_vlan . reset_mock ( ) <EOL> mgr . delete_external_routed_network . reset_mock ( ) <EOL> mgr . unset_l3out_for_bd . reset_mock ( ) <EOL> self . update_l3_policy ( <EOL> l3p [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , tenant_id = l3p [ '<STR_LIT>' ] , <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] , <EOL> es2 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } ) <EOL> mgr . set_context_for_external_routed_network . reset_mock ( ) <EOL> self . update_l3_policy ( <EOL> l3p [ '<STR_LIT:id>' ] , tenant_id = l3p [ '<STR_LIT>' ] , <EOL> expected_res_status = <NUM_LIT:200> , external_segments = { } ) <EOL> expected_delete_calls = [ ] <EOL> if not self . pre_l3out : <EOL> expected_delete_calls . extend ( [ <EOL> mock . call ( es1 [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) , <EOL> mock . call ( es2 [ '<STR_LIT:id>' ] , owner = owner , transaction = mock . ANY ) ] ) <EOL> if self . nat_enabled : <EOL> expected_delete_calls . extend ( [ <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es1 [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , transaction = mock . ANY ) , <EOL> mock . call ( l3out_str % ( l3p [ '<STR_LIT:id>' ] , es2 [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , transaction = mock . ANY ) ] ) <EOL> self . _check_call_list ( <EOL> expected_delete_calls , <EOL> mgr . delete_external_routed_network . call_args_list ) <EOL> if self . nat_enabled : <EOL> expected_unset_calls = [ <EOL> mock . call ( owner , "<STR_LIT>" % es1 [ '<STR_LIT:id>' ] , <EOL> es1 [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( owner , "<STR_LIT>" % es2 [ '<STR_LIT:id>' ] , <EOL> es2 [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_unset_calls , mgr . unset_l3out_for_bd . call_args_list ) <EOL> if self . pre_l3out and not self . nat_enabled : <EOL> expected_calls = [ <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , <EOL> es1 [ '<STR_LIT:name>' ] , None , transaction = mock . ANY ) , <EOL> mock . call ( APIC_PRE_L3OUT_TENANT , <EOL> es2 [ '<STR_LIT:name>' ] , None , transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , <EOL> mgr . set_context_for_external_routed_network . call_args_list ) <EOL> if is_edge_nat and self . nat_enabled : <EOL> expected_release_vlan_calls = [ mock . call ( es1 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) , <EOL> mock . call ( es2 [ '<STR_LIT:name>' ] , l3p [ '<STR_LIT:id>' ] ) ] <EOL> self . _check_call_list ( <EOL> expected_release_vlan_calls , <EOL> self . driver . l3out_vlan_alloc . release_vlan . call_args_list ) <EOL> def test_l3p_unplugged_from_es_on_update_1 ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_update ( shared_es = True , <EOL> shared_l3p = False ) <EOL> def test_l3p_unplugged_from_es_on_update_2 ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_update ( shared_es = True , <EOL> shared_l3p = True ) <EOL> def test_l3p_unplugged_from_es_on_update_3 ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_update ( shared_es = False , <EOL> shared_l3p = False ) <EOL> def test_l3p_unplugged_from_es_on_update_edge_nat_mode ( self ) : <EOL> self . _test_l3p_unplugged_from_es_on_update ( shared_es = False , <EOL> shared_l3p = False , <EOL> is_edge_nat = True ) <EOL> def test_verify_unsupported_es_noop ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . create_l3_policy ( <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] } , <EOL> expected_res_status = <NUM_LIT> ) <EOL> mgr = self . driver . apic_manager <EOL> self . assertFalse ( mgr . ensure_external_routed_network_created . called ) <EOL> self . assertFalse ( mgr . set_domain_for_external_routed_network . called ) <EOL> self . assertFalse ( mgr . ensure_logical_node_profile_created . called ) <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> def test_l3p_external_address ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es1 = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> es2 = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( [ '<STR_LIT>' ] , l3p [ '<STR_LIT>' ] [ es1 [ '<STR_LIT:id>' ] ] ) <EOL> l3p = self . update_l3_policy ( <EOL> l3p [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ ] , es2 [ '<STR_LIT:id>' ] : [ ] } ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( [ '<STR_LIT>' ] , l3p [ '<STR_LIT>' ] [ es1 [ '<STR_LIT:id>' ] ] ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] , l3p [ '<STR_LIT>' ] [ es2 [ '<STR_LIT:id>' ] ] ) <EOL> l3p = self . update_l3_policy ( <EOL> l3p [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ '<STR_LIT>' ] , <EOL> es2 [ '<STR_LIT:id>' ] : [ ] } ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( [ '<STR_LIT>' ] , l3p [ '<STR_LIT>' ] [ es1 [ '<STR_LIT:id>' ] ] ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] , l3p [ '<STR_LIT>' ] [ es2 [ '<STR_LIT:id>' ] ] ) <EOL> def _test_multi_es_with_ptg ( self , shared_es ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es1 = self . create_external_segment ( shared = shared_es , <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> es2 = self . create_external_segment ( shared = shared_es , <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( <EOL> external_segments = { es1 [ '<STR_LIT:id>' ] : [ ] , es2 [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l2p = self . create_l2_policy ( l3_policy_id = l3p [ '<STR_LIT:id>' ] ) [ '<STR_LIT>' ] <EOL> ptg = self . create_policy_target_group ( name = "<STR_LIT>" , <EOL> l2_policy_id = l2p [ '<STR_LIT:id>' ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> res = self . new_delete_request ( '<STR_LIT>' , ptg [ '<STR_LIT:id>' ] , <EOL> self . fmt ) . get_response ( self . ext_api ) <EOL> self . assertEqual ( webob . exc . HTTPNoContent . code , res . status_int ) <EOL> def test_multi_es_with_ptg_1 ( self ) : <EOL> self . _test_multi_es_with_ptg ( False ) <EOL> def test_multi_es_with_ptg_2 ( self ) : <EOL> self . _test_multi_es_with_ptg ( True ) <EOL> class TestL3PolicyNoNat ( TestL3Policy ) : <EOL> def setUp ( self ) : <EOL> super ( TestL3PolicyNoNat , self ) . setUp ( nat_enabled = False ) <EOL> class TestL3PolicyPreL3Out ( TestL3Policy ) : <EOL> def setUp ( self ) : <EOL> super ( TestL3PolicyPreL3Out , self ) . setUp ( pre_existing_l3out = True ) <EOL> class TestL3PolicyNoNatPreL3Out ( TestL3Policy ) : <EOL> def setUp ( self ) : <EOL> super ( TestL3PolicyNoNatPreL3Out , self ) . setUp ( <EOL> nat_enabled = False , pre_existing_l3out = True ) <EOL> class TestPolicyRuleSet ( ApicMappingTestCase ) : <EOL> def _test_policy_rule_set_created_on_apic ( self , shared = False ) : <EOL> ct = self . create_policy_rule_set ( name = "<STR_LIT>" , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> tenant = self . common_tenant if shared else ct [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . create_contract . assert_called_once_with ( <EOL> ct [ '<STR_LIT:id>' ] , owner = tenant , transaction = mock . ANY ) <EOL> def test_policy_rule_set_created_on_apic ( self ) : <EOL> self . _test_policy_rule_set_created_on_apic ( ) <EOL> def test_policy_rule_set_created_on_apic_shared ( self ) : <EOL> self . _test_policy_rule_set_created_on_apic ( shared = True ) <EOL> def _test_policy_rule_set_created_with_rules ( self , shared = False ) : <EOL> bi , in_d , out = range ( <NUM_LIT:3> ) <EOL> rules = self . _create_3_direction_rules ( shared = shared ) <EOL> ctr = self . create_policy_rule_set ( <EOL> name = "<STR_LIT>" , policy_rules = [ x [ '<STR_LIT:id>' ] for x in rules [ <NUM_LIT:1> : ] ] ) [ <EOL> '<STR_LIT>' ] <EOL> rule_owner = self . common_tenant if shared else rules [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> expected_calls = [ <EOL> mock . call ( ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , rules [ in_d ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = mock . ANY , <EOL> unset = False , rule_owner = rule_owner ) , <EOL> mock . call ( ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , <EOL> amap . REVERSE_PREFIX + rules [ out ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = mock . ANY , <EOL> unset = False , rule_owner = rule_owner ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , <EOL> mgr . manage_contract_subject_in_filter . call_args_list ) <EOL> expected_calls = [ <EOL> mock . call ( ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , rules [ out ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = mock . ANY , <EOL> unset = False , rule_owner = rule_owner ) , <EOL> mock . call ( ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , <EOL> amap . REVERSE_PREFIX + rules [ in_d ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = mock . ANY , <EOL> unset = False , rule_owner = rule_owner ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , <EOL> mgr . manage_contract_subject_out_filter . call_args_list ) <EOL> ctr = self . create_policy_rule_set ( <EOL> name = "<STR_LIT>" , policy_rules = [ rules [ bi ] [ '<STR_LIT:id>' ] ] ) [ '<STR_LIT>' ] <EOL> mgr . manage_contract_subject_in_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , rules [ bi ] [ '<STR_LIT:id>' ] , owner = ctr [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY , unset = False , <EOL> rule_owner = rule_owner ) <EOL> mgr . manage_contract_subject_out_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , rules [ bi ] [ '<STR_LIT:id>' ] , owner = ctr [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY , unset = False , <EOL> rule_owner = rule_owner ) <EOL> mgr . manage_contract_subject_in_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , amap . REVERSE_PREFIX + rules [ bi ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = mock . ANY , unset = False , <EOL> rule_owner = rule_owner ) <EOL> mgr . manage_contract_subject_out_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , amap . REVERSE_PREFIX + rules [ bi ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = mock . ANY , unset = False , <EOL> rule_owner = rule_owner ) <EOL> def test_policy_rule_set_created_with_rules ( self ) : <EOL> self . _test_policy_rule_set_created_with_rules ( ) <EOL> def test_policy_rule_set_created_with_rules_shared ( self ) : <EOL> self . _test_policy_rule_set_created_with_rules ( shared = True ) <EOL> def _test_policy_rule_set_updated_with_new_rules ( self , shared = False ) : <EOL> bi , in_d , out = range ( <NUM_LIT:3> ) <EOL> old_rules = self . _create_3_direction_rules ( shared = shared ) <EOL> new_rules = self . _create_3_direction_rules ( shared = shared ) <EOL> ctr = self . create_policy_rule_set ( <EOL> name = "<STR_LIT>" , <EOL> policy_rules = [ x [ '<STR_LIT:id>' ] for x in old_rules [ <NUM_LIT:1> : ] ] ) [ '<STR_LIT>' ] <EOL> data = { '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ x [ '<STR_LIT:id>' ] for x in new_rules [ <NUM_LIT:1> : ] ] } } <EOL> rule_owner = ( self . common_tenant if shared else <EOL> old_rules [ in_d ] [ '<STR_LIT>' ] ) <EOL> mgr = self . driver . apic_manager <EOL> mgr . manage_contract_subject_in_filter = MockCallRecorder ( ) <EOL> mgr . manage_contract_subject_out_filter = MockCallRecorder ( ) <EOL> self . new_update_request ( <EOL> '<STR_LIT>' , data , ctr [ '<STR_LIT:id>' ] , self . fmt ) . get_response ( <EOL> self . ext_api ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_in_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , old_rules [ in_d ] [ '<STR_LIT:id>' ] , <EOL> rule_owner = rule_owner , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , unset = True ) ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_in_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , new_rules [ in_d ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , <EOL> unset = False , rule_owner = rule_owner ) ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_out_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , old_rules [ out ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , unset = True , <EOL> rule_owner = rule_owner ) ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_out_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , new_rules [ out ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , <EOL> unset = False , rule_owner = rule_owner ) ) <EOL> ctr = self . create_policy_rule_set ( <EOL> name = "<STR_LIT>" , <EOL> policy_rules = [ old_rules [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ] ) [ '<STR_LIT>' ] <EOL> data = { '<STR_LIT>' : { '<STR_LIT>' : [ new_rules [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ] } } <EOL> self . new_update_request ( <EOL> '<STR_LIT>' , data , ctr [ '<STR_LIT:id>' ] , self . fmt ) . get_response ( <EOL> self . ext_api ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_in_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , old_rules [ bi ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , unset = True , <EOL> rule_owner = rule_owner ) ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_out_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , old_rules [ bi ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , unset = True , <EOL> rule_owner = rule_owner ) ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_in_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , new_rules [ bi ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , <EOL> unset = False , rule_owner = rule_owner ) ) <EOL> self . assertTrue ( <EOL> mgr . manage_contract_subject_out_filter . call_happened_with ( <EOL> ctr [ '<STR_LIT:id>' ] , ctr [ '<STR_LIT:id>' ] , new_rules [ bi ] [ '<STR_LIT:id>' ] , <EOL> owner = ctr [ '<STR_LIT>' ] , transaction = '<STR_LIT>' , <EOL> unset = False , rule_owner = rule_owner ) ) <EOL> def test_policy_rule_set_updated_with_new_rules ( self ) : <EOL> self . _test_policy_rule_set_updated_with_new_rules ( ) <EOL> def test_policy_rule_set_updated_with_new_rules_shared ( self ) : <EOL> self . _test_policy_rule_set_updated_with_new_rules ( shared = True ) <EOL> def _create_3_direction_rules ( self , shared = False ) : <EOL> a1 = self . create_policy_action ( name = '<STR_LIT>' , <EOL> action_type = '<STR_LIT>' , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> cl_attr = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } <EOL> cls = [ ] <EOL> for direction in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> if direction == '<STR_LIT>' : <EOL> cl_attr [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> cls . append ( self . create_policy_classifier ( <EOL> direction = direction , shared = shared , <EOL> ** cl_attr ) [ '<STR_LIT>' ] ) <EOL> rules = [ ] <EOL> for classifier in cls : <EOL> rules . append ( self . create_policy_rule ( <EOL> policy_classifier_id = classifier [ '<STR_LIT:id>' ] , <EOL> policy_actions = [ a1 [ '<STR_LIT:id>' ] ] , <EOL> shared = shared ) [ '<STR_LIT>' ] ) <EOL> return rules <EOL> class TestPolicyRule ( ApicMappingTestCase ) : <EOL> def _test_policy_rule_created_on_apic ( self , shared = False ) : <EOL> pr = self . _create_simple_policy_rule ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> , shared = shared ) <EOL> pr1 = self . _create_simple_policy_rule ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> , shared = shared ) <EOL> pr2 = self . _create_simple_policy_rule ( '<STR_LIT>' , None , <NUM_LIT> , shared = shared ) <EOL> tenant = self . common_tenant if shared else pr [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> expected_calls = [ <EOL> mock . call ( pr [ '<STR_LIT:id>' ] , owner = tenant , entry = '<STR_LIT>' , etherT = '<STR_LIT>' , <EOL> prot = '<STR_LIT>' , dToPort = <NUM_LIT> , dFromPort = <NUM_LIT> , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr [ '<STR_LIT:id>' ] , owner = tenant , <EOL> entry = '<STR_LIT>' , etherT = '<STR_LIT>' , prot = '<STR_LIT>' , sToPort = <NUM_LIT> , <EOL> sFromPort = <NUM_LIT> , tcpRules = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = tenant , entry = '<STR_LIT>' , <EOL> etherT = '<STR_LIT>' , prot = '<STR_LIT>' , dToPort = <NUM_LIT> , dFromPort = <NUM_LIT> , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , owner = tenant , <EOL> entry = '<STR_LIT>' , etherT = '<STR_LIT>' , prot = '<STR_LIT>' , sToPort = <NUM_LIT> , <EOL> sFromPort = <NUM_LIT> , transaction = mock . ANY ) , <EOL> mock . call ( pr2 [ '<STR_LIT:id>' ] , owner = tenant , entry = '<STR_LIT>' , <EOL> etherT = '<STR_LIT>' , dToPort = <NUM_LIT> , dFromPort = <NUM_LIT> , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . create_tenant_filter . call_args_list ) <EOL> mgr . reset_mock ( ) <EOL> pr = self . _create_simple_policy_rule ( '<STR_LIT>' , None , None , shared = shared ) <EOL> expected_calls = [ <EOL> mock . call ( pr [ '<STR_LIT:id>' ] , owner = tenant , entry = '<STR_LIT>' , <EOL> etherT = '<STR_LIT>' , transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . create_tenant_filter . call_args_list ) <EOL> def test_policy_rule_created_on_apic ( self ) : <EOL> self . _test_policy_rule_created_on_apic ( ) <EOL> def test_policy_rule_created_on_apic_shared ( self ) : <EOL> self . _test_policy_rule_created_on_apic ( shared = True ) <EOL> def _test_policy_rule_deleted_on_apic ( self , shared = False ) : <EOL> pr = self . _create_simple_policy_rule ( shared = shared ) <EOL> pr1 = self . _create_simple_policy_rule ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> , shared = shared ) <EOL> self . delete_policy_rule ( pr [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT> ) <EOL> tenant = self . common_tenant if shared else pr [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> expected_calls = [ <EOL> mock . call ( pr [ '<STR_LIT:id>' ] , owner = tenant , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr [ '<STR_LIT:id>' ] , owner = tenant , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . delete_tenant_filter . call_args_list ) <EOL> mgr . delete_tenant_filter . reset_mock ( ) <EOL> self . delete_policy_rule ( pr1 [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT> ) <EOL> expected_calls = [ <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = tenant , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , owner = tenant , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . delete_tenant_filter . call_args_list ) <EOL> def test_policy_rule_deleted_on_apic ( self ) : <EOL> self . _test_policy_rule_deleted_on_apic ( ) <EOL> def test_policy_rule_deleted_on_apic_shared ( self ) : <EOL> self . _test_policy_rule_deleted_on_apic ( shared = True ) <EOL> def test_policy_classifier_updated ( self ) : <EOL> pa = self . create_policy_action ( <EOL> action_type = '<STR_LIT>' , is_admin_context = True , <EOL> tenant_id = '<STR_LIT>' , shared = True ) [ '<STR_LIT>' ] <EOL> pc = self . create_policy_classifier ( <EOL> direction = '<STR_LIT>' , protocol = '<STR_LIT>' , port_range = <NUM_LIT> , <EOL> shared = True , is_admin_context = True , <EOL> tenant_id = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> pr1 = self . create_policy_rule ( <EOL> policy_classifier_id = pc [ '<STR_LIT:id>' ] , policy_actions = [ pa [ '<STR_LIT:id>' ] ] , <EOL> shared = True , is_admin_context = True , <EOL> tenant_id = '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> pr2 = self . create_policy_rule ( policy_classifier_id = pc [ '<STR_LIT:id>' ] , <EOL> policy_actions = [ pa [ '<STR_LIT:id>' ] ] ) [ '<STR_LIT>' ] <EOL> prs1 = self . create_policy_rule_set ( <EOL> policy_rules = [ pr1 [ '<STR_LIT:id>' ] ] ) [ '<STR_LIT>' ] <EOL> prs2 = self . create_policy_rule_set ( <EOL> policy_rules = [ pr2 [ '<STR_LIT:id>' ] , pr1 [ '<STR_LIT:id>' ] ] ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . reset_mock ( ) <EOL> self . update_policy_classifier ( pc [ '<STR_LIT:id>' ] , port_range = None , <EOL> is_admin_context = True ) <EOL> expected_calls = [ <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , etherT = '<STR_LIT>' , prot = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , etherT = '<STR_LIT>' , prot = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> etherT = '<STR_LIT>' , prot = '<STR_LIT>' , entry = '<STR_LIT>' , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> etherT = '<STR_LIT>' , prot = '<STR_LIT>' , entry = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . create_tenant_filter . call_args_list ) <EOL> expected_calls = [ <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . delete_tenant_filter . call_args_list ) <EOL> self . assertFalse ( mgr . manage_contract_subject_in_filter . called ) <EOL> self . assertFalse ( mgr . manage_contract_subject_out_filter . called ) <EOL> mgr . reset_mock ( ) <EOL> self . update_policy_classifier ( pc [ '<STR_LIT:id>' ] , protocol = None , <EOL> is_admin_context = True ) <EOL> expected_calls = [ <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , etherT = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , etherT = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . create_tenant_filter . call_args_list ) <EOL> expected_calls = [ <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . delete_tenant_filter . call_args_list ) <EOL> self . assertTrue ( mgr . manage_contract_subject_in_filter . called ) <EOL> self . assertTrue ( mgr . manage_contract_subject_out_filter . called ) <EOL> mgr . reset_mock ( ) <EOL> self . update_policy_classifier ( pc [ '<STR_LIT:id>' ] , protocol = '<STR_LIT>' , <EOL> is_admin_context = True ) <EOL> expected_calls = [ <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . delete_tenant_filter . call_args_list ) <EOL> expected_calls = [ <EOL> mock . call ( pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , etherT = '<STR_LIT>' , prot = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , etherT = '<STR_LIT>' , prot = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> etherT = '<STR_LIT>' , prot = '<STR_LIT>' , tcpRules = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> etherT = '<STR_LIT>' , prot = '<STR_LIT>' , tcpRules = '<STR_LIT>' , <EOL> entry = '<STR_LIT>' , transaction = mock . ANY ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . create_tenant_filter . call_args_list ) <EOL> expected_calls = [ <EOL> mock . call ( prs1 [ '<STR_LIT:id>' ] , prs1 [ '<STR_LIT:id>' ] , pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY , unset = True , rule_owner = '<STR_LIT>' ) , <EOL> mock . call ( prs2 [ '<STR_LIT:id>' ] , prs2 [ '<STR_LIT:id>' ] , pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY , unset = True , rule_owner = '<STR_LIT>' ) , <EOL> mock . call ( prs2 [ '<STR_LIT:id>' ] , prs2 [ '<STR_LIT:id>' ] , pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY , unset = True , <EOL> rule_owner = '<STR_LIT>' ) , <EOL> mock . call ( prs1 [ '<STR_LIT:id>' ] , prs1 [ '<STR_LIT:id>' ] , pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY , unset = False , rule_owner = '<STR_LIT>' ) , <EOL> mock . call ( prs2 [ '<STR_LIT:id>' ] , prs2 [ '<STR_LIT:id>' ] , pr1 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY , unset = False , rule_owner = '<STR_LIT>' ) , <EOL> mock . call ( prs2 [ '<STR_LIT:id>' ] , prs2 [ '<STR_LIT:id>' ] , pr2 [ '<STR_LIT:id>' ] , owner = '<STR_LIT>' , <EOL> transaction = mock . ANY , unset = False , <EOL> rule_owner = '<STR_LIT>' ) <EOL> ] <EOL> self . _check_call_list ( <EOL> expected_calls , <EOL> mgr . manage_contract_subject_in_filter . call_args_list ) <EOL> expected_calls = [ <EOL> mock . call ( prs1 [ '<STR_LIT:id>' ] , prs1 [ '<STR_LIT:id>' ] , amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , <EOL> owner = '<STR_LIT>' , transaction = mock . ANY , unset = False , <EOL> rule_owner = '<STR_LIT>' ) , <EOL> mock . call ( prs2 [ '<STR_LIT:id>' ] , prs2 [ '<STR_LIT:id>' ] , amap . REVERSE_PREFIX + pr1 [ '<STR_LIT:id>' ] , <EOL> owner = '<STR_LIT>' , transaction = mock . ANY , unset = False , <EOL> rule_owner = '<STR_LIT>' ) , <EOL> mock . call ( prs2 [ '<STR_LIT:id>' ] , prs2 [ '<STR_LIT:id>' ] , amap . REVERSE_PREFIX + pr2 [ '<STR_LIT:id>' ] , <EOL> owner = '<STR_LIT>' , transaction = mock . ANY , unset = False , <EOL> rule_owner = '<STR_LIT>' ) <EOL> ] <EOL> self . _check_call_list ( <EOL> expected_calls , <EOL> mgr . manage_contract_subject_out_filter . call_args_list ) <EOL> def test_icmp_rule_created_on_apic ( self ) : <EOL> pr = self . _create_simple_policy_rule ( '<STR_LIT>' , '<STR_LIT>' , None ) <EOL> tenant = pr [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> expected_calls = [ <EOL> mock . call ( pr [ '<STR_LIT:id>' ] , owner = tenant , entry = '<STR_LIT>' , etherT = '<STR_LIT>' , <EOL> prot = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr [ '<STR_LIT:id>' ] , owner = tenant , <EOL> entry = mock . ANY , etherT = '<STR_LIT>' , icmpv4T = '<STR_LIT>' , <EOL> prot = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr [ '<STR_LIT:id>' ] , owner = tenant , <EOL> entry = mock . ANY , etherT = '<STR_LIT>' , icmpv4T = '<STR_LIT>' , <EOL> prot = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr [ '<STR_LIT:id>' ] , owner = tenant , <EOL> entry = mock . ANY , etherT = '<STR_LIT>' , icmpv4T = '<STR_LIT>' , <EOL> prot = '<STR_LIT>' , transaction = mock . ANY ) , <EOL> mock . call ( amap . REVERSE_PREFIX + pr [ '<STR_LIT:id>' ] , owner = tenant , <EOL> entry = mock . ANY , etherT = '<STR_LIT>' , icmpv4T = '<STR_LIT>' , <EOL> prot = '<STR_LIT>' , transaction = mock . ANY ) ] <EOL> found = set ( ) <EOL> for call in mgr . create_tenant_filter . call_args_list : <EOL> if call [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . startswith ( amap . REVERSE_PREFIX ) : <EOL> self . assertFalse ( call [ <NUM_LIT:1> ] [ '<STR_LIT>' ] in found ) <EOL> found . add ( call [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . create_tenant_filter . call_args_list ) <EOL> class TestExternalSegment ( ApicMappingTestCase ) : <EOL> def test_pat_rejected ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> res = self . create_external_segment ( <EOL> name = '<STR_LIT>' , port_address_translation = True , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , expected_res_status = <NUM_LIT> , <EOL> port_address_translation = False ) [ '<STR_LIT>' ] <EOL> res = self . update_external_segment ( <EOL> es [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT> , port_address_translation = True ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_edge_nat_invalid_vlan_range_rejected ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat = True ) <EOL> self . driver . l3out_vlan_alloc . l3out_vlan_ranges = { } <EOL> res = self . create_external_segment ( <EOL> name = '<STR_LIT>' , expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> ext_info = self . driver . apic_manager . ext_net_dict . get ( '<STR_LIT>' ) <EOL> del ext_info [ '<STR_LIT>' ] <EOL> res = self . create_external_segment ( <EOL> name = '<STR_LIT>' , expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def _test_create_delete ( self , shared = False ) : <EOL> mgr = self . driver . apic_manager <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> mgr . ext_net_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> es = self . create_external_segment ( name = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> , shared = shared ) [ '<STR_LIT>' ] <EOL> self . create_external_segment ( name = '<STR_LIT>' , expected_res_status = <NUM_LIT> , <EOL> shared = shared ) <EOL> self . assertEqual ( '<STR_LIT>' , es [ '<STR_LIT>' ] ) <EOL> self . assertIsNotNone ( es [ '<STR_LIT>' ] ) <EOL> subnet = self . _get_object ( '<STR_LIT>' , es [ '<STR_LIT>' ] , <EOL> self . api ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( '<STR_LIT>' , subnet [ '<STR_LIT>' ] ) <EOL> owner = es [ '<STR_LIT>' ] if not shared else self . common_tenant <EOL> prs = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> if self . nat_enabled : <EOL> ctx = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> ctx_owner = owner <EOL> contract_owner = owner <EOL> if self . pre_l3out : <EOL> ctx = APIC_PRE_VRF <EOL> ctx_owner = APIC_PRE_VRF_TENANT <EOL> contract_owner = APIC_PRE_L3OUT_TENANT <EOL> self . assertFalse ( mgr . ensure_context_enforced . called ) <EOL> else : <EOL> mgr . ensure_context_enforced . assert_called_with ( <EOL> owner = owner , ctx_id = ctx , <EOL> transaction = mock . ANY ) <EOL> mgr . ensure_bd_created_on_apic ( <EOL> owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , ctx_owner = ctx_owner , <EOL> ctx_name = ctx , transaction = mock . ANY ) <EOL> mgr . ensure_epg_created . assert_called_with ( <EOL> owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , bd_name = "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> transaction = mock . ANY ) <EOL> mgr . create_tenant_filter . assert_called_with ( <EOL> prs , owner = contract_owner , <EOL> entry = "<STR_LIT>" , transaction = mock . ANY ) <EOL> mgr . manage_contract_subject_bi_filter . assert_called_with ( <EOL> prs , prs , prs , owner = contract_owner , transaction = mock . ANY ) <EOL> expected_calls = [ <EOL> mock . call ( owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , prs , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , prs , <EOL> provider = True , transaction = mock . ANY ) ] <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . set_contract_for_epg . call_args_list ) <EOL> ctx = context . get_admin_context ( ) <EOL> internal_subnets = self . _db_plugin . get_subnets ( <EOL> ctx , filters = { '<STR_LIT:name>' : [ amap . HOST_SNAT_POOL ] } ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( internal_subnets ) ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_bd_created_on_apic . called ) <EOL> self . assertFalse ( mgr . ensure_epg_created . called ) <EOL> self . assertFalse ( mgr . create_tenant_filter . called ) <EOL> self . assertFalse ( mgr . manage_contract_subject_bi_filter . called ) <EOL> self . assertFalse ( mgr . set_contract_for_epg . called ) <EOL> subnet_id = es [ '<STR_LIT>' ] <EOL> self . delete_external_segment ( es [ '<STR_LIT:id>' ] , <EOL> expected_res_status = webob . exc . HTTPNoContent . code ) <EOL> self . _get_object ( '<STR_LIT>' , subnet_id , self . api , <EOL> expected_res_status = <NUM_LIT> ) <EOL> if self . nat_enabled : <EOL> ctx = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> ctx_owner = owner <EOL> contract_owner = owner <EOL> if self . pre_l3out : <EOL> ctx = APIC_PRE_VRF <EOL> ctx_owner = APIC_PRE_VRF_TENANT <EOL> contract_owner = APIC_PRE_L3OUT_TENANT <EOL> self . assertFalse ( mgr . ensure_context_enforced . called ) <EOL> else : <EOL> mgr . ensure_context_deleted . assert_called_with ( <EOL> ctx_owner , ctx , transaction = mock . ANY ) <EOL> mgr . delete_bd_on_apic . assert_called_with ( <EOL> owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> mgr . delete_epg_for_network . assert_called_with ( <EOL> owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , transaction = mock . ANY ) <EOL> mgr . delete_contract . assert_called_with ( <EOL> prs , owner = contract_owner , transaction = mock . ANY ) <EOL> mgr . delete_tenant_filter . assert_called_with ( <EOL> prs , owner = contract_owner , transaction = mock . ANY ) <EOL> else : <EOL> self . assertFalse ( mgr . delete_bd_on_apic . called ) <EOL> self . assertFalse ( mgr . delete_epg_for_network . called ) <EOL> self . assertFalse ( mgr . delete_contract . called ) <EOL> self . assertFalse ( mgr . delete_tenant_filter . called ) <EOL> def test_create_delete_unshared ( self ) : <EOL> self . _test_create_delete ( False ) <EOL> def test_create_delete_shared ( self ) : <EOL> self . _test_create_delete ( True ) <EOL> def test_update_unsupported_noop ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> self . update_external_segment ( es [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_routes = [ ] ) <EOL> mgr = self . driver . apic_manager <EOL> self . assertFalse ( mgr . ensure_static_route_deleted . called ) <EOL> self . assertFalse ( mgr . ensure_external_epg_routes_deleted . called ) <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> self . assertFalse ( mgr . ensure_external_epg_created . called ) <EOL> self . assertFalse ( mgr . ensure_next_hop_deleted . called ) <EOL> def _test_route_update_remove ( self , shared_es ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> if self . pre_l3out and not self . nat_enabled : <EOL> tenants = [ es [ '<STR_LIT>' ] ] <EOL> else : <EOL> tenants = ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if self . nat_enabled and shared_es <EOL> else [ es [ '<STR_LIT>' ] ] ) <EOL> l3p_list = [ ] <EOL> for x in xrange ( len ( tenants ) ) : <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = tenants [ x ] , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l3p_list . append ( l3p ) <EOL> f = self . create_external_policy <EOL> eps = [ f ( external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> tenant_id = tenants [ x ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> for x in xrange ( len ( tenants ) ) ] <EOL> mgr = self . driver . apic_manager <EOL> owner = es [ '<STR_LIT>' ] if not shared_es else self . common_tenant <EOL> mgr . ensure_external_epg_created . reset_mock ( ) <EOL> mgr . ensure_static_route_created . reset_mock ( ) <EOL> self . update_external_segment ( es [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_routes = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) <EOL> mgr = self . driver . apic_manager <EOL> if not self . pre_l3out : <EOL> mgr . ensure_static_route_deleted . assert_called_with ( <EOL> es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> owner = owner , transaction = mock . ANY ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_static_route_deleted . called ) <EOL> expected_delete_calls = [ ] <EOL> for x in range ( len ( tenants ) ) : <EOL> ep = eps [ x ] <EOL> l3p = l3p_list [ x ] <EOL> l3out = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> ext_epg = ep [ '<STR_LIT:id>' ] <EOL> tenant = APIC_PRE_L3OUT_TENANT if self . pre_l3out else owner <EOL> if self . nat_enabled : <EOL> l3out = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) <EOL> ext_epg = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ext_epg ) <EOL> tenant = tenants [ x ] <EOL> expected_delete_calls . append ( <EOL> mock . call ( l3out , subnets = [ '<STR_LIT>' ] , <EOL> external_epg = ext_epg , owner = tenant , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( <EOL> expected_delete_calls , <EOL> mgr . ensure_external_epg_routes_deleted . call_args_list ) <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> self . assertFalse ( mgr . ensure_external_epg_created . called ) <EOL> self . assertFalse ( mgr . ensure_next_hop_deleted . called ) <EOL> mgr . ensure_static_route_deleted . reset_mock ( ) <EOL> mgr . ensure_external_epg_routes_deleted . reset_mock ( ) <EOL> self . update_external_segment ( es [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_routes = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] ) <EOL> if not self . pre_l3out : <EOL> mgr . ensure_next_hop_deleted . assert_called_with ( <EOL> es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , '<STR_LIT>' , <EOL> owner = owner , transaction = mock . ANY ) <EOL> mgr . ensure_static_route_created . assert_called_with ( <EOL> es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> subnet = '<STR_LIT>' , owner = owner , transaction = mock . ANY ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> self . assertFalse ( mgr . ensure_next_hop_deleted . called ) <EOL> expected_delete_calls = [ ] <EOL> for x in range ( len ( tenants ) ) : <EOL> ep = eps [ x ] <EOL> l3p = l3p_list [ x ] <EOL> l3out = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> ext_epg = ep [ '<STR_LIT:id>' ] <EOL> tenant = APIC_PRE_L3OUT_TENANT if self . pre_l3out else owner <EOL> if self . nat_enabled : <EOL> l3out = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) <EOL> ext_epg = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ext_epg ) <EOL> tenant = tenants [ x ] <EOL> expected_delete_calls . append ( <EOL> mock . call ( l3out , subnet = '<STR_LIT>' , <EOL> external_epg = ext_epg , owner = tenant , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_delete_calls , <EOL> mgr . ensure_external_epg_created . call_args_list ) <EOL> self . assertFalse ( mgr . ensure_static_route_deleted . called ) <EOL> self . assertFalse ( mgr . ensure_external_epg_routes_deleted . called ) <EOL> def test_route_update_remove_1 ( self ) : <EOL> self . _test_route_update_remove ( shared_es = True ) <EOL> def test_route_update_remove_2 ( self ) : <EOL> self . _test_route_update_remove ( shared_es = False ) <EOL> def _test_route_update_add ( self , shared_es ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> external_routes = [ ] , expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> if self . pre_l3out and not self . nat_enabled : <EOL> tenants = [ es [ '<STR_LIT>' ] ] <EOL> else : <EOL> tenants = ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if self . nat_enabled and shared_es <EOL> else [ es [ '<STR_LIT>' ] ] ) <EOL> l3p_list = [ ] <EOL> for x in xrange ( len ( tenants ) ) : <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = tenants [ x ] , <EOL> external_segments = { es [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l3p_list . append ( l3p ) <EOL> f = self . create_external_policy <EOL> eps = [ f ( external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> tenant_id = tenants [ x ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> for x in xrange ( len ( tenants ) ) ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . ensure_external_epg_created . reset_mock ( ) <EOL> owner = es [ '<STR_LIT>' ] if not shared_es else self . common_tenant <EOL> self . update_external_segment ( es [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_routes = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) <EOL> if not self . pre_l3out : <EOL> mgr . ensure_static_route_created . assert_called_with ( <EOL> es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> subnet = '<STR_LIT>' , owner = owner , transaction = mock . ANY ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> expected_create_calls = [ ] <EOL> for x in range ( len ( tenants ) ) : <EOL> ep = eps [ x ] <EOL> l3p = l3p_list [ x ] <EOL> l3out = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> ext_epg = ep [ '<STR_LIT:id>' ] <EOL> tenant = APIC_PRE_L3OUT_TENANT if self . pre_l3out else owner <EOL> if self . nat_enabled : <EOL> l3out = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) <EOL> ext_epg = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ext_epg ) <EOL> tenant = tenants [ x ] <EOL> expected_create_calls . append ( <EOL> mock . call ( l3out , subnet = '<STR_LIT>' , <EOL> external_epg = ext_epg , owner = tenant , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_create_calls , <EOL> mgr . ensure_external_epg_created . call_args_list ) <EOL> self . assertFalse ( mgr . ensure_static_route_deleted . called ) <EOL> self . assertFalse ( mgr . ensure_external_epg_routes_deleted . called ) <EOL> self . assertFalse ( mgr . ensure_next_hop_deleted . called ) <EOL> mgr . ensure_static_route_created . reset_mock ( ) <EOL> mgr . ensure_external_epg_created . reset_mock ( ) <EOL> self . update_external_segment ( es [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , <EOL> external_routes = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] ) <EOL> if not self . pre_l3out : <EOL> mgr . ensure_static_route_created . assert_called_with ( <EOL> es [ '<STR_LIT:id>' ] , mocked . APIC_EXT_SWITCH , '<STR_LIT>' , <EOL> subnet = '<STR_LIT>' , owner = owner , transaction = mock . ANY ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_static_route_created . called ) <EOL> expected_create_calls = [ ] <EOL> for x in range ( len ( tenants ) ) : <EOL> ep = eps [ x ] <EOL> l3p = l3p_list [ x ] <EOL> l3out = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> ext_epg = ep [ '<STR_LIT:id>' ] <EOL> tenant = APIC_PRE_L3OUT_TENANT if self . pre_l3out else owner <EOL> if self . nat_enabled : <EOL> l3out = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) <EOL> ext_epg = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ext_epg ) <EOL> tenant = tenants [ x ] <EOL> expected_create_calls . append ( <EOL> mock . call ( l3out , subnet = '<STR_LIT>' , <EOL> external_epg = ext_epg , owner = tenant , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_create_calls , <EOL> mgr . ensure_external_epg_created . call_args_list ) <EOL> self . assertFalse ( mgr . ensure_static_route_deleted . called ) <EOL> self . assertFalse ( mgr . ensure_external_epg_routes_deleted . called ) <EOL> self . assertFalse ( mgr . ensure_next_hop_deleted . called ) <EOL> def test_route_update_add_1 ( self ) : <EOL> self . _test_route_update_add ( shared_es = True ) <EOL> def test_route_update_add_2 ( self ) : <EOL> self . _test_route_update_add ( shared_es = False ) <EOL> def test_es_create_no_cidr_with_routes ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> nh = '<STR_LIT>' if self . pre_l3out else '<STR_LIT>' <EOL> self . create_external_segment ( <EOL> name = '<STR_LIT>' , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : nh } ] , <EOL> expected_res_status = <NUM_LIT> ) <EOL> def test_implicit_es_router_gw_ip ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT:default>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT:default>' , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None } ] ) [ '<STR_LIT>' ] <EOL> l3p = self . create_l3_policy ( ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( es [ '<STR_LIT:id>' ] , <EOL> l3p [ '<STR_LIT>' ] . keys ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> l3p [ '<STR_LIT>' ] [ es [ '<STR_LIT:id>' ] ] [ <NUM_LIT:0> ] ) <EOL> def test_plug_l3p_to_es_with_multi_ep ( self ) : <EOL> tenants = ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if self . nat_enabled else [ '<STR_LIT>' ] ) <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> ext_routes = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> es_list = [ <EOL> self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = True , <EOL> expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { <EOL> '<STR_LIT>' : ext_routes [ x ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> for x in range ( <NUM_LIT:2> ) ] <EOL> ep_list = [ ] <EOL> for x in range ( len ( tenants ) ) : <EOL> ep = self . create_external_policy ( <EOL> name = ( x < <NUM_LIT:2> and APIC_EXTERNAL_EPG or '<STR_LIT>' ) , <EOL> external_segments = [ e [ '<STR_LIT:id>' ] for e in es_list ] , <EOL> tenant_id = tenants [ x ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> ep_list . append ( ep ) <EOL> mgr = self . driver . apic_manager <EOL> mgr . ensure_external_epg_created . reset_mock ( ) <EOL> mgr . set_contract_for_external_epg . reset_mock ( ) <EOL> ep = ep_list [ <NUM_LIT:0> ] <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = tenants [ <NUM_LIT:0> ] , <EOL> external_segments = { x [ '<STR_LIT:id>' ] : [ ] for x in es_list } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> expected_create_calls = [ ] <EOL> expected_assoc_calls = [ ] <EOL> expected_contract_calls = [ ] <EOL> if self . nat_enabled : <EOL> for es in es_list : <EOL> if not self . pre_l3out : <EOL> expected_create_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , subnet = '<STR_LIT>' , <EOL> external_epg = '<STR_LIT>' % es [ '<STR_LIT:id>' ] , <EOL> owner = self . common_tenant , <EOL> transaction = mock . ANY ) ) <EOL> expected_create_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> subnet = es [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> external_epg = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY ) ) <EOL> expected_assoc_calls . append ( <EOL> mock . call ( l3p [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) , <EOL> "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> target_owner = self . common_tenant , <EOL> transaction = mock . ANY ) ) <EOL> l3out = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> l3out_owner = ( APIC_PRE_L3OUT_TENANT <EOL> if self . pre_l3out else self . common_tenant ) <EOL> nat_contract = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> ext_epg = ( ep [ '<STR_LIT:name>' ] <EOL> if self . pre_l3out else ( '<STR_LIT>' % es [ '<STR_LIT:id>' ] ) ) <EOL> expected_contract_calls . append ( <EOL> mock . call ( l3out , nat_contract , <EOL> external_epg = ext_epg , <EOL> owner = l3out_owner , <EOL> provided = True , transaction = mock . ANY ) ) <EOL> expected_contract_calls . append ( <EOL> mock . call ( l3out , nat_contract , <EOL> external_epg = ext_epg , <EOL> owner = l3out_owner , <EOL> provided = False , transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_create_calls , <EOL> mgr . ensure_external_epg_created . call_args_list ) <EOL> self . _check_call_list ( expected_assoc_calls , <EOL> mgr . associate_external_epg_to_nat_epg . call_args_list ) <EOL> self . _check_call_list ( expected_contract_calls , <EOL> mgr . set_contract_for_external_epg . call_args_list ) <EOL> class TestExternalSegmentNoNat ( TestExternalSegment ) : <EOL> def setUp ( self ) : <EOL> super ( TestExternalSegmentNoNat , self ) . setUp ( nat_enabled = False ) <EOL> class TestExternalSegmentPreL3Out ( TestExternalSegment ) : <EOL> def setUp ( self , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> super ( TestExternalSegmentPreL3Out , self ) . setUp ( ** kwargs ) <EOL> def test_query_l3out_info ( self ) : <EOL> self . driver . _query_l3out_info = self . orig_query_l3out_info <EOL> ctx1 = [ { <EOL> '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } } ] <EOL> mgr = self . driver . apic_manager <EOL> mgr . apic . l3extOut . get_subtree . return_value = ctx1 <EOL> info = self . driver . _query_l3out_info ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , info [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , info [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , info [ '<STR_LIT>' ] ) <EOL> mgr . apic . l3extOut . get_subtree . reset_mock ( ) <EOL> mgr . apic . l3extOut . get_subtree . return_value = [ ] <EOL> info = self . driver . _query_l3out_info ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( None , info ) <EOL> expected_calls = [ <EOL> mock . call ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> mock . call ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . apic . l3extOut . get_subtree . call_args_list ) <EOL> def test_l3out_tenant ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> self . driver . _query_l3out_info . return_value [ '<STR_LIT>' ] = ( <EOL> apic_mapper . ApicName ( '<STR_LIT>' ) ) <EOL> res = self . create_external_segment ( name = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> self . create_external_segment ( name = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> ) <EOL> def test_edge_nat_wrong_L3out_IF_type_rejected ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat = True ) <EOL> self . driver . _query_l3out_info . return_value [ '<STR_LIT>' ] = ( <EOL> [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' } , <EOL> u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' <EOL> } } } ] } } ] } } ] ) <EOL> res = self . create_external_segment ( <EOL> name = '<STR_LIT>' , expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_edge_nat_wrong_L3out_OSPF_Auth_type_rejected ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat = True ) <EOL> self . driver . _query_l3out_info . return_value [ '<STR_LIT>' ] = ( <EOL> [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' } , <EOL> u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' <EOL> } } } ] } } ] } } ] ) <EOL> res = self . create_external_segment ( <EOL> name = '<STR_LIT>' , expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_edge_nat_wrong_L3out_BGP_Auth_type_rejected ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> is_edge_nat = True ) <EOL> self . driver . _query_l3out_info . return_value [ '<STR_LIT>' ] = ( <EOL> [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' } , <EOL> u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT:type>' : u'<STR_LIT>' } } } , <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT:type>' : u'<STR_LIT>' } } } , <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT:type>' : u'<STR_LIT>' } } } ] } } ] } } ] ) <EOL> res = self . create_external_segment ( <EOL> name = '<STR_LIT>' , expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> self . driver . _query_l3out_info . return_value [ '<STR_LIT>' ] = ( <EOL> [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : u'<STR_LIT>' } , <EOL> u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : [ { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT:type>' : u'<STR_LIT>' } } } , <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT:type>' : u'<STR_LIT:none>' } } } , <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT>' : <EOL> { u'<STR_LIT:type>' : u'<STR_LIT>' } } } ] } } ] } } ] ) <EOL> res = self . create_external_segment ( <EOL> name = '<STR_LIT>' , expected_res_status = <NUM_LIT> ) <EOL> class TestExternalSegmentNoNatPreL3Out ( TestExternalSegmentPreL3Out ) : <EOL> def setUp ( self ) : <EOL> super ( TestExternalSegmentNoNatPreL3Out , self ) . setUp ( <EOL> nat_enabled = False ) <EOL> class TestExternalPolicy ( ApicMappingTestCase ) : <EOL> def test_creation_noop ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> external_routes = [ ] , expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , expected_res_status = <NUM_LIT> ) <EOL> mgr = self . driver . apic_manager <EOL> if self . nat_enabled and not self . pre_l3out : <EOL> mgr . ensure_external_epg_created . assert_called_once_with ( <EOL> es [ '<STR_LIT:id>' ] , subnet = '<STR_LIT>' , <EOL> external_epg = ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) , owner = es [ '<STR_LIT>' ] , <EOL> transaction = mock . ANY ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_external_epg_created . called ) <EOL> mgr . ensure_external_epg_created . reset_mock ( ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> self . create_external_policy ( <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , expected_res_status = <NUM_LIT> ) <EOL> self . assertFalse ( mgr . ensure_external_epg_created . called ) <EOL> def test_create_shared ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> external_routes = [ ] , shared = True , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> res = self . create_external_policy ( <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , shared = True , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_update_shared ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> external_routes = [ ] , shared = True , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> ep = self . create_external_policy ( <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> res = self . update_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , shared = True , expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def _test_creation_no_prs ( self , shared_es ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es_list = [ <EOL> self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> for x in range ( <NUM_LIT:3> ) ] <EOL> l3p_list = [ ] <EOL> for x in xrange ( len ( es_list ) ) : <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = shared_es and '<STR_LIT>' or es_list [ x ] [ '<STR_LIT>' ] , <EOL> external_segments = { es_list [ x ] [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l3p_list . append ( l3p ) <EOL> ep = self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> external_segments = [ x [ '<STR_LIT:id>' ] for x in es_list ] , <EOL> tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> owner = ( es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es <EOL> else self . common_tenant ) <EOL> l3p_owner = l3p_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> expected_create_calls = [ ] <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> l3p = l3p_list [ x ] <EOL> if self . nat_enabled : <EOL> if not self . pre_l3out : <EOL> expected_create_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , subnet = '<STR_LIT>' , <EOL> external_epg = "<STR_LIT>" % es [ '<STR_LIT:id>' ] , owner = owner , <EOL> transaction = mock . ANY ) ) <EOL> expected_create_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> subnet = '<STR_LIT>' , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> owner = l3p_owner , <EOL> transaction = mock . ANY ) ) <EOL> elif not self . pre_l3out : <EOL> expected_create_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , subnet = '<STR_LIT>' , <EOL> external_epg = ep [ '<STR_LIT:id>' ] , owner = owner , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_create_calls , <EOL> mgr . ensure_external_epg_created . call_args_list ) <EOL> if self . nat_enabled : <EOL> expected_contract_calls = [ ] <EOL> ext_epg_tenant = ( APIC_PRE_L3OUT_TENANT if self . pre_l3out <EOL> else owner ) <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> ext_epg = ( APIC_EXTERNAL_EPG if self . pre_l3out <EOL> else "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) <EOL> es_name = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> nat_contract = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> expected_contract_calls . extend ( [ <EOL> mock . call ( es_name , nat_contract , <EOL> external_epg = ext_epg , owner = ext_epg_tenant , <EOL> provided = True , transaction = mock . ANY ) , <EOL> mock . call ( es_name , nat_contract , <EOL> external_epg = ext_epg , owner = ext_epg_tenant , <EOL> provided = False , transaction = mock . ANY ) ] ) <EOL> self . _check_call_list ( expected_contract_calls , <EOL> mgr . set_contract_for_external_epg . call_args_list ) <EOL> else : <EOL> self . assertFalse ( mgr . set_contract_for_external_epg . called ) <EOL> def test_creation_no_prs_1 ( self ) : <EOL> self . _test_creation_no_prs ( shared_es = True ) <EOL> def test_creation_no_prs_2 ( self ) : <EOL> self . _test_creation_no_prs ( shared_es = False ) <EOL> def _test_update_no_prs ( self , shared_es ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es_list = [ <EOL> self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> for x in range ( <NUM_LIT:3> ) ] <EOL> l3p_list = [ ] <EOL> for x in xrange ( len ( es_list ) ) : <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = shared_es and '<STR_LIT>' or es_list [ x ] [ '<STR_LIT>' ] , <EOL> external_segments = { es_list [ x ] [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l3p_list . append ( l3p ) <EOL> ep = self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> ep = self . update_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , tenant_id = ep [ '<STR_LIT>' ] , <EOL> external_segments = [ x [ '<STR_LIT:id>' ] for x in es_list ] ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> owner = ( es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es <EOL> else self . common_tenant ) <EOL> l3p_owner = l3p_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> expected_create_calls = [ ] <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> l3p = l3p_list [ x ] <EOL> if self . nat_enabled : <EOL> if not self . pre_l3out : <EOL> expected_create_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , subnet = '<STR_LIT>' , <EOL> external_epg = "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> owner = owner , transaction = mock . ANY ) ) <EOL> expected_create_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> subnet = '<STR_LIT>' , <EOL> external_epg = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , transaction = mock . ANY ) ) <EOL> elif not self . pre_l3out : <EOL> expected_create_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , subnet = '<STR_LIT>' , <EOL> external_epg = ep [ '<STR_LIT:id>' ] , <EOL> owner = owner , transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_create_calls , <EOL> mgr . ensure_external_epg_created . call_args_list ) <EOL> if self . nat_enabled : <EOL> expected_contract_calls = [ ] <EOL> ext_epg_tenant = ( APIC_PRE_L3OUT_TENANT if self . pre_l3out <EOL> else owner ) <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> ext_epg = ( APIC_EXTERNAL_EPG if self . pre_l3out <EOL> else "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) <EOL> es_name = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> nat_contract = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> expected_contract_calls . extend ( [ <EOL> mock . call ( es_name , nat_contract , <EOL> external_epg = ext_epg , owner = ext_epg_tenant , <EOL> provided = True , transaction = mock . ANY ) , <EOL> mock . call ( es_name , nat_contract , <EOL> external_epg = ext_epg , owner = ext_epg_tenant , <EOL> provided = False , transaction = mock . ANY ) ] ) <EOL> self . _check_call_list ( expected_contract_calls , <EOL> mgr . set_contract_for_external_epg . call_args_list ) <EOL> else : <EOL> self . assertFalse ( mgr . set_contract_for_external_epg . called ) <EOL> ep = self . update_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , tenant_id = ep [ '<STR_LIT>' ] , <EOL> external_segments = [ ] ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> expected_create_calls = [ ] <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> l3p = l3p_list [ x ] <EOL> if self . nat_enabled : <EOL> if not self . pre_l3out : <EOL> expected_create_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , <EOL> external_epg = "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) ) <EOL> expected_create_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> owner = l3p_owner , <EOL> external_epg = "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) ) <EOL> elif not self . pre_l3out : <EOL> expected_create_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , owner = owner , external_epg = ep [ '<STR_LIT:id>' ] ) ) <EOL> self . _check_call_list ( expected_create_calls , <EOL> mgr . ensure_external_epg_deleted . call_args_list ) <EOL> if self . nat_enabled and self . pre_l3out : <EOL> expected_contract_calls = [ ] <EOL> ext_epg_tenant = APIC_PRE_L3OUT_TENANT <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> nat_contract = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> expected_contract_calls . extend ( [ <EOL> mock . call ( es [ '<STR_LIT:name>' ] , nat_contract , <EOL> external_epg = APIC_EXTERNAL_EPG , owner = ext_epg_tenant , <EOL> provided = True , transaction = mock . ANY ) , <EOL> mock . call ( es [ '<STR_LIT:name>' ] , nat_contract , <EOL> external_epg = APIC_EXTERNAL_EPG , owner = ext_epg_tenant , <EOL> provided = False , transaction = mock . ANY ) ] ) <EOL> self . _check_call_list ( expected_contract_calls , <EOL> mgr . unset_contract_for_external_epg . call_args_list ) <EOL> else : <EOL> self . assertFalse ( mgr . unset_contract_for_external_epg . called ) <EOL> def test_update_no_prs_1 ( self ) : <EOL> self . _test_update_no_prs ( shared_es = True ) <EOL> def test_update_no_prs_2 ( self ) : <EOL> self . _test_update_no_prs ( shared_es = False ) <EOL> def _test_create_with_prs ( self , shared_es , shared_prs ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es_list = [ <EOL> self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> for x in range ( <NUM_LIT:3> ) ] <EOL> l3p_list = [ ] <EOL> for x in xrange ( len ( es_list ) ) : <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = shared_es and '<STR_LIT>' or es_list [ x ] [ '<STR_LIT>' ] , <EOL> external_segments = { es_list [ x ] [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l3p_list . append ( l3p ) <EOL> prov = self . _create_policy_rule_set_on_shared ( <EOL> shared = shared_prs , <EOL> tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not ( <EOL> shared_es | shared_prs ) else '<STR_LIT>' ) <EOL> cons = self . _create_policy_rule_set_on_shared ( <EOL> shared = shared_prs , <EOL> tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not ( <EOL> shared_es | shared_prs ) else '<STR_LIT>' ) <EOL> ep = self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> provided_policy_rule_sets = { prov [ '<STR_LIT:id>' ] : '<STR_LIT>' } , <EOL> consumed_policy_rule_sets = { cons [ '<STR_LIT:id>' ] : '<STR_LIT>' } , <EOL> tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> external_segments = [ x [ '<STR_LIT:id>' ] for x in es_list ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> owner = ( es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es <EOL> else self . common_tenant ) <EOL> l3p_owner = l3p_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> expected_calls = [ ] <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> l3p = l3p_list [ x ] <EOL> nat = self . nat_enabled <EOL> external_epg = APIC_EXTERNAL_EPG if self . pre_l3out else ( <EOL> ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) if nat else ep [ '<STR_LIT:id>' ] ) <EOL> ext_epg_tenant = ( APIC_PRE_L3OUT_TENANT if self . pre_l3out else <EOL> owner ) <EOL> es_name = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> expected_calls . append ( <EOL> mock . call ( es_name , <EOL> ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) if nat else prov [ '<STR_LIT:id>' ] , <EOL> external_epg = external_epg , <EOL> provided = True , owner = ext_epg_tenant , <EOL> transaction = mock . ANY ) ) <EOL> expected_calls . append ( <EOL> mock . call ( es_name , <EOL> ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) if nat else cons [ '<STR_LIT:id>' ] , <EOL> external_epg = external_epg , <EOL> provided = False , owner = ext_epg_tenant , <EOL> transaction = mock . ANY ) ) <EOL> if nat : <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , prov [ '<STR_LIT:id>' ] , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> provided = True , owner = l3p_owner , <EOL> transaction = mock . ANY ) ) <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , cons [ '<STR_LIT:id>' ] , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> provided = False , owner = l3p_owner , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . set_contract_for_external_epg . call_args_list ) <EOL> def test_create_with_prs_1 ( self ) : <EOL> self . _test_create_with_prs ( shared_es = True , shared_prs = True ) <EOL> def test_create_with_prs_2 ( self ) : <EOL> self . _test_create_with_prs ( shared_es = True , shared_prs = False ) <EOL> def test_create_with_prs_3 ( self ) : <EOL> self . _test_create_with_prs ( shared_es = False , shared_prs = False ) <EOL> def test_create_with_prs_4 ( self ) : <EOL> self . _test_create_with_prs ( shared_es = False , shared_prs = True ) <EOL> def _test_update_add_prs ( self , shared_es , shared_prs ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es_list = [ <EOL> self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = shared_es , <EOL> expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> for x in range ( <NUM_LIT:3> ) ] <EOL> l3p_list = [ ] <EOL> for x in xrange ( len ( es_list ) ) : <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = shared_es and '<STR_LIT>' or es_list [ x ] [ '<STR_LIT>' ] , <EOL> external_segments = { es_list [ x ] [ '<STR_LIT:id>' ] : [ ] } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l3p_list . append ( l3p ) <EOL> prov = self . _create_policy_rule_set_on_shared ( <EOL> shared = shared_prs , tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not ( <EOL> shared_es | shared_prs ) else '<STR_LIT>' ) <EOL> cons = self . _create_policy_rule_set_on_shared ( <EOL> shared = shared_prs , tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not ( <EOL> shared_es | shared_prs ) else '<STR_LIT>' ) <EOL> ep = self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> external_segments = [ x [ '<STR_LIT:id>' ] for x in es_list ] , <EOL> tenant_id = es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es else '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> ep = self . update_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , tenant_id = ep [ '<STR_LIT>' ] , <EOL> provided_policy_rule_sets = { prov [ '<STR_LIT:id>' ] : '<STR_LIT>' } , <EOL> consumed_policy_rule_sets = { cons [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> owner = ( es_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] if not shared_es <EOL> else self . common_tenant ) <EOL> l3p_owner = l3p_list [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> expected_calls = [ ] <EOL> nat = self . nat_enabled <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> l3p = l3p_list [ x ] <EOL> external_epg = APIC_EXTERNAL_EPG if self . pre_l3out else ( <EOL> ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) if nat else ep [ '<STR_LIT:id>' ] ) <EOL> ext_epg_tenant = ( APIC_PRE_L3OUT_TENANT if self . pre_l3out else <EOL> owner ) <EOL> es_name = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> expected_calls . append ( <EOL> mock . call ( es_name , <EOL> ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) if nat else prov [ '<STR_LIT:id>' ] , <EOL> external_epg = external_epg , <EOL> provided = True , owner = ext_epg_tenant , <EOL> transaction = mock . ANY ) ) <EOL> expected_calls . append ( <EOL> mock . call ( es_name , <EOL> ( "<STR_LIT>" % es [ '<STR_LIT:id>' ] ) if nat else cons [ '<STR_LIT:id>' ] , <EOL> external_epg = external_epg , <EOL> provided = False , owner = ext_epg_tenant , <EOL> transaction = mock . ANY ) ) <EOL> if nat : <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , prov [ '<STR_LIT:id>' ] , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> provided = True , owner = l3p_owner , <EOL> transaction = mock . ANY ) ) <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , cons [ '<STR_LIT:id>' ] , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> provided = False , owner = l3p_owner , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . set_contract_for_external_epg . call_args_list ) <EOL> ep = self . update_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , provided_policy_rule_sets = { } , <EOL> consumed_policy_rule_sets = { } , <EOL> tenant_id = ep [ '<STR_LIT>' ] ) [ '<STR_LIT>' ] <EOL> expected_calls = [ ] <EOL> for x in range ( len ( es_list ) ) : <EOL> es = es_list [ x ] <EOL> l3p = l3p_list [ x ] <EOL> if nat : <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , prov [ '<STR_LIT:id>' ] , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> provided = True , owner = l3p_owner , <EOL> transaction = mock . ANY ) ) <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , cons [ '<STR_LIT:id>' ] , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> provided = False , owner = l3p_owner , <EOL> transaction = mock . ANY ) ) <EOL> else : <EOL> external_epg = ( APIC_EXTERNAL_EPG if self . pre_l3out <EOL> else ep [ '<STR_LIT:id>' ] ) <EOL> ext_epg_tenant = ( APIC_PRE_L3OUT_TENANT if self . pre_l3out else <EOL> owner ) <EOL> es_name = es [ '<STR_LIT:name>' if self . pre_l3out else '<STR_LIT:id>' ] <EOL> expected_calls . append ( <EOL> mock . call ( es_name , prov [ '<STR_LIT:id>' ] , <EOL> external_epg = external_epg , <EOL> provided = True , owner = ext_epg_tenant , <EOL> transaction = mock . ANY ) ) <EOL> expected_calls . append ( <EOL> mock . call ( es_name , cons [ '<STR_LIT:id>' ] , <EOL> external_epg = external_epg , <EOL> provided = False , owner = ext_epg_tenant , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( <EOL> expected_calls , mgr . unset_contract_for_external_epg . call_args_list ) <EOL> def test_update_add_prs_1 ( self ) : <EOL> self . _test_update_add_prs ( shared_es = True , shared_prs = True ) <EOL> def test_update_add_prs_2 ( self ) : <EOL> self . _test_update_add_prs ( shared_es = True , shared_prs = False ) <EOL> def test_update_add_prs_3 ( self ) : <EOL> self . _test_update_add_prs ( shared_es = False , shared_prs = False ) <EOL> def test_update_add_prs_4 ( self ) : <EOL> self . _test_update_add_prs ( shared_es = False , shared_prs = True ) <EOL> def test_update_add_prs_unsupported ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> prov = self . _create_policy_rule_set_on_shared ( ) <EOL> cons = self . _create_policy_rule_set_on_shared ( ) <EOL> ep = self . create_external_policy ( <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> self . update_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , expected_res_status = <NUM_LIT:200> , tenant_id = ep [ '<STR_LIT>' ] , <EOL> provided_policy_rule_sets = { prov [ '<STR_LIT:id>' ] : '<STR_LIT>' } , <EOL> consumed_policy_rule_sets = { cons [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) [ '<STR_LIT>' ] <EOL> mgr = self . driver . apic_manager <EOL> self . assertFalse ( mgr . set_contract_for_external_epg . called ) <EOL> def _test_multi_policy_single_tenant ( self , shared_es ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , <EOL> expected_res_status = <NUM_LIT> , shared = shared_es , <EOL> external_routes = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> owner = '<STR_LIT>' if shared_es else es [ '<STR_LIT>' ] <EOL> self . create_external_policy ( <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> tenant_id = owner , <EOL> expected_res_status = <NUM_LIT> ) <EOL> res = self . create_external_policy ( <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> tenant_id = owner , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> ep2 = self . create_external_policy ( <EOL> tenant_id = owner , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> res = self . update_external_policy ( <EOL> ep2 [ '<STR_LIT:id>' ] , external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> tenant_id = owner , <EOL> expected_res_status = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_multi_policy_single_tenant_1 ( self ) : <EOL> self . _test_multi_policy_single_tenant ( True ) <EOL> def test_multi_policy_single_tenant_2 ( self ) : <EOL> self . _test_multi_policy_single_tenant ( False ) <EOL> def test_multi_policy_multi_tenant ( self ) : <EOL> tenants = ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if self . nat_enabled else [ '<STR_LIT>' ] ) <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> ext_routes = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> es_list = [ <EOL> self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = True , <EOL> expected_res_status = <NUM_LIT> , <EOL> external_routes = [ { <EOL> '<STR_LIT>' : ext_routes [ x ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) [ '<STR_LIT>' ] <EOL> for x in range ( <NUM_LIT:2> ) ] <EOL> l3p_list = [ ] <EOL> for x in xrange ( len ( tenants ) ) : <EOL> l3p = self . create_l3_policy ( <EOL> shared = False , <EOL> tenant_id = tenants [ x ] , <EOL> external_segments = { x [ '<STR_LIT:id>' ] : [ ] for x in es_list } , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> l3p_list . append ( l3p ) <EOL> ep_list = [ ] <EOL> mgr = self . driver . apic_manager <EOL> for x in range ( len ( tenants ) ) : <EOL> ep = self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> external_segments = [ e [ '<STR_LIT:id>' ] for e in es_list ] , <EOL> tenant_id = tenants [ x ] , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> ep_list . append ( ep ) <EOL> l3p = l3p_list [ x ] <EOL> expected_calls = [ ] <EOL> for es in es_list : <EOL> if self . nat_enabled : <EOL> if not self . pre_l3out : <EOL> expected_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , subnet = '<STR_LIT>' , <EOL> external_epg = "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> owner = self . common_tenant , <EOL> transaction = mock . ANY ) ) <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> subnet = es [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> owner = tenants [ x ] , <EOL> transaction = mock . ANY ) ) <EOL> elif not self . pre_l3out : <EOL> expected_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , <EOL> subnet = es [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> external_epg = ep [ '<STR_LIT:id>' ] , owner = self . common_tenant , <EOL> transaction = mock . ANY ) ) <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . ensure_external_epg_created . call_args_list ) <EOL> mgr . ensure_external_epg_created . reset_mock ( ) <EOL> expected_calls = [ ] <EOL> for x in range ( len ( tenants ) ) : <EOL> ep = ep_list [ x ] <EOL> self . delete_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , tenant_id = ep [ '<STR_LIT>' ] , <EOL> expected_res_status = webob . exc . HTTPNoContent . code ) <EOL> l3p = l3p_list [ x ] <EOL> for es in es_list : <EOL> if self . nat_enabled : <EOL> expected_calls . append ( <EOL> mock . call ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , es [ '<STR_LIT:id>' ] ) , <EOL> external_epg = ( "<STR_LIT>" % ( l3p [ '<STR_LIT:id>' ] , ep [ '<STR_LIT:id>' ] ) ) , <EOL> owner = tenants [ x ] ) ) <EOL> elif not self . pre_l3out : <EOL> expected_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , external_epg = ep [ '<STR_LIT:id>' ] , <EOL> owner = self . common_tenant ) ) <EOL> if self . nat_enabled and not self . pre_l3out : <EOL> for es in es_list : <EOL> expected_calls . append ( <EOL> mock . call ( es [ '<STR_LIT:id>' ] , external_epg = "<STR_LIT>" % es [ '<STR_LIT:id>' ] , <EOL> owner = self . common_tenant ) ) <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . ensure_external_epg_deleted . call_args_list ) <EOL> class TestExternalPolicyNoNat ( TestExternalPolicy ) : <EOL> def setUp ( self ) : <EOL> super ( TestExternalPolicyNoNat , self ) . setUp ( nat_enabled = False ) <EOL> class TestExternalPolicyPreL3Out ( TestExternalPolicy ) : <EOL> def setUp ( self ) : <EOL> super ( TestExternalPolicyPreL3Out , self ) . setUp ( <EOL> pre_existing_l3out = True ) <EOL> def test_multi_tenant_delete ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = True , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> ep_list = [ <EOL> self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> tenant_id = tnnt , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> for tnnt in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> for ep in ep_list : <EOL> self . delete_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , tenant_id = ep [ '<STR_LIT>' ] , <EOL> expected_res_status = webob . exc . HTTPNoContent . code ) <EOL> nat_contract = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> expected_calls = [ <EOL> mock . call ( es [ '<STR_LIT:name>' ] , nat_contract , <EOL> external_epg = APIC_EXTERNAL_EPG , <EOL> provided = True , owner = APIC_PRE_L3OUT_TENANT , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( es [ '<STR_LIT:name>' ] , nat_contract , <EOL> external_epg = APIC_EXTERNAL_EPG , <EOL> provided = False , owner = APIC_PRE_L3OUT_TENANT , <EOL> transaction = mock . ANY ) <EOL> ] <EOL> mgr = self . driver . apic_manager <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . unset_contract_for_external_epg . call_args_list ) <EOL> def test_multi_tenant_update_dissociate ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( <EOL> name = '<STR_LIT>' , cidr = '<STR_LIT>' , shared = True , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> ep_list = [ <EOL> self . create_external_policy ( <EOL> name = APIC_EXTERNAL_EPG , <EOL> external_segments = [ es [ '<STR_LIT:id>' ] ] , <EOL> tenant_id = tnnt , <EOL> expected_res_status = <NUM_LIT> ) [ '<STR_LIT>' ] <EOL> for tnnt in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> for ep in ep_list : <EOL> self . update_external_policy ( <EOL> ep [ '<STR_LIT:id>' ] , tenant_id = ep [ '<STR_LIT>' ] , <EOL> external_segments = [ ] , <EOL> expected_res_status = <NUM_LIT:200> ) <EOL> nat_contract = "<STR_LIT>" % es [ '<STR_LIT:id>' ] <EOL> expected_calls = [ <EOL> mock . call ( es [ '<STR_LIT:name>' ] , nat_contract , <EOL> external_epg = APIC_EXTERNAL_EPG , <EOL> provided = True , owner = APIC_PRE_L3OUT_TENANT , <EOL> transaction = mock . ANY ) , <EOL> mock . call ( es [ '<STR_LIT:name>' ] , nat_contract , <EOL> external_epg = APIC_EXTERNAL_EPG , <EOL> provided = False , owner = APIC_PRE_L3OUT_TENANT , <EOL> transaction = mock . ANY ) <EOL> ] <EOL> mgr = self . driver . apic_manager <EOL> self . _check_call_list ( expected_calls , <EOL> mgr . unset_contract_for_external_epg . call_args_list ) <EOL> class TestExternalPolicyNoNatPreL3Out ( TestExternalPolicy ) : <EOL> def setUp ( self ) : <EOL> super ( TestExternalPolicyNoNatPreL3Out , self ) . setUp ( <EOL> nat_enabled = False , pre_existing_l3out = True ) <EOL> class TestNatPool ( ApicMappingTestCase ) : <EOL> def test_overlap_nat_pool_create ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> mgr = self . driver . apic_manager <EOL> mgr . ext_net_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> es = self . create_external_segment ( name = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code ) [ '<STR_LIT>' ] <EOL> res = self . create_nat_pool ( <EOL> external_segment_id = es [ '<STR_LIT:id>' ] , <EOL> ip_version = <NUM_LIT:4> , ip_pool = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPBadRequest . code ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> res = self . create_nat_pool ( <EOL> external_segment_id = es [ '<STR_LIT:id>' ] , <EOL> ip_version = <NUM_LIT:4> , ip_pool = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPBadRequest . code ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def test_overlap_nat_pool_update ( self ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es1 = self . create_external_segment ( name = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code ) [ '<STR_LIT>' ] <EOL> es2 = self . create_external_segment ( name = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code ) [ '<STR_LIT>' ] <EOL> nat_pool = self . create_nat_pool ( <EOL> external_segment_id = es1 [ '<STR_LIT:id>' ] , <EOL> ip_version = <NUM_LIT:4> , ip_pool = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code ) [ '<STR_LIT>' ] <EOL> res = self . update_nat_pool ( nat_pool [ '<STR_LIT:id>' ] , <EOL> external_segment_id = es2 [ '<STR_LIT:id>' ] , <EOL> expected_res_status = webob . exc . HTTPBadRequest . code ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> res [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] ) <EOL> def _test_nat_bd_subnet_created_deleted ( self , shared ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( name = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> nat_pool = self . create_nat_pool ( <EOL> external_segment_id = es [ '<STR_LIT:id>' ] , <EOL> ip_version = <NUM_LIT:4> , ip_pool = '<STR_LIT>' , shared = shared , <EOL> expected_res_status = webob . exc . HTTPCreated . code ) [ '<STR_LIT>' ] <EOL> owner = es [ '<STR_LIT>' ] if not shared else self . common_tenant <EOL> mgr = self . driver . apic_manager <EOL> if self . nat_enabled : <EOL> mgr . ensure_subnet_created_on_apic . assert_called_with ( <EOL> owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_subnet_created_on_apic . called ) <EOL> self . delete_nat_pool ( nat_pool [ '<STR_LIT:id>' ] , <EOL> expected_res_status = webob . exc . HTTPNoContent . code ) <EOL> if self . nat_enabled : <EOL> mgr . ensure_subnet_deleted_on_apic . assert_called_with ( <EOL> owner , "<STR_LIT>" % es [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_subnet_deleted_on_apic . called ) <EOL> def test_nat_bd_subnet_create_delete_1 ( self ) : <EOL> self . _test_nat_bd_subnet_created_deleted ( False ) <EOL> def test_nat_bd_subnet_create_delete_2 ( self ) : <EOL> self . _test_nat_bd_subnet_created_deleted ( True ) <EOL> def _test_nat_bd_subnet_updated ( self , shared ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es1 = self . create_external_segment ( name = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> es2 = self . create_external_segment ( name = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> nat_pool = self . create_nat_pool ( <EOL> external_segment_id = es1 [ '<STR_LIT:id>' ] , <EOL> ip_version = <NUM_LIT:4> , ip_pool = '<STR_LIT>' , shared = shared , <EOL> expected_res_status = webob . exc . HTTPCreated . code ) [ '<STR_LIT>' ] <EOL> owner = es1 [ '<STR_LIT>' ] if not shared else self . common_tenant <EOL> mgr = self . driver . apic_manager <EOL> mgr . ensure_subnet_created_on_apic . reset_mock ( ) <EOL> nat_pool = self . update_nat_pool ( nat_pool [ '<STR_LIT:id>' ] , <EOL> external_segment_id = es2 [ '<STR_LIT:id>' ] , <EOL> expected_res_status = webob . exc . HTTPOk . code ) [ '<STR_LIT>' ] <EOL> if self . nat_enabled : <EOL> mgr . ensure_subnet_deleted_on_apic . assert_called_with ( <EOL> owner , "<STR_LIT>" % es1 [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> mgr . ensure_subnet_created_on_apic . assert_called_with ( <EOL> owner , "<STR_LIT>" % es2 [ '<STR_LIT:id>' ] , '<STR_LIT>' ) <EOL> else : <EOL> self . assertFalse ( mgr . ensure_subnet_created_on_apic . called ) <EOL> self . assertFalse ( mgr . ensure_subnet_deleted_on_apic . called ) <EOL> def test_nat_bd_subnet_update_1 ( self ) : <EOL> self . _test_nat_bd_subnet_updated ( False ) <EOL> def test_nat_bd_subnet_update_2 ( self ) : <EOL> self . _test_nat_bd_subnet_updated ( True ) <EOL> def _test_create_fip ( self , shared ) : <EOL> self . _mock_external_dict ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> es = self . create_external_segment ( name = '<STR_LIT>' , <EOL> expected_res_status = webob . exc . HTTPCreated . code , <EOL> shared = shared ) [ '<STR_LIT>' ] <EOL> self . create_nat_pool ( external_segment_id = es [ '<STR_LIT:id>' ] , <EOL> ip_version = <NUM_LIT:4> , ip_pool = '<STR_LIT>' , shared = shared , <EOL> expected_res_status = webob . exc . HTTPCreated . code ) <EOL> subnet = self . _get_object ( '<STR_LIT>' , es [ '<STR_LIT>' ] , <EOL> self . api ) [ '<STR_LIT>' ] <EOL> fip_dict = { '<STR_LIT>' : subnet [ '<STR_LIT>' ] } <EOL> fip_id = self . driver . create_floatingip_in_nat_pool ( <EOL> context . get_admin_context ( ) , <EOL> es [ '<STR_LIT>' ] , { '<STR_LIT>' : fip_dict } ) <EOL> self . assertIsNotNone ( fip_id ) <EOL> fip = self . _get_object ( <EOL> '<STR_LIT>' , fip_id , self . ext_api ) [ '<STR_LIT>' ] <EOL> self . assertTrue ( <EOL> netaddr . IPAddress ( fip [ '<STR_LIT>' ] ) in <EOL> netaddr . IPNetwork ( '<STR_LIT>' ) ) <EOL> def test_create_fip ( self ) : <EOL> self . _test_create_fip ( False ) <EOL> def test_create_fip_shared ( self ) : <EOL> self . _test_create_fip ( True ) <EOL> class TestNatPoolNoNat ( TestNatPool ) : <EOL> def setUp ( self ) : <EOL> super ( TestNatPoolNoNat , self ) . setUp ( nat_enabled = False ) </s>
