x
/
group-based-policy
Code Issues Proposed changes

[AIM] Add extra provided/consumed contracts to network extension 

Add apic:extra_provided_contracts and apic:extra_consumed_contracts
attributes to the cisco_apic network extension. The named contracts
are provided/consumed by the network's default EPG, in addition to any
applicable router contracts. At least one subnet on the network must
be attached as a router interface for the extra contracts to have any
effect on the network's connectivity. Attempting to specify extra
contracts for an external network or SVI network results is rejected
with an exception.

(cherry picked from commit 496f54b84a)
(cherry picked from commit 3e16a0c590)

Change-Id: I784107f4e7d7d5d39377c583bcd2163c7688eb5b
This commit is contained in:
Robert Kukura 2019-12-16 06:43:39 -05:00
parent b438fbb7e8
commit 525b9bbf24
9 changed files with 313 additions and 24 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 @@
b76dc22f2e23
f28141ea1bbf

44
gbpservice/neutron/db/migration/alembic_migrations/versions/f28141ea1bbf_extra_contracts.py Normal file
View File

@ -0,0 +1,44 @@
# 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.
"""Table for cisco_apic network extra contracts extension attributes
Revision ID: f28141ea1bbf
Revises: b76dc22f2e23
Create Date: 2019-12-20 13:08:03.603312
"""
# revision identifiers, used by Alembic.
revision = 'f28141ea1bbf'
down_revision = 'b76dc22f2e23'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.create_table(
'apic_aim_network_extra_contracts',
sa.Column('network_id', sa.String(36), nullable=False),
sa.Column('contract_name', sa.String(64), nullable=False),
sa.Column('provides', sa.Boolean, nullable=False),
sa.ForeignKeyConstraint(['network_id'], ['networks.id'],
name=
'apic_aim_network_contract_extn_fk_network',
ondelete='CASCADE'),
sa.PrimaryKeyConstraint('network_id', 'contract_name', 'provides')
)
def downgrade():
pass

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

