x
/
group-based-policy
Code Issues Proposed changes

Merge "Add subnet scope extension support"

This commit is contained in:
Zuul 2023-10-14 17:50:43 +00:00 committed by Gerrit Code Review
parent 1a825ef921 717ab3b5f4
commit aac6fa7031
7 changed files with 411 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
gbpservice/neutron/db/migration/alembic_migrations/versions/HEAD
View File

@ -1 +1 @@
8c5b556b4df1
dc99863d1f2b

41
gbpservice/neutron/db/migration/alembic_migrations/versions/dc99863d1f2b_add_subnet_scope_extension.py Normal file
View File

@ -0,0 +1,41 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""adds subnet scope extension support
Revision ID: dc99863d1f2b
Revises: 8c5b556b4df1
"""
# revision identifiers, used by Alembic.
revision = 'dc99863d1f2b'
down_revision = '8c5b556b4df1'
from alembic import op
import sqlalchemy as sa
from sqlalchemy import sql
def upgrade():
op.add_column('apic_aim_subnet_extensions',
sa.Column('advertised_externally', sa.Boolean,
nullable=False, server_default=sql.true()))
op.add_column('apic_aim_subnet_extensions',
sa.Column('shared_between_vrfs', sa.Boolean,
nullable=False, server_default=sql.false()))
pass
def downgrade():
pass

14
gbpservice/neutron/extensions/cisco_apic.py
View File

@ -55,6 +55,8 @@ SNAT_SUBNET_ONLY = 'apic:snat_subnet_only'
EPG_SUBNET = 'apic:epg_subnet'
NO_NAT_CIDRS = 'apic:no_nat_cidrs'
MULTI_EXT_NETS = 'apic:multi_ext_nets'
ADVERTISED_EXTERNALLY = 'apic:advertised_externally'
SHARED_BETWEEN_VRFS = 'apic:shared_between_vrfs'
BD = 'BridgeDomain'
EPG = 'EndpointGroup'
@ -410,6 +412,18 @@ EXT_SUBNET_ATTRIBUTES = {
'allow_post': True, 'allow_put': False,
'is_visible': True, 'default': False,
'convert_to': conv.convert_to_boolean,
},
ADVERTISED_EXTERNALLY: {
# Whether this subnet is visible outside of ACI or not
'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': True,
'convert_to': conv.convert_to_boolean,
},
SHARED_BETWEEN_VRFS: {
# Whether this subnet is seen across VRFs or only its own
'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': False,
'convert_to': conv.convert_to_boolean,
}
}

12
gbpservice/neutron/plugins/ml2plus/drivers/apic_aim/extension_db.py
View File

@ -167,6 +167,8 @@ class SubnetExtensionDb(model_base.BASEV2):
active_active_aap = sa.Column(sa.Boolean)
snat_subnet_only = sa.Column(sa.Boolean)
epg_subnet = sa.Column(sa.Boolean)
advertised_externally = sa.Column(sa.Boolean)
shared_between_vrfs = sa.Column(sa.Boolean)
subnet = orm.relationship(models_v2.Subnet,
backref=orm.backref(
'aim_extension_mapping', lazy='joined',
@ -556,6 +558,10 @@ class ExtensionDbMixin(object):
db_obj['snat_subnet_only'])
self._set_if_not_none(result, cisco_apic.EPG_SUBNET,
db_obj['epg_subnet'])
self._set_if_not_none(result, cisco_apic.ADVERTISED_EXTERNALLY,
db_obj['advertised_externally'])
self._set_if_not_none(result, cisco_apic.SHARED_BETWEEN_VRFS,
db_obj['shared_between_vrfs'])
return result
def set_subnet_extn_db(self, session, subnet_id, res_dict):
@ -577,6 +583,12 @@ class ExtensionDbMixin(object):
cisco_apic.SNAT_SUBNET_ONLY]
if cisco_apic.EPG_SUBNET in res_dict:
db_obj['epg_subnet'] = res_dict[cisco_apic.EPG_SUBNET]
if cisco_apic.ADVERTISED_EXTERNALLY in res_dict:
db_obj['advertised_externally'] = res_dict[
cisco_apic.ADVERTISED_EXTERNALLY]
if cisco_apic.SHARED_BETWEEN_VRFS in res_dict:
db_obj['shared_between_vrfs'] = res_dict[
cisco_apic.SHARED_BETWEEN_VRFS]
session.add(db_obj)
def get_router_extn_db(self, session, router_id):

26
gbpservice/neutron/plugins/ml2plus/drivers/apic_aim/extension_driver.py
View File

@ -278,6 +278,10 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
res_dict.get(cisco_apic.SNAT_SUBNET_ONLY, False))
result[cisco_apic.EPG_SUBNET] = (
res_dict.get(cisco_apic.EPG_SUBNET, False))
result[cisco_apic.ADVERTISED_EXTERNALLY] = (
res_dict.get(cisco_apic.ADVERTISED_EXTERNALLY, True))
result[cisco_apic.SHARED_BETWEEN_VRFS] = (
res_dict.get(cisco_apic.SHARED_BETWEEN_VRFS, False))
except Exception as e:
with excutils.save_and_reraise_exception():
if db_api.is_retriable(e):
@ -299,6 +303,10 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
res_dict.get(cisco_apic.SNAT_SUBNET_ONLY, False))
result[cisco_apic.EPG_SUBNET] = (
res_dict.get(cisco_apic.EPG_SUBNET, False))
result[cisco_apic.ADVERTISED_EXTERNALLY] = (
res_dict.get(cisco_apic.ADVERTISED_EXTERNALLY, True))
result[cisco_apic.SHARED_BETWEEN_VRFS] = (
res_dict.get(cisco_apic.SHARED_BETWEEN_VRFS, False))
except Exception as e:
with excutils.save_and_reraise_exception():
if db_api.is_retriable(e):
@ -315,14 +323,20 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
cisco_apic.SNAT_SUBNET_ONLY:
data.get(cisco_apic.SNAT_SUBNET_ONLY, False),
cisco_apic.EPG_SUBNET:
data.get(cisco_apic.EPG_SUBNET, False)}
data.get(cisco_apic.EPG_SUBNET, False),
cisco_apic.ADVERTISED_EXTERNALLY:
data.get(cisco_apic.ADVERTISED_EXTERNALLY, True),
cisco_apic.SHARED_BETWEEN_VRFS:
data.get(cisco_apic.SHARED_BETWEEN_VRFS, False)}
self.set_subnet_extn_db(plugin_context.session, result['id'],
res_dict)
result.update(res_dict)
def process_update_subnet(self, plugin_context, data, result):
if (cisco_apic.SNAT_HOST_POOL not in data and
cisco_apic.SNAT_SUBNET_ONLY not in data):
cisco_apic.SNAT_SUBNET_ONLY not in data and
cisco_apic.ADVERTISED_EXTERNALLY not in data and
cisco_apic.SHARED_BETWEEN_VRFS not in data):
return
res_dict = {}
@ -334,6 +348,14 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
res_dict.update({cisco_apic.SNAT_SUBNET_ONLY:
data[cisco_apic.SNAT_SUBNET_ONLY]})
if cisco_apic.ADVERTISED_EXTERNALLY in data:
res_dict.update({cisco_apic.ADVERTISED_EXTERNALLY:
data[cisco_apic.ADVERTISED_EXTERNALLY]})
if cisco_apic.SHARED_BETWEEN_VRFS in data:
res_dict.update({cisco_apic.SHARED_BETWEEN_VRFS:
data[cisco_apic.SHARED_BETWEEN_VRFS]})
self.set_subnet_extn_db(plugin_context.session, result['id'],
res_dict)
result.update(res_dict)

60
gbpservice/neutron/plugins/ml2plus/drivers/apic_aim/mechanism_driver.py
View File

@ -1516,6 +1516,9 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
network_id = current['network_id']
network_db = self.plugin._get_network(context._plugin_context,
network_id)
subnet_scope = self._determine_subnet_scope(
current.get(cisco_apic.ADVERTISED_EXTERNALLY, True),
current.get(cisco_apic.SHARED_BETWEEN_VRFS, False))
if network_db.external is not None and current['gateway_ip']:
l3out, ext_net, ns = self._get_aim_nat_strategy_db(session,
network_db)
@ -1543,10 +1546,12 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
cidr=current['cidr'], l3out=l3out.dn)
if current[cisco_apic.EPG_SUBNET]:
ns.create_epg_subnet(aim_ctx, l3out,
self._subnet_to_gw_ip_mask(current))
self._subnet_to_gw_ip_mask(current),
scope=subnet_scope)
else:
ns.create_subnet(aim_ctx, l3out,
self._subnet_to_gw_ip_mask(current))
self._subnet_to_gw_ip_mask(current),
scope=subnet_scope)
# Send a port update for those existing VMs because
# SNAT info has been added.
if current[cisco_apic.SNAT_HOST_POOL]:
@ -1622,6 +1627,23 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
network_db = self.plugin._get_network(context._plugin_context,
network_id)
subnet_scope = self._determine_subnet_scope(
current.get(cisco_apic.ADVERTISED_EXTERNALLY, True),
current.get(cisco_apic.SHARED_BETWEEN_VRFS, False))
if ((original[cisco_apic.ADVERTISED_EXTERNALLY] !=
current[cisco_apic.ADVERTISED_EXTERNALLY]) or
(original[cisco_apic.SHARED_BETWEEN_VRFS] !=
current[cisco_apic.SHARED_BETWEEN_VRFS])):
bd = self._get_network_bd(network_db.aim_mapping)
gw_ip = current['gateway_ip']
if current.get(cisco_apic.EPG_SUBNET, False):
epg = self._get_network_epg(network_db.aim_mapping)
sn = self._map_epg_subnet(current, gw_ip, epg)
else:
sn = self._map_subnet(current, gw_ip, bd)
self.aim.update(aim_ctx, sn, scope=subnet_scope)
# This should apply to both external and internal networks
if (current['host_routes'] != original['host_routes'] or
current['dns_nameservers'] != original['dns_nameservers']):
@ -1653,7 +1675,8 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
self._subnet_to_gw_ip_mask(original))
if current['gateway_ip']:
ns.create_subnet(aim_ctx, l3out,
self._subnet_to_gw_ip_mask(current))
self._subnet_to_gw_ip_mask(current),
scope=subnet_scope)
return
if current['name'] != original['name']:
@ -2431,13 +2454,18 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
(subnet['name'] or subnet['cidr']))
sn_ext = self.get_subnet_extn_db(session, subnet['id'])
subnet_scope = self._determine_subnet_scope(
sn_ext.get(cisco_apic.ADVERTISED_EXTERNALLY, True),
sn_ext.get(cisco_apic.SHARED_BETWEEN_VRFS, False))
if sn_ext.get(cisco_apic.EPG_SUBNET, False):
epg_sn = self._map_epg_subnet(subnet, gw_ip, epg)
epg_sn.display_name = dname
epg_sn.scope = subnet_scope
epg_sn = self.aim.create(aim_ctx, epg_sn)
else:
sn = self._map_subnet(subnet, gw_ip, bd)
sn.display_name = dname
sn.scope = subnet_scope
sn = self.aim.create(aim_ctx, sn)
# Ensure network's EPG provides/consumes router's Contract.
@ -4855,6 +4883,19 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
return self._get_aim_external_objects_db(session, network_db)
return None, None, None
def _determine_subnet_scope(self,
advertised_externally,
shared_between_vrfs):
if advertised_externally is True and shared_between_vrfs is False:
return aim_resource.Subnet.SCOPE_PUBLIC
elif advertised_externally is False and shared_between_vrfs is True:
return aim_resource.Subnet.SCOPE_SHARED
elif advertised_externally is True and shared_between_vrfs is True:
return aim_resource.Subnet.SCOPE_PUBLIC_SHARED
elif advertised_externally is False and shared_between_vrfs is False:
return aim_resource.Subnet.SCOPE_PRIVATE
return aim_resource.Subnet.SCOPE_PUBLIC
def _subnet_to_gw_ip_mask(self, subnet):
cidr = subnet['cidr'].split('/')
return aim_resource.Subnet.to_gw_ip_mask(
@ -7204,7 +7245,9 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
res_dict = {
cisco_apic.SNAT_HOST_POOL: False,
cisco_apic.ACTIVE_ACTIVE_AAP: False,
cisco_apic.EPG_SUBNET: False
cisco_apic.EPG_SUBNET: False,
cisco_apic.ADVERTISED_EXTERNALLY: True,
cisco_apic.SHARED_BETWEEN_VRFS: False
}
self.set_subnet_extn_db(mgr.actual_session, subnet_db.id, res_dict)
@ -7363,15 +7406,20 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
for subnet_db in net_db.subnets:
if not subnet_db.aim_extension_mapping:
self._missing_subnet_extension_mapping(mgr, subnet_db)
scope = self._determine_subnet_scope(
subnet_db.aim_extension_mapping.advertised_externally,
subnet_db.aim_extension_mapping.shared_between_vrfs)
if subnet_db.gateway_ip:
if subnet_db.aim_extension_mapping.epg_subnet:
ns.create_epg_subnet(
mgr.expected_aim_ctx, l3out,
self._subnet_to_gw_ip_mask(subnet_db))
self._subnet_to_gw_ip_mask(subnet_db),
scope=scope)
else:
ns.create_subnet(
mgr.expected_aim_ctx, l3out,
self._subnet_to_gw_ip_mask(subnet_db))
self._subnet_to_gw_ip_mask(subnet_db),
scope=scope)
# REVISIT: Process each AIM ExternalNetwork rather than each
# external Neutron network?

272
gbpservice/neutron/tests/unit/plugins/ml2plus/test_apic_aim.py
View File

@ -132,6 +132,8 @@ ASN = 'apic:bgp_asn'
BGP_TYPE = 'apic:bgp_type'
SNAT_SUBNET_ONLY = 'apic:snat_subnet_only'
EPG_SUBNET = 'apic:epg_subnet'
ADVERTISED_EXTERNALLY = 'apic:advertised_externally'
SHARED_BETWEEN_VRFS = 'apic:shared_between_vrfs'
def sort_if_list(attr):
@ -363,7 +365,9 @@ class ApicAimTestCase(test_address_scope.AddressScopeTestCase,
CIDR, PROV, CONS, SVI,
BGP, BGP_TYPE, ASN,
'provider:network_type',
'apic:multi_ext_nets'
'apic:multi_ext_nets',
ADVERTISED_EXTERNALLY,
SHARED_BETWEEN_VRFS
)
self.name_mapper = apic_mapper.APICNameMapper()
self.t1_aname = self.name_mapper.project(None, 't1')
@ -897,6 +901,7 @@ class TestAimMapping(ApicAimTestCase):
self.call_wrapper = CallRecordWrapper()
self.mock_ns = self.call_wrapper.setUp(
nat_strategy.DistributedNatStrategy)
self._actual_scopes = {}
self._scope_vrf_dnames = {}
super(TestAimMapping, self).setUp()
@ -965,6 +970,19 @@ class TestAimMapping(ApicAimTestCase):
self.assertIsNotNone(subnet)
return subnet
def _get_epg_subnet(self, gw_ip_mask, tenant_name, app_profile_name,
epg_name):
ctx = n_context.get_admin_context()
with db_api.CONTEXT_READER.using(ctx):
aim_ctx = aim_context.AimContext(ctx.session)
subnet = aim_resource.EPGSubnet(tenant_name=tenant_name,
app_profile_name=app_profile_name,
epg_name=epg_name,
gw_ip_mask=gw_ip_mask)
subnet = self.aim_mgr.get(aim_ctx, subnet)
self.assertIsNotNone(subnet)
return subnet
def _subnet_should_not_exist(self, gw_ip_mask, bd_name):
ctx = n_context.get_admin_context()
with db_api.CONTEXT_READER.using(ctx):
@ -1172,7 +1190,8 @@ class TestAimMapping(ApicAimTestCase):
self._epg_should_not_exist(aname)
def _check_subnet(self, subnet, net, expected_gws, unexpected_gw_ips,
scope=None, project=None):
scope=None, project=None,
expected_subnet_scope='public'):
dns = copy.copy(subnet.get(DN))
prefix_len = subnet['cidr'].split('/')[1]
@ -1191,7 +1210,7 @@ class TestAimMapping(ApicAimTestCase):
self.assertEqual(tenant_aname, aim_subnet.tenant_name)
self.assertEqual(net_aname, aim_subnet.bd_name)
self.assertEqual(gw_ip_mask, aim_subnet.gw_ip_mask)
self.assertEqual('public', aim_subnet.scope)
self.assertEqual(expected_subnet_scope, aim_subnet.scope)
display_name = ("%s-%s" %
(router['name'],
(subnet['name'] or subnet['cidr'])))
@ -1797,6 +1816,189 @@ class TestAimMapping(ApicAimTestCase):
self._sg_should_not_exist(sg_id)
self._sg_rule_should_not_exist(sg_rule['id'])
def test_subnet_scope(self):
net_resp = self._make_network(self.fmt, 'net1', True)
net = net_resp['network']
ext_net = self._make_ext_network(
'ext-net', dn=self.dn_t1_l1_n1)
l3out = aim_resource.L3Outside(tenant_name=self.t1_aname, name='l1')
self.mock_ns.reset_mock()
router = self._make_router(
self.fmt, self._tenant_id, 'router1',
external_gateway_info={'network_id': ext_net['id']})['router']
self._check_router(router)
# create public/shared subnet
subnet_ps = self._create_subnet_with_extension(
self.fmt, ext_net, '10.0.0.1', '10.0.0.0/24',
**{ADVERTISED_EXTERNALLY: 'True',
SHARED_BETWEEN_VRFS: 'True',
'gateway_ip': '10.0.0.1'})['subnet']
# check extension & scope values
subnet_ps = self._show('subnets', subnet_ps['id'])['subnet']
self.assertTrue(subnet_ps[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_ps[SHARED_BETWEEN_VRFS])
self.mock_ns.create_subnet.assert_called_once_with(
mock.ANY, l3out, '10.0.0.1/24', scope='public,shared')
aim_subnet = self._get_subnet('10.0.0.1/24', 'EXT-l1', 'prj_t1')
self.assertEqual('public,shared', aim_subnet.scope)
self.mock_ns.reset_mock()
# create shared subnet
subnet_shared = self._create_subnet_with_extension(
self.fmt, ext_net, '20.0.0.1', '20.0.0.0/24',
**{ADVERTISED_EXTERNALLY: 'False',
SHARED_BETWEEN_VRFS: 'True'})['subnet']
# check extension & scope values
subnet_shared = self._show('subnets', subnet_shared['id'])['subnet']
self.assertFalse(subnet_shared[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_shared[SHARED_BETWEEN_VRFS])
self.mock_ns.create_subnet.assert_called_once_with(
mock.ANY, l3out, '20.0.0.1/24', scope='shared')
aim_subnet = self._get_subnet('20.0.0.1/24', 'EXT-l1', 'prj_t1')
self.assertEqual('shared', aim_subnet.scope)
self.mock_ns.reset_mock()
# create private subnet
subnet_private = self._create_subnet_with_extension(
self.fmt, ext_net, '30.0.0.1', '30.0.0.0/24',
**{ADVERTISED_EXTERNALLY: 'False',
SHARED_BETWEEN_VRFS: 'False'})['subnet']
# check extension & scope values
subnet_private = self._show('subnets', subnet_private['id'])['subnet']
self.assertFalse(subnet_private[ADVERTISED_EXTERNALLY])
self.assertFalse(subnet_private[SHARED_BETWEEN_VRFS])
self.mock_ns.create_subnet.assert_called_once_with(
mock.ANY, l3out, '30.0.0.1/24', scope='private')
aim_subnet = self._get_subnet('30.0.0.1/24', 'EXT-l1', 'prj_t1')
self.assertEqual('private', aim_subnet.scope)
self.mock_ns.reset_mock()
# update it's scope to public
self._update('subnets', subnet_private['id'],
{'subnet': {ADVERTISED_EXTERNALLY: True,
SHARED_BETWEEN_VRFS: False}})
subnet_private = self._show('subnets', subnet_private['id'])['subnet']
aim_subnet = self._get_subnet('30.0.0.1/24', 'EXT-l1', 'prj_t1')
self.assertTrue(subnet_private[ADVERTISED_EXTERNALLY])
self.assertFalse(subnet_private[SHARED_BETWEEN_VRFS])
self.assertEqual('public', aim_subnet.scope)
# update private subnet scope to public,shared
self._update('subnets', subnet_private['id'],
{'subnet': {SHARED_BETWEEN_VRFS: True}})
subnet_private = self._show('subnets', subnet_private['id'])['subnet']
aim_subnet = self._get_subnet('30.0.0.1/24', 'EXT-l1', 'prj_t1')
self.assertTrue(subnet_private[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_private[SHARED_BETWEEN_VRFS])
self.assertEqual('public,shared', aim_subnet.scope)
self.mock_ns.reset_mock()
# create internal subnet
subnet_int = self._create_subnet_with_extension(
self.fmt, net, '80.0.0.1', '80.0.0.0/24',
**{ADVERTISED_EXTERNALLY: 'False',
SHARED_BETWEEN_VRFS: 'True'})['subnet']
self._router_interface_action('add', router['id'],
subnet_int['id'], None)
# check extension & scope values
subnet_int = self._show('subnets', subnet_int['id'])['subnet']
self.assertFalse(subnet_int[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_int[SHARED_BETWEEN_VRFS])
aim_subnet = self._get_subnet('80.0.0.1/24',
'net_%s' % net['id'],
'prj_%s' % net['project_id'])
self.assertEqual('shared', aim_subnet.scope)
self.mock_ns.reset_mock()
# update shared to public,shared
self._update('subnets', subnet_int['id'],
{'subnet': {ADVERTISED_EXTERNALLY: True}})
subnet_int = self._show('subnets', subnet_int['id'])['subnet']
aim_subnet = self._get_subnet('80.0.0.1/24',
'net_%s' % net['id'],
'prj_%s' % net['project_id'])
self.assertTrue(subnet_private[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_private[SHARED_BETWEEN_VRFS])
self.assertEqual('public,shared', aim_subnet.scope)
self.mock_ns.reset_mock()
# create EPG_SUBNET
subnet_epg = self._create_subnet_with_extension(
self.fmt, ext_net, '60.0.0.1', '60.0.0.0/24',
**{ADVERTISED_EXTERNALLY: 'True',
SHARED_BETWEEN_VRFS: 'True',
EPG_SUBNET: 'True',
'gateway_ip': '60.0.0.1'})['subnet']
# check extension & scope values
subnet_epg = self._show('subnets', subnet_epg['id'])['subnet']
self.assertTrue(subnet_epg[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_epg[SHARED_BETWEEN_VRFS])
self.mock_ns.create_epg_subnet.assert_called_once_with(
mock.ANY, l3out, '60.0.0.1/24', scope='public,shared')
aim_subnet = self._get_epg_subnet('60.0.0.1/24', 'prj_t1',
'OpenStack', 'EXT-l1')
self.assertEqual('public,shared', aim_subnet.scope)
self.mock_ns.reset_mock()
# update scope to shared.
self._update('subnets', subnet_epg['id'],
{'subnet': {ADVERTISED_EXTERNALLY: False}})
subnet_epg = self._show('subnets', subnet_epg['id'])['subnet']
aim_subnet = self._get_epg_subnet('60.0.0.1/24', 'prj_t1',
'OpenStack', 'EXT-l1')
self.assertFalse(subnet_epg[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_epg[SHARED_BETWEEN_VRFS])
self.assertEqual('shared', aim_subnet.scope)
self._delete('subnets', subnet_epg['id'])
# create internal EPG_SUBNET
subnet_epg = self._create_subnet_with_extension(
self.fmt, net, '60.0.0.1', '60.0.0.0/24',
**{ADVERTISED_EXTERNALLY: 'True',
SHARED_BETWEEN_VRFS: 'True',
EPG_SUBNET: 'True',
'gateway_ip': '60.0.0.1'})['subnet']
self._router_interface_action('add', router['id'],
subnet_epg['id'], None)
# check extension & scope values
subnet_epg = self._show('subnets', subnet_epg['id'])['subnet']
self.assertTrue(subnet_epg[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_epg[SHARED_BETWEEN_VRFS])
aim_subnet = self._get_epg_subnet('60.0.0.1/24',
'prj_%s' % net['project_id'],
'OpenStack',
'net_%s' % net['id'])
self.assertEqual('public,shared', aim_subnet.scope)
self.mock_ns.reset_mock()
aim_subnet = self._get_subnet('80.0.0.1/24',
'net_%s' % net['id'],
'prj_%s' % net['project_id'])
# update scope to shared.
self._update('subnets', subnet_epg['id'],
{'subnet': {ADVERTISED_EXTERNALLY: False}})
subnet_epg = self._show('subnets', subnet_epg['id'])['subnet']
aim_subnet = self._get_epg_subnet('60.0.0.1/24',
'prj_%s' % net['project_id'],
'OpenStack',
'net_%s' % net['id'])
self.assertFalse(subnet_epg[ADVERTISED_EXTERNALLY])
self.assertTrue(subnet_epg[SHARED_BETWEEN_VRFS])
self.assertEqual('shared', aim_subnet.scope)
def test_subnet_lifecycle(self):
self._test_subnet_lifecycle()
@ -7612,6 +7814,40 @@ class TestExtensionAttributes(ApicAimTestCase):
self.assertFalse(extn.get_subnet_extn_db(ctx.session,
epg_subnet['id']))
# create ADVERTISED_EXTERNALLY subnet
ae_subnet = self._create_subnet_with_extension(
self.fmt, net1, '10.1.0.1', '10.1.0.0/24',
**{ADVERTISED_EXTERNALLY: 'True'})['subnet']
ae_subnet = self._show('subnets', ae_subnet['id'])['subnet']
self.assertTrue(ae_subnet[ADVERTISED_EXTERNALLY])
ae_subnet = self._list(
'subnets', query_params=('id=%s' % ae_subnet['id']))['subnets'][0]
self.assertTrue(ae_subnet[ADVERTISED_EXTERNALLY])
# delete ADVERTISED_EXTERNALLY subnet
self._delete('subnets', ae_subnet['id'])
with db_api.CONTEXT_READER.using(ctx):
self.assertFalse(extn.get_subnet_extn_db(ctx.session,
ae_subnet['id']))
# create SHARED_BETWEEN_VRFS subnet
sbv_subnet = self._create_subnet_with_extension(
self.fmt, net1, '10.1.0.1', '10.1.0.0/24',
**{SHARED_BETWEEN_VRFS: 'True'})['subnet']
sbv_subnet = self._show('subnets', sbv_subnet['id'])['subnet']
self.assertTrue(sbv_subnet[SHARED_BETWEEN_VRFS])
sbv_subnet = self._list(
'subnets', query_params=('id=%s' % sbv_subnet['id']))['subnets'][0]
self.assertTrue(sbv_subnet[SHARED_BETWEEN_VRFS])
# delete SHARED_BETWEEN_VRFS subnet
self._delete('subnets', sbv_subnet['id'])
with db_api.CONTEXT_READER.using(ctx):
self.assertFalse(extn.get_subnet_extn_db(ctx.session,
sbv_subnet['id']))
def test_router_lifecycle(self):
ctx = n_context.get_admin_context()
extn = extn_db.ExtensionDbMixin()
@ -8322,7 +8558,7 @@ class TestExternalConnectivityBase(object):
l3out = aim_resource.L3Outside(tenant_name=self.t1_aname, name='l1')
self.mock_ns.create_subnet.assert_called_once_with(
mock.ANY, l3out, '10.0.0.1/24')
mock.ANY, l3out, '10.0.0.1/24', scope='public')
ext_sub = aim_resource.Subnet(
tenant_name=self.t1_aname, bd_name='EXT-l1',
gw_ip_mask='10.0.0.1/24')
@ -8338,7 +8574,7 @@ class TestExternalConnectivityBase(object):
self.mock_ns.delete_subnet.assert_called_once_with(
mock.ANY, l3out, '10.0.0.1/24')
self.mock_ns.create_subnet.assert_called_once_with(
mock.ANY, l3out, '10.0.0.251/24')
mock.ANY, l3out, '10.0.0.251/24', scope='public')
self._check_dn(subnet, ext_sub, 'Subnet')
self._validate()
@ -8361,7 +8597,7 @@ class TestExternalConnectivityBase(object):
subnet = self._show('subnets', subnet['id'])['subnet']
self.mock_ns.delete_subnet.assert_not_called()
self.mock_ns.create_subnet.assert_called_once_with(
mock.ANY, l3out, '10.0.0.251/24')
mock.ANY, l3out, '10.0.0.251/24', scope='public')
self._check_dn(subnet, ext_sub, 'Subnet')
self._validate()
@ -8379,7 +8615,7 @@ class TestExternalConnectivityBase(object):
l3out = aim_resource.L3Outside(tenant_name=self.t1_aname, name='l1')
self.mock_ns.create_epg_subnet.assert_called_once_with(
mock.ANY, l3out, '20.0.0.1/24')
mock.ANY, l3out, '20.0.0.1/24', scope='public')
ext_epg_sub = aim_resource.EPGSubnet(
tenant_name=self.t1_aname, app_profile_name='OpenStack',
epg_name='EXT-l1', gw_ip_mask='20.0.0.1/24')
@ -8392,6 +8628,28 @@ class TestExternalConnectivityBase(object):
self.mock_ns.delete_epg_subnet.assert_called_once_with(
mock.ANY, l3out, '20.0.0.1/24')
# create ADVERTISED_EXTERNALLY & SHARED_BETWEEN_VRFs subnet
ae_subnet = self._create_subnet_with_extension(
self.fmt, net1, '20.0.0.1', '20.0.0.0/24',
**{ADVERTISED_EXTERNALLY: 'True',
SHARED_BETWEEN_VRFS: 'True'})['subnet']
ae_subnet = self._show('subnets', ae_subnet['id'])['subnet']
l3out = aim_resource.L3Outside(tenant_name=self.t1_aname, name='l1')
self.mock_ns.create_subnet.assert_called_once_with(
mock.ANY, l3out, '20.0.0.1/24', scope='public,shared')
ext_sub = aim_resource.Subnet(
tenant_name=self.t1_aname, bd_name='EXT-l1',
gw_ip_mask='20.0.0.1/24')
self._check_dn(ae_subnet, ext_sub, 'Subnet')
self._validate()
# delete subnet
self.mock_ns.reset_mock()
self._delete('subnets', ae_subnet['id'])
self.mock_ns.delete_subnet.assert_called_once_with(
mock.ANY, l3out, '20.0.0.1/24')
def test_unmanaged_external_subnet_lifecycle(self):
net = self._make_network(self.fmt, 'net1', True)
subnet = self._make_subnet(