@ -43,6 +43,8 @@ NESTED_DOMAIN_INFRA_VLAN = 'apic:nested_domain_infra_vlan'
NESTED_DOMAIN_ALLOWED_VLANS = 'apic:nested_domain_allowed_vlans'
NESTED_DOMAIN_SERVICE_VLAN = 'apic:nested_domain_service_vlan'
NESTED_DOMAIN_NODE_NETWORK_VLAN = 'apic:nested_domain_node_network_vlan'
EXTRA_PROVIDED_CONTRACTS = 'apic:extra_provided_contracts'
EXTRA_CONSUMED_CONTRACTS = 'apic:extra_consumed_contracts'
BD = 'BridgeDomain'
EPG = 'EndpointGroup'
@ -218,6 +220,18 @@ NET_ATTRIBUTES = {
},
'convert_to': convert_nested_domain_allowed_vlans,
},
EXTRA_PROVIDED_CONTRACTS: {
'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': None,
'convert_to': conv.convert_none_to_empty_list,
'validate': {'type:list_of_unique_strings': None},
},
EXTRA_CONSUMED_CONTRACTS: {
'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': None,
'convert_to': conv.convert_none_to_empty_list,
'validate': {'type:list_of_unique_strings': None},
},
}
EXT_NET_ATTRIBUTES = {

6
gbpservice/neutron/plugins/ml2plus/drivers/apic_aim/exceptions.py
View File

@ -101,3 +101,9 @@ class AAPNotAllowedOnDifferentActiveActiveAAPSubnet(exceptions.BadRequest):
message = _("Allowed address pair can not be added to this port "
"because its subnets %(subnet_ids)s active active AAP mode is "
"different than other port's subnets %(other_subnet_ids)s.")
class InvalidNetworkForExtraContracts(exceptions.BadRequest):
message = _("Cannot specify apic:extra_provided_contracts or "
"apic:extra_consumed_consumed contracts for an external or "
"SVI network.")

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

@ -84,6 +84,21 @@ class NetworkExtNestedDomainAllowedVlansDb(model_base.BASEV2):
lazy='joined', cascade='delete'))
class NetworkExtExtraContractDb(model_base.BASEV2):
__tablename__ = 'apic_aim_network_extra_contracts'
network_id = sa.Column(
sa.String(36), sa.ForeignKey('networks.id', ondelete="CASCADE"))
contract_name = sa.Column(sa.String(64), primary_key=True)
provides = sa.Column(sa.Boolean, primary_key=True)
network = orm.relationship(models_v2.Network,
backref=orm.backref(
'aim_extension_extra_contract_mapping',
uselist=True,
lazy='joined', cascade='delete'))
class SubnetExtensionDb(model_base.BASEV2):
__tablename__ = 'apic_aim_subnet_extensions'
@ -133,24 +148,33 @@ class ExtensionDbMixin(object):
NetworkExtNestedDomainAllowedVlansDb).filter(
NetworkExtNestedDomainAllowedVlansDb.network_id.in_(
network_ids)).all())
db_contracts = (session.query(NetworkExtExtraContractDb).filter(
NetworkExtExtraContractDb.network_id.in_(network_ids)).all())
cidrs_by_net_id = {}
vlans_by_net_id = {}
contracts_by_net_id = {}
for db_cidr in db_cidrs:
cidrs_by_net_id.setdefault(db_cidr.network_id, []).append(
db_cidr)
for db_vlan in db_vlans:
vlans_by_net_id.setdefault(db_vlan.network_id, []).append(
db_vlan)
for db_contract in db_contracts:
contracts_by_net_id.setdefault(db_contract.network_id, []).append(
db_contract)
result = {}
for db_obj in db_objs:
net_id = db_obj.network_id
result.setdefault(net_id, self.make_network_extn_db_conf_dict(
db_obj, cidrs_by_net_id.get(net_id, []),
vlans_by_net_id.get(net_id, [])))
vlans_by_net_id.get(net_id, []),
contracts_by_net_id.get(net_id, [])))
return result
def make_network_extn_db_conf_dict(self, ext_db, db_cidrs, db_vlans):
def make_network_extn_db_conf_dict(self, ext_db, db_cidrs, db_vlans,
db_contracts):
net_res = {}
db_obj = ext_db
if db_obj:
@ -174,6 +198,10 @@ class ExtensionDbMixin(object):
db_obj['nested_domain_node_network_vlan'])
net_res[cisco_apic.NESTED_DOMAIN_ALLOWED_VLANS] = [
c.vlan for c in db_vlans]
net_res[cisco_apic.EXTRA_PROVIDED_CONTRACTS] = [
c.contract_name for c in db_contracts if c.provides]
net_res[cisco_apic.EXTRA_CONSUMED_CONTRACTS] = [
c.contract_name for c in db_contracts if not c.provides]
if net_res.get(cisco_apic.EXTERNAL_NETWORK):
net_res[cisco_apic.EXTERNAL_CIDRS] = [c.cidr for c in db_cidrs]
return net_res
@ -229,6 +257,18 @@ class ExtensionDbMixin(object):
res_dict[cisco_apic.NESTED_DOMAIN_ALLOWED_VLANS],
network_id=network_id)
if cisco_apic.EXTRA_PROVIDED_CONTRACTS in res_dict:
self._update_list_attr(
session, NetworkExtExtraContractDb, 'contract_name',
res_dict[cisco_apic.EXTRA_PROVIDED_CONTRACTS],
network_id=network_id, provides=True)
if cisco_apic.EXTRA_CONSUMED_CONTRACTS in res_dict:
self._update_list_attr(
session, NetworkExtExtraContractDb, 'contract_name',
res_dict[cisco_apic.EXTRA_CONSUMED_CONTRACTS],
network_id=network_id, provides=False)
def get_network_ids_by_ext_net_dn(self, session, dn, lock_update=False):
query = BAKERY(lambda s: s.query(
NetworkExtensionDb.network_id))

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

@ -115,6 +115,10 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
data.get(cisco_apic.NESTED_DOMAIN_SERVICE_VLAN),
cisco_apic.NESTED_DOMAIN_NODE_NETWORK_VLAN:
data.get(cisco_apic.NESTED_DOMAIN_NODE_NETWORK_VLAN),
cisco_apic.EXTRA_PROVIDED_CONTRACTS:
data.get(cisco_apic.EXTRA_PROVIDED_CONTRACTS),
cisco_apic.EXTRA_CONSUMED_CONTRACTS:
data.get(cisco_apic.EXTRA_CONSUMED_CONTRACTS),
}
if cisco_apic.VLANS_LIST in (data.get(
cisco_apic.NESTED_DOMAIN_ALLOWED_VLANS) or {}):
@ -150,8 +154,7 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
result.update(res_dict)
def process_update_network(self, plugin_context, data, result):
# External_cidr, bgp_enable, bgp_type and bgp_asn or
# nested domain attributes could be updated
# Extension attributes that could be updated.
update_attrs = [
cisco_apic.EXTERNAL_CIDRS, cisco_apic.BGP, cisco_apic.BGP_TYPE,
cisco_apic.BGP_ASN,
@ -159,8 +162,9 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
cisco_apic.NESTED_DOMAIN_INFRA_VLAN,
cisco_apic.NESTED_DOMAIN_SERVICE_VLAN,
cisco_apic.NESTED_DOMAIN_NODE_NETWORK_VLAN,
cisco_apic.NESTED_DOMAIN_ALLOWED_VLANS]
cisco_apic.NESTED_DOMAIN_ALLOWED_VLANS,
cisco_apic.EXTRA_PROVIDED_CONTRACTS,
cisco_apic.EXTRA_CONSUMED_CONTRACTS]
if not(set(update_attrs) & set(data.keys())):
return
@ -176,7 +180,9 @@ class ApicExtensionDriver(api_plus.ExtensionDriver,
cisco_apic.NESTED_DOMAIN_NAME, cisco_apic.NESTED_DOMAIN_TYPE,
cisco_apic.NESTED_DOMAIN_INFRA_VLAN,
cisco_apic.NESTED_DOMAIN_SERVICE_VLAN,
cisco_apic.NESTED_DOMAIN_NODE_NETWORK_VLAN]
cisco_apic.NESTED_DOMAIN_NODE_NETWORK_VLAN,
cisco_apic.EXTRA_PROVIDED_CONTRACTS,
cisco_apic.EXTRA_CONSUMED_CONTRACTS]
for e_k in ext_keys:
if e_k in data:
res_dict.update({e_k: data[e_k]})

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

@ -600,7 +600,15 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
session = context._plugin_context.session
aim_ctx = aim_context.AimContext(session)
if self._is_external(current):
is_ext = self._is_external(current)
is_svi = self._is_svi(current)
if ((current[cisco_apic.EXTRA_PROVIDED_CONTRACTS] or
current[cisco_apic.EXTRA_CONSUMED_CONTRACTS]) and
(is_ext or is_svi)):
raise exceptions.InvalidNetworkForExtraContracts()
if is_ext:
l3out, ext_net, ns = self._get_aim_nat_strategy(current)
if not ext_net:
return # Unmanaged external network
@ -621,7 +629,7 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
epg = resource
elif isinstance(resource, aim_resource.VRF):
vrf = resource
elif self._is_svi(current):
elif is_svi:
l3out, ext_net, _ = self._get_aim_external_objects(current)
if ext_net:
other_nets = set(
@ -714,6 +722,10 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
epg.display_name = dname
epg.bd_name = bd.name
epg.provided_contract_names = current[
cisco_apic.EXTRA_PROVIDED_CONTRACTS]
epg.consumed_contract_names = current[
cisco_apic.EXTRA_CONSUMED_CONTRACTS]
self.aim.create(aim_ctx, epg)
self._add_network_mapping_and_notify(
@ -732,8 +744,10 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
mapping = self._get_network_mapping(session, current['id'])
is_ext = self._is_external(current)
# REVISIT: Remove is_ext from condition and add UT for
# updating external network name.
is_svi = self._is_svi(current)
# Update name if changed. REVISIT: Remove is_ext from
# condition and add UT for updating external network name.
if (not is_ext and
current['name'] != original['name']):
dname = aim_utils.sanitize_display_name(current['name'])
@ -747,6 +761,37 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
if l3out:
self.aim.update(aim_ctx, l3out, display_name=dname)
# Update extra provided/consumed contracts if changed.
curr_prov = set(current[cisco_apic.EXTRA_PROVIDED_CONTRACTS])
curr_cons = set(current[cisco_apic.EXTRA_CONSUMED_CONTRACTS])
orig_prov = set(original[cisco_apic.EXTRA_PROVIDED_CONTRACTS])
orig_cons = set(original[cisco_apic.EXTRA_CONSUMED_CONTRACTS])
if (curr_prov != orig_prov or curr_cons != orig_cons):
if is_ext or is_svi:
raise exceptions.InvalidNetworkForExtraContracts()
added_prov = curr_prov - orig_prov
removed_prov = orig_prov - curr_prov
added_cons = curr_cons - orig_cons
removed_cons = orig_cons - curr_cons
# REVISIT: AIM needs methods to atomically add/remove
# items to/from lists, as concurrent changes from router
# operations are possible.
epg = self.aim.get(aim_ctx, self._get_network_epg(mapping))
if added_prov or removed_prov:
contracts = ((set(epg.provided_contract_names) | added_prov)
- removed_prov)
self.aim.update(
aim_ctx, epg, provided_contract_names=contracts)
if added_cons or removed_cons:
contracts = ((set(epg.consumed_contract_names) | added_cons)
- removed_cons)
self.aim.update(
aim_ctx, epg, consumed_contract_names=contracts)
if is_ext:
_, ext_net, ns = self._get_aim_nat_strategy(current)
if ext_net:
@ -762,7 +807,7 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
# TODO(amitbose) Propagate name updates to AIM
else:
# BGP config is supported only for svi networks.
if not self._is_svi(current):
if not is_svi:
return
# Check for pre-existing l3out SVI.
network_db = self.plugin._get_network(context._plugin_context,
@ -979,7 +1024,7 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
# Needed because of commit
# d8c1e153f88952b7670399715c2f88f1ecf0a94a in Neutron that
# put the extension call in Pike+ *before* the precommit
# calls happen in network creation. I believe this is a but
# calls happen in network creation. I believe this is a bug
# and should be discussed with the Neutron team.
mapping = self._get_network_mapping(session, net_db.id)
if mapping:
@ -1009,14 +1054,15 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
# Needed because of commit
# d8c1e153f88952b7670399715c2f88f1ecf0a94a in Neutron that
# put the extension call in Pike+ *before* the precommit
# calls happen in network creation. I believe this is a but
# calls happen in network creation. I believe this is a bug
# and should be discussed with the Neutron team.
ext_dict = self.get_network_extn_db(session, net_db.id)
else:
ext_dict = self.make_network_extn_db_conf_dict(
net_db.aim_extension_mapping,
net_db.aim_extension_cidr_mapping,
net_db.aim_extension_domain_mapping)
net_db.aim_extension_domain_mapping,
net_db.aim_extension_extra_contract_mapping)
if cisco_apic.EXTERNAL_NETWORK in ext_dict:
dn = ext_dict.pop(cisco_apic.EXTERNAL_NETWORK)
a_ext_net = aim_resource.ExternalNetwork.from_dn(dn)
@ -5295,7 +5341,17 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
contract = self._map_router(
mgr.expected_session, router_db, True)
router_contract_names.add(contract.name)
router_contract_names = list(router_contract_names)
provided_contract_names = (
router_contract_names |
set([x.contract_name for x in
net_db.aim_extension_extra_contract_mapping
if x.provides]))
consumed_contract_names = (
router_contract_names |
set([x.contract_name for x in
net_db.aim_extension_extra_contract_mapping
if not x.provides]))
# REVISIT: Refactor to share code.
dname = aim_utils.sanitize_display_name(net_db.name)
@ -5313,8 +5369,8 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
epg.display_name = dname
epg.bd_name = bd.name
epg.policy_enforcement_pref = 'unenforced'
epg.provided_contract_names = router_contract_names
epg.consumed_contract_names = router_contract_names
epg.provided_contract_names = provided_contract_names
epg.consumed_contract_names = consumed_contract_names
epg.openstack_vmm_domain_names = []
epg.physical_domain_names = []
epg.vmm_domains = []

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

@ -875,6 +875,10 @@ class TestAimMapping(ApicAimTestCase):
aname = self.name_mapper.network(None, net['id'])
router_anames = [self.name_mapper.router(None, router['id'])
for router in routers or []]
provided_contract_names = set(
router_anames + net['apic:extra_provided_contracts'])
consumed_contract_names = set(
router_anames + net['apic:extra_consumed_contracts'])
if routers:
if vrf:
@ -945,9 +949,9 @@ class TestAimMapping(ApicAimTestCase):
self.assertEqual(aname, aim_epg.name)
self.assertEqual(net['name'], aim_epg.display_name)
self.assertEqual(aname, aim_epg.bd_name)
self.assertItemsEqual(router_anames,
self.assertItemsEqual(provided_contract_names,
aim_epg.provided_contract_names)
self.assertItemsEqual(router_anames,
self.assertItemsEqual(consumed_contract_names,
aim_epg.consumed_contract_names)
# REVISIT(rkukura): Check openstack_vmm_domain_names and
# physical_domain_names?
@ -1294,7 +1298,11 @@ class TestAimMapping(ApicAimTestCase):
def test_network_lifecycle(self):
# Test create.
net = self._make_network(self.fmt, 'net1', True)['network']
kwargs = {'apic:extra_provided_contracts': ['ep1', 'ep2'],
'apic:extra_consumed_contracts': ['ec1', 'ec2']}
net = self._make_network(
self.fmt, 'net1', True, arg_list=tuple(kwargs.keys()),
**kwargs)['network']
net_id = net['id']
self._check_network(net)
@ -1303,7 +1311,10 @@ class TestAimMapping(ApicAimTestCase):
self._check_network(net)
# Test update.
data = {'network': {'name': 'newnamefornet'}}
data = {'network':
{'name': 'newnamefornet',
'apic:extra_provided_contracts': ['ep2', 'ep3'],
'apic:extra_consumed_contracts': ['ec2', 'ec3']}}
net = self._update('networks', net_id, data)['network']
self._check_network(net)
@ -1341,6 +1352,56 @@ class TestAimMapping(ApicAimTestCase):
self.assertFalse(extn.get_network_extn_db(session, net['id']))
self._check_network_deleted(net)
def _test_invalid_network_exceptions(self, kwargs):
# Verify creating network with extra provided contracts fails.
kwargs['apic:extra_provided_contracts'] = ['ep1']
resp = self._create_network(
self.fmt, 'net', True, arg_list=tuple(kwargs.keys()), **kwargs)
result = self.deserialize(self.fmt, resp)
self.assertEqual(
'InvalidNetworkForExtraContracts',
result['NeutronError']['type'])
del kwargs['apic:extra_provided_contracts']
# Verify creating network with extra consumed contracts fails.
kwargs['apic:extra_consumed_contracts'] = ['ec1']
resp = self._create_network(
self.fmt, 'net', True, arg_list=tuple(kwargs.keys()), **kwargs)
result = self.deserialize(self.fmt, resp)
self.assertEqual(
'InvalidNetworkForExtraContracts',
result['NeutronError']['type'])
del kwargs['apic:extra_consumed_contracts']
# Create network without extra provided or consumed contracts.
net_id = self._make_network(
self.fmt, 'net', True,
arg_list=tuple(kwargs.keys()), **kwargs)['network']['id']
# Verify setting extra provided contracts on network fails.
result = self._update(
'networks', net_id,
{'network': {'apic:extra_provided_contracts': ['ep1']}},
webob.exc.HTTPBadRequest.code)
self.assertEqual(
'InvalidNetworkForExtraContracts',
result['NeutronError']['type'])
# Verify setting extra consumed contracts on network fails.
result = self._update(
'networks', net_id,
{'network': {'apic:extra_consumed_contracts': ['ec1']}},
webob.exc.HTTPBadRequest.code)
self.assertEqual(
'InvalidNetworkForExtraContracts',
result['NeutronError']['type'])
def test_external_network_exceptions(self):
self._test_invalid_network_exceptions({'router:external': True})
def test_svi_network_exceptions(self):
self._test_invalid_network_exceptions({'apic:svi': True})
def test_security_group_lifecycle(self):
# Test create
sg = self._make_security_group(self.fmt,
@ -3051,6 +3112,14 @@ class TestAimMapping(ApicAimTestCase):
gw3C = '10.0.3.3'
gw4C = '10.0.4.3'
# Add extra contracts to two of the networks.
self._update(
'networks', net2, {'network': {'apic:extra_provided_contracts':
['ep1', 'ep2']}})
self._update(
'networks', net3, {'network': {'apic:extra_consumed_contracts':
['ec1', 'ec2']}})
# Check initial state with no routing.
check_net(net1, sn1, [], [], [gw1A], t1)
check_net(net2, sn2, [], [], [gw2A, gw2B], t1)
@ -5087,6 +5156,57 @@ class TestExtensionAttributes(ApicAimTestCase):
network_id=net1['id']).all())
self.assertEqual([], db_vlans)
def test_network_with_extra_contracts_lifecycle(self):
session = db_api.get_reader_session()
extn = extn_db.ExtensionDbMixin()
# Create network with extra contracts.
provided = ['ep1', 'ep2']
consumed = ['ec1', 'ec2']
kwargs = {'apic:extra_provided_contracts': provided,
'apic:extra_consumed_contracts': consumed}
net = self._make_network(
self.fmt, 'net1', True, arg_list=tuple(kwargs.keys()),
**kwargs)['network']
net_id = net['id']
self.assertItemsEqual(provided, net['apic:extra_provided_contracts'])
self.assertItemsEqual(consumed, net['apic:extra_consumed_contracts'])
# Test show.
net = self._show('networks', net_id)['network']
self.assertItemsEqual(provided, net['apic:extra_provided_contracts'])
self.assertItemsEqual(consumed, net['apic:extra_consumed_contracts'])
# Test list.
net = self._list(
'networks', query_params=('id=%s' % net_id))['networks'][0]
self.assertItemsEqual(provided, net['apic:extra_provided_contracts'])
self.assertItemsEqual(consumed, net['apic:extra_consumed_contracts'])
# Test update extra contracts.
provided = ['ep2', 'ep3']
consumed = ['ec2', 'ec3']
net = self._update(
'networks', net_id,
{'network':
{'apic:extra_provided_contracts': provided,
'apic:extra_consumed_contracts': consumed}})['network']
self.assertItemsEqual(provided, net['apic:extra_provided_contracts'])
self.assertItemsEqual(consumed, net['apic:extra_consumed_contracts'])
# Test show after update.
net = self._show('networks', net_id)['network']
self.assertItemsEqual(provided, net['apic:extra_provided_contracts'])
self.assertItemsEqual(consumed, net['apic:extra_consumed_contracts'])
# Test delete.
self._delete('networks', net_id)
self.assertFalse(extn.get_network_extn_db(session, net_id))
db_contracts = (session.query(
extn_db.NetworkExtExtraContractDb).filter_by(
network_id=net_id).all())
self.assertEqual([], db_contracts)
def test_external_network_lifecycle(self):
session = db_api.get_reader_session()
extn = extn_db.ExtensionDbMixin()

5
gbpservice/neutron/tests/unit/services/grouppolicy/test_aim_validation.py
View File

@ -431,7 +431,10 @@ class TestNeutronMapping(AimValidationTestCase):
def test_unrouted_network(self):
# Create network.
net_resp = self._make_network(self.fmt, 'net1', True)
kwargs = {'apic:extra_provided_contracts': ['ep1', 'ep2'],
'apic:extra_consumed_contracts': ['ec1', 'ec2']}
net_resp = self._make_network(
self.fmt, 'net1', True, arg_list=tuple(kwargs.keys()), **kwargs)
net = net_resp['network']
net_id = net['id']
self._validate()