x/vmware-nsx
Code Issues Proposed changes

Merge "NEC plugin: delete old OFC ID mapping tables"

Browse Source
This commit is contained in:
Jenkins 2014-03-06 11:58:24 +00:00 committed by Gerrit Code Review
commit 547b43880f
11 changed files with 270 additions and 746 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

208
neutron/db/migration/alembic_migrations/versions/117643811bca_nec_delete_ofc_mapping.py Normal file
View File

@ -0,0 +1,208 @@
# Copyright 2014 NEC Corporation
#
# 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.
#
"""nec: delete old ofc mapping tables
Revision ID: 117643811bca
Revises: 81c553f3776c
Create Date: 2014-03-02 05:26:47.073318
"""
# revision identifiers, used by Alembic.
revision = '117643811bca'
down_revision = '81c553f3776c'
# Change to ['*'] if this migration applies to all plugins
migration_for_plugins = [
'neutron.plugins.nec.nec_plugin.NECPluginV2'
]
from alembic import op
import sqlalchemy as sa
from sqlalchemy.ext import compiler as sa_compiler
from sqlalchemy.sql import expression as sa_expr
from neutron.db import migration
# sqlalchemy does not support the expression:
# INSERT INTO <table> (<column>, ...) (SELECT ...)
# The following class is to support this expression.
# Reference: http://docs.sqlalchemy.org/en/rel_0_9/core/compiler.html
# section: "Compiling sub-elements of a custom expression construct"
class InsertFromSelect(sa_expr.Executable, sa_expr.ClauseElement):
_execution_options = (sa_expr.Executable._execution_options.
union({'autocommit': True}))
def __init__(self, insert_spec, select):
self.insert_spec = insert_spec
self.select = select
@sa_compiler.compiles(InsertFromSelect)
def visit_insert_from_select(element, compiler, **kw):
if type(element.insert_spec) == list:
columns = []
for column in element.insert_spec:
columns.append(column.name)
table = compiler.process(element.insert_spec[0].table, asfrom=True)
columns = ", ".join(columns)
sql = ("INSERT INTO %s (%s) (%s)" %
(table, columns, compiler.process(element.select)))
else:
sql = ("INSERT INTO %s (%s)" %
(compiler.process(element.insert_spec, asfrom=True),
compiler.process(element.select)))
return sql
def upgrade(active_plugins=None, options=None):
if not migration.should_run(active_plugins, migration_for_plugins):
return
# Table definitions below are only used for sqlalchemy to generate
# SQL statements, so in networks/ports tables only required field
# are declared. Note that 'quantum_id' in OFC ID mapping tables
# will be renamed in a later patch (bug 1287432).
ofctenants = sa_expr.table(
'ofctenants',
sa_expr.column('id'),
sa_expr.column('quantum_id'))
ofcnetworks = sa_expr.table(
'ofcnetworks',
sa_expr.column('id'),
sa_expr.column('quantum_id'))
ofcports = sa_expr.table(
'ofcports',
sa_expr.column('id'),
sa_expr.column('quantum_id'))
ofcfilters = sa_expr.table(
'ofcfilters',
sa_expr.column('id'),
sa_expr.column('quantum_id'))
ofctenantmappings = sa_expr.table(
'ofctenantmappings',
sa_expr.column('ofc_id'),
sa_expr.column('quantum_id'))
ofcnetworkmappings = sa_expr.table(
'ofcnetworkmappings',
sa_expr.column('ofc_id'),
sa_expr.column('quantum_id'))
ofcportmappings = sa_expr.table(
'ofcportmappings',
sa_expr.column('ofc_id'),
sa_expr.column('quantum_id'))
ofcfiltermappings = sa_expr.table(
'ofcfiltermappings',
sa_expr.column('ofc_id'),
sa_expr.column('quantum_id'))
networks = sa_expr.table(
'networks',
sa_expr.column('id'),
sa_expr.column('tenant_id'))
ports = sa_expr.table(
'ports',
sa_expr.column('id'),
sa_expr.column('network_id'))
# ofctenants -> ofctenantmappings
select_obj = sa.select([ofctenants.c.quantum_id,
op.inline_literal('/tenants/') + ofctenants.c.id])
stmt = InsertFromSelect([ofctenantmappings.c.quantum_id,
ofctenantmappings.c.ofc_id],
select_obj)
op.execute(stmt)
# ofcnetworks -> ofcnetworkmappings
select_obj = ofcnetworks.join(
networks,
ofcnetworks.c.quantum_id == networks.c.id)
select_obj = select_obj.join(
ofctenantmappings,
ofctenantmappings.c.quantum_id == networks.c.tenant_id)
select_obj = sa.select(
[ofcnetworks.c.quantum_id,
(ofctenantmappings.c.ofc_id +
op.inline_literal('/networks/') + ofcnetworks.c.id)],
from_obj=select_obj)
stmt = InsertFromSelect([ofcnetworkmappings.c.quantum_id,
ofcnetworkmappings.c.ofc_id],
select_obj)
op.execute(stmt)
# ofcports -> ofcportmappings
select_obj = ofcports.join(ports, ofcports.c.quantum_id == ports.c.id)
select_obj = select_obj.join(
ofcnetworkmappings,
ofcnetworkmappings.c.quantum_id == ports.c.network_id)
select_obj = sa.select(
[ofcports.c.quantum_id,
(ofcnetworkmappings.c.ofc_id +
op.inline_literal('/ports/') + ofcports.c.id)],
from_obj=select_obj)
stmt = InsertFromSelect([ofcportmappings.c.quantum_id,
ofcportmappings.c.ofc_id],
select_obj)
op.execute(stmt)
# ofcfilters -> ofcfiltermappings
select_obj = sa.select([ofcfilters.c.quantum_id,
op.inline_literal('/filters/') + ofcfilters.c.id])
stmt = InsertFromSelect([ofcfiltermappings.c.quantum_id,
ofcfiltermappings.c.ofc_id],
select_obj)
op.execute(stmt)
# drop old mapping tables
op.drop_table('ofctenants')
op.drop_table('ofcnetworks')
op.drop_table('ofcports')
op.drop_table('ofcfilters')
def downgrade(active_plugins=None, options=None):
if not migration.should_run(active_plugins, migration_for_plugins):
return
op.create_table(
'ofctenants',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('quantum_id', sa.String(length=36), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table(
'ofcnetworks',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('quantum_id', sa.String(length=36), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table(
'ofcports',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('quantum_id', sa.String(length=36), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table(
'ofcfilters',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('quantum_id', sa.String(length=36), nullable=False),
sa.PrimaryKeyConstraint('id')
)

105
neutron/plugins/nec/db/api.py
View File

@ -41,74 +41,56 @@ resource_map = {'ofc_tenant': nmodels.OFCTenantMapping,
'ofc_router': nmodels.OFCRouterMapping,
'ofc_packet_filter': nmodels.OFCFilterMapping}
old_resource_map = {'ofc_tenant': nmodels.OFCTenant,
'ofc_network': nmodels.OFCNetwork,
'ofc_port': nmodels.OFCPort,
'ofc_packet_filter': nmodels.OFCFilter}
# utitlity methods
def _get_resource_model(resource, old_style):
if old_style:
# NOTE: Some new resources are not defined in old_resource_map.
# In such case None is returned.
return old_resource_map.get(resource)
else:
return resource_map[resource]
def _get_resource_model(resource):
return resource_map[resource]
def clear_db(base=model_base.BASEV2):
db.clear_db(base)
def get_ofc_item(session, resource, neutron_id, old_style=False):
model = _get_resource_model(resource, old_style)
def get_ofc_item(session, resource, neutron_id):
model = _get_resource_model(resource)
if not model:
return None
return
try:
return session.query(model).filter_by(quantum_id=neutron_id).one()
except sa.orm.exc.NoResultFound:
return None
return
def get_ofc_id(session, resource, neutron_id, old_style=False):
ofc_item = get_ofc_item(session, resource, neutron_id, old_style)
def get_ofc_id(session, resource, neutron_id):
ofc_item = get_ofc_item(session, resource, neutron_id)
if ofc_item:
if old_style:
return ofc_item.id
else:
return ofc_item.ofc_id
return ofc_item.ofc_id
else:
return None
raise nexc.OFCMappingNotFound(resource=resource,
neutron_id=neutron_id)
def exists_ofc_item(session, resource, neutron_id, old_style=False):
if get_ofc_item(session, resource, neutron_id, old_style):
def exists_ofc_item(session, resource, neutron_id):
if get_ofc_item(session, resource, neutron_id):
return True
else:
return False
def find_ofc_item(session, resource, ofc_id, old_style=False):
def find_ofc_item(session, resource, ofc_id):
try:
model = _get_resource_model(resource, old_style)
if old_style:
params = dict(id=ofc_id)
else:
params = dict(ofc_id=ofc_id)
model = _get_resource_model(resource)
params = dict(ofc_id=ofc_id)
return (session.query(model).filter_by(**params).one())
except sa.orm.exc.NoResultFound:
return None
def add_ofc_item(session, resource, neutron_id, ofc_id, old_style=False):
def add_ofc_item(session, resource, neutron_id, ofc_id):
try:
model = _get_resource_model(resource, old_style)
if old_style:
params = dict(quantum_id=neutron_id, id=ofc_id)
else:
params = dict(quantum_id=neutron_id, ofc_id=ofc_id)
model = _get_resource_model(resource)
params = dict(quantum_id=neutron_id, ofc_id=ofc_id)
item = model(**params)
with session.begin(subtransactions=True):
session.add(item)
@ -119,59 +101,20 @@ def add_ofc_item(session, resource, neutron_id, ofc_id, old_style=False):
return item
def del_ofc_item(session, resource, neutron_id, old_style=False,
warning=True):
def del_ofc_item(session, resource, neutron_id):
try:
model = _get_resource_model(resource, old_style)
model = _get_resource_model(resource)
with session.begin(subtransactions=True):
item = session.query(model).filter_by(quantum_id=neutron_id).one()
session.delete(item)
return True
except sa.orm.exc.NoResultFound:
if warning:
LOG.warning(_("_del_ofc_item(): NotFound item "
"(model=%(model)s, id=%(id)s) "),
{'model': model, 'id': neutron_id})
LOG.warning(_("del_ofc_item(): NotFound item "
"(resource=%(resource)s, id=%(id)s) "),
{'resource': resource, 'id': neutron_id})
return False
def get_ofc_id_lookup_both(session, resource, neutron_id):
ofc_id = get_ofc_id(session, resource, neutron_id)
# Lookup old style of OFC mapping table
if not ofc_id:
ofc_id = get_ofc_id(session, resource, neutron_id,
old_style=True)
if not ofc_id:
raise nexc.OFCMappingNotFound(resource=resource,
neutron_id=neutron_id)
return ofc_id
def exists_ofc_item_lookup_both(session, resource, neutron_id):
if exists_ofc_item(session, resource, neutron_id):
return True
# Check old style of OFC mapping table
if exists_ofc_item(session, resource, neutron_id,
old_style=True):
return True
return False
def del_ofc_item_lookup_both(session, resource, neutron_id):
# Delete the mapping from new style of OFC mapping table
if del_ofc_item(session, resource, neutron_id,
old_style=False, warning=False):
return
# Delete old style of OFC mapping table
if del_ofc_item(session, resource, neutron_id,
old_style=True, warning=False):
return
# The specified resource not found
LOG.warning(_("_del_ofc_item(): NotFound item "
"(resource=%(resource)s, id=%(id)s) "),
{'resource': resource, 'id': neutron_id})
def get_portinfo(session, id):
try:
return (session.query(nmodels.PortInfo).

24
neutron/plugins/nec/db/models.py
View File

@ -55,30 +55,6 @@ class OFCFilterMapping(model_base.BASEV2, NeutronId, OFCId):
"""Represents a Filter on OpenFlow Network/Controller."""
"""Old mapping tables."""
class HasNeutronId(object):
"""Logical ID on Quantum."""
quantum_id = sa.Column(sa.String(36), nullable=False)
class OFCTenant(model_base.BASEV2, models_v2.HasId, HasNeutronId):
"""Represents a Tenant on OpenFlow Network/Controller."""
class OFCNetwork(model_base.BASEV2, models_v2.HasId, HasNeutronId):
"""Represents a Network on OpenFlow Network/Controller."""
class OFCPort(model_base.BASEV2, models_v2.HasId, HasNeutronId):
"""Represents a Port on OpenFlow Network/Controller."""
class OFCFilter(model_base.BASEV2, models_v2.HasId, HasNeutronId):
"""Represents a Filter on OpenFlow Network/Controller."""
class PortInfo(model_base.BASEV2):
"""Represents a Virtual Interface."""
id = sa.Column(sa.String(36),

30
neutron/plugins/nec/drivers/pfc.py
View File

@ -27,7 +27,6 @@ from neutron.common import exceptions as qexc
from neutron.common import log as call_log
from neutron import manager
from neutron.plugins.nec.common import ofc_client
from neutron.plugins.nec.db import api as ndb
from neutron.plugins.nec.extensions import packetfilter as ext_pf
from neutron.plugins.nec import ofc_driver_base
@ -143,35 +142,6 @@ class PFCDriverBase(ofc_driver_base.OFCDriverBase):
def delete_port(self, ofc_port_id):
return self.client.delete(ofc_port_id)
def convert_ofc_tenant_id(self, context, ofc_tenant_id):
# If ofc_tenant_id starts with '/', it is already new-style
if ofc_tenant_id[0] == '/':
return ofc_tenant_id
return '/tenants/%s' % ofc_tenant_id
def convert_ofc_network_id(self, context, ofc_network_id, tenant_id):
# If ofc_network_id starts with '/', it is already new-style
if ofc_network_id[0] == '/':
return ofc_network_id
ofc_tenant_id = ndb.get_ofc_id_lookup_both(
context.session, 'ofc_tenant', tenant_id)
ofc_tenant_id = self.convert_ofc_tenant_id(context, ofc_tenant_id)
params = dict(tenant=ofc_tenant_id, network=ofc_network_id)
return '%(tenant)s/networks/%(network)s' % params
def convert_ofc_port_id(self, context, ofc_port_id, tenant_id, network_id):
# If ofc_port_id starts with '/', it is already new-style
if ofc_port_id[0] == '/':
return ofc_port_id
ofc_network_id = ndb.get_ofc_id_lookup_both(
context.session, 'ofc_network', network_id)
ofc_network_id = self.convert_ofc_network_id(
context, ofc_network_id, tenant_id)
params = dict(network=ofc_network_id, port=ofc_port_id)
return '%(network)s/ports/%(port)s' % params
class PFCFilterDriverMixin(object):
"""PFC PacketFilter Driver Mixin."""

60
neutron/plugins/nec/drivers/trema.py
View File

@ -18,7 +18,6 @@
from neutron.openstack.common import uuidutils
from neutron.plugins.nec.common import ofc_client
from neutron.plugins.nec.db import api as ndb
from neutron.plugins.nec import ofc_driver_base
@ -61,20 +60,6 @@ class TremaDriverBase(ofc_driver_base.OFCDriverBase):
def delete_network(self, ofc_network_id):
return self.client.delete(ofc_network_id)
def convert_ofc_tenant_id(self, context, ofc_tenant_id):
# If ofc_network_id starts with '/', it is already new-style
if ofc_tenant_id[0] == '/':
return ofc_tenant_id
return self._get_tenant_id(ofc_tenant_id)
def convert_ofc_network_id(self, context, ofc_network_id, tenant_id):
# If ofc_network_id starts with '/', it is already new-style
if ofc_network_id[0] == '/':
return ofc_network_id
# Trema sliceable switch does not use tenant_id,
# so we can convert ofc_network_id from old id only
return self.network_path % ofc_network_id
class TremaFilterDriverMixin(object):
"""Trema (Sliceable Switch) PacketFilter Driver Mixin."""
@ -170,12 +155,6 @@ class TremaFilterDriverMixin(object):
def delete_filter(self, ofc_filter_id):
return self.client.delete(ofc_filter_id)
def convert_ofc_filter_id(self, context, ofc_filter_id):
# If ofc_filter_id starts with '/', it is already new-style
if ofc_filter_id[0] == '/':
return ofc_filter_id
return self.filter_path % ofc_filter_id
class TremaPortBaseDriver(TremaDriverBase, TremaFilterDriverMixin):
"""Trema (Sliceable Switch) Driver for port base binding.
@ -201,19 +180,6 @@ class TremaPortBaseDriver(TremaDriverBase, TremaFilterDriverMixin):
def delete_port(self, ofc_port_id):
return self.client.delete(ofc_port_id)
def convert_ofc_port_id(self, context, ofc_port_id,
tenant_id, network_id):
# If ofc_port_id starts with '/', it is already new-style
if ofc_port_id[0] == '/':
return ofc_port_id
ofc_network_id = ndb.get_ofc_id_lookup_both(
context.session, 'ofc_network', network_id)
ofc_network_id = self.convert_ofc_network_id(
context, ofc_network_id, tenant_id)
return self.port_path % {'network': ofc_network_id,
'port': ofc_port_id}
class TremaPortMACBaseDriver(TremaDriverBase, TremaFilterDriverMixin):
"""Trema (Sliceable Switch) Driver for port-mac base binding.
@ -257,20 +223,6 @@ class TremaPortMACBaseDriver(TremaDriverBase, TremaFilterDriverMixin):
def delete_port(self, ofc_port_id):
return self.client.delete(ofc_port_id)
def convert_ofc_port_id(self, context, ofc_port_id, tenant_id, network_id):
# If ofc_port_id starts with '/', it is already new-style
if ofc_port_id[0] == '/':
return ofc_port_id
ofc_network_id = ndb.get_ofc_id_lookup_both(
context.session, 'ofc_network', network_id)
ofc_network_id = self.convert_ofc_network_id(
context, ofc_network_id, tenant_id)
dummy_port_id = 'dummy-%s' % ofc_port_id
return self.attachment_path % {'network': ofc_network_id,
'port': dummy_port_id,
'attachment': ofc_port_id}
class TremaMACBaseDriver(TremaDriverBase):
"""Trema (Sliceable Switch) Driver for mac base binding.
@ -296,15 +248,3 @@ class TremaMACBaseDriver(TremaDriverBase):
def delete_port(self, ofc_port_id):
return self.client.delete(ofc_port_id)
def convert_ofc_port_id(self, context, ofc_port_id, tenant_id, network_id):
# If ofc_port_id starts with '/', it is already new-style
if ofc_port_id[0] == '/':
return ofc_port_id
ofc_network_id = ndb.get_ofc_id_lookup_both(
context.session, 'ofc_network', network_id)
ofc_network_id = self.convert_ofc_network_id(
context, ofc_network_id, tenant_id)
return self.attachment_path % {'network': ofc_network_id,
'attachment': ofc_port_id}

32
neutron/plugins/nec/ofc_driver_base.py
View File

@ -104,35 +104,3 @@ class OFCDriverBase(object):
:raises: neutron.plugin.nec.common.exceptions.OFCException
"""
pass
@abstractmethod
def convert_ofc_tenant_id(self, context, ofc_tenant_id):
"""Convert old-style ofc tenand id to new-style one.
:param context: neutron context object
:param ofc_tenant_id: ofc_tenant_id to be converted
"""
pass
@abstractmethod
def convert_ofc_network_id(self, context, ofc_network_id,
tenant_id):
"""Convert old-style ofc network id to new-style one.
:param context: neutron context object
:param ofc_network_id: ofc_network_id to be converted
:param tenant_id: neutron tenant_id of the network
"""
pass
@abstractmethod
def convert_ofc_port_id(self, context, ofc_port_id,
tenant_id, network_id):
"""Convert old-style ofc port id to new-style one.
:param context: neutron context object
:param ofc_port_id: ofc_port_id to be converted
:param tenant_id: neutron tenant_id of the port
:param network_id: neutron network_id of the port
"""
pass

27
neutron/plugins/nec/ofc_manager.py
View File

@ -44,19 +44,17 @@ class OFCManager(object):
self.plugin = plugin
def _get_ofc_id(self, context, resource, neutron_id):
return ndb.get_ofc_id_lookup_both(context.session,
resource, neutron_id)
return ndb.get_ofc_id(context.session, resource, neutron_id)
def _exists_ofc_item(self, context, resource, neutron_id):
return ndb.exists_ofc_item_lookup_both(context.session,
resource, neutron_id)
return ndb.exists_ofc_item(context.session, resource, neutron_id)
def _add_ofc_item(self, context, resource, neutron_id, ofc_id):
# Ensure a new item is added to the new mapping table
ndb.add_ofc_item(context.session, resource, neutron_id, ofc_id)
def _del_ofc_item(self, context, resource, neutron_id):
ndb.del_ofc_item_lookup_both(context.session, resource, neutron_id)
ndb.del_ofc_item(context.session, resource, neutron_id)
def ensure_ofc_tenant(self, context, tenant_id):
if not self.exists_ofc_tenant(context, tenant_id):
@ -72,18 +70,12 @@ class OFCManager(object):
def delete_ofc_tenant(self, context, tenant_id):
ofc_tenant_id = self._get_ofc_id(context, "ofc_tenant", tenant_id)
ofc_tenant_id = self.driver.convert_ofc_tenant_id(
context, ofc_tenant_id)
self.driver.delete_tenant(ofc_tenant_id)
self._del_ofc_item(context, "ofc_tenant", tenant_id)
def create_ofc_network(self, context, tenant_id, network_id,
network_name=None):
ofc_tenant_id = self._get_ofc_id(context, "ofc_tenant", tenant_id)
ofc_tenant_id = self.driver.convert_ofc_tenant_id(
context, ofc_tenant_id)
desc = "ID=%s Name=%s at Neutron." % (network_id, network_name)
ofc_net_id = self.driver.create_network(ofc_tenant_id, desc,
network_id)
@ -94,16 +86,12 @@ class OFCManager(object):
def delete_ofc_network(self, context, network_id, network):
ofc_net_id = self._get_ofc_id(context, "ofc_network", network_id)
ofc_net_id = self.driver.convert_ofc_network_id(
context, ofc_net_id, network['tenant_id'])
self.driver.delete_network(ofc_net_id)
self._del_ofc_item(context, "ofc_network", network_id)
def create_ofc_port(self, context, port_id, port):
ofc_net_id = self._get_ofc_id(context, "ofc_network",
port['network_id'])
ofc_net_id = self.driver.convert_ofc_network_id(
context, ofc_net_id, port['tenant_id'])
portinfo = ndb.get_portinfo(context.session, port_id)
if not portinfo:
raise nexc.PortInfoNotFound(id=port_id)
@ -124,16 +112,12 @@ class OFCManager(object):
def delete_ofc_port(self, context, port_id, port):
ofc_port_id = self._get_ofc_id(context, "ofc_port", port_id)
ofc_port_id = self.driver.convert_ofc_port_id(
context, ofc_port_id, port['tenant_id'], port['network_id'])
self.driver.delete_port(ofc_port_id)
self._del_ofc_item(context, "ofc_port", port_id)
def create_ofc_packet_filter(self, context, filter_id, filter_dict):
ofc_net_id = self._get_ofc_id(context, "ofc_network",
filter_dict['network_id'])
ofc_net_id = self.driver.convert_ofc_network_id(
context, ofc_net_id, filter_dict['tenant_id'])
in_port_id = filter_dict.get('in_port')
portinfo = None
if in_port_id:
@ -159,16 +143,11 @@ class OFCManager(object):
def delete_ofc_packet_filter(self, context, filter_id):
ofc_pf_id = self._get_ofc_id(context, "ofc_packet_filter", filter_id)
ofc_pf_id = self.driver.convert_ofc_filter_id(context, ofc_pf_id)
self.driver.delete_filter(ofc_pf_id)
self._del_ofc_item(context, "ofc_packet_filter", filter_id)
def create_ofc_router(self, context, tenant_id, router_id, name=None):
ofc_tenant_id = self._get_ofc_id(context, "ofc_tenant", tenant_id)
ofc_tenant_id = self.driver.convert_ofc_tenant_id(
context, ofc_tenant_id)
desc = "ID=%s Name=%s at Neutron." % (router_id, name)
ofc_router_id = self.driver.create_router(ofc_tenant_id, router_id,
desc)

171
neutron/tests/unit/nec/test_db.py
View File

@ -55,73 +55,79 @@ class NECPluginV2DBTestBase(test_nec_plugin.NecPluginV2TestCase):
yield params
class NECPluginV2DBTest(NECPluginV2DBTestBase):
class NECPluginV2DBOfcMappingTest(NECPluginV2DBTestBase):
def testa_add_ofc_item(self):
def test_add_ofc_item(self):
"""test add OFC item."""
o, q, n = self.get_ofc_item_random_params()
tenant = ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
self.assertEqual(tenant.ofc_id, o)
self.assertEqual(tenant.quantum_id, q)
def test_add_ofc_item_duplicate_entry(self):
o, q, n = self.get_ofc_item_random_params()
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
self.assertRaises(nexc.NECDBException,
ndb.add_ofc_item,
self.session, 'ofc_tenant', q, o)
def testb_get_ofc_item(self):
"""test get OFC item."""
def test_get_ofc_item(self):
o, q, n = self.get_ofc_item_random_params()
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
tenant = ndb.get_ofc_item(self.session, 'ofc_tenant', q)
self.assertEqual(tenant.ofc_id, o)
self.assertEqual(tenant.quantum_id, q)
tenant_none = ndb.get_ofc_item(self.session, 'ofc_tenant', n)
self.assertIsNone(tenant_none)
def test_get_ofc_item_for_nonexisting_entry(self):
self.assertIsNone(
ndb.get_ofc_item(self.session, 'ofc_tenant', 'non-exist-id'))
def testb_get_ofc_id(self):
"""test get OFC d."""
def test_get_ofc_id(self):
o, q, n = self.get_ofc_item_random_params()
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
tenant_id = ndb.get_ofc_id(self.session, 'ofc_tenant', q)
self.assertEqual(tenant_id, o)
tenant_none = ndb.get_ofc_item(self.session, 'ofc_tenant', n)
self.assertIsNone(tenant_none)
def test_get_ofc_id_for_nonexisting_entry(self):
self.assertRaises(nexc.OFCMappingNotFound,
ndb.get_ofc_id,
self.session, 'ofc_tenant', 'non-exist-id')
def testb_exists_ofc_item(self):
"""test get OFC d."""
def test_exists_ofc_item(self):
o, q, n = self.get_ofc_item_random_params()
self.assertFalse(ndb.exists_ofc_item(self.session, 'ofc_tenant', q))
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
ret = ndb.exists_ofc_item(self.session, 'ofc_tenant', q)
self.assertTrue(ret)
self.assertTrue(ndb.exists_ofc_item(self.session, 'ofc_tenant', q))
tenant_none = ndb.get_ofc_item(self.session, 'ofc_tenant', n)
self.assertIsNone(tenant_none)
ndb.del_ofc_item(self.session, 'ofc_tenant', q)
self.assertFalse(ndb.exists_ofc_item(self.session, 'ofc_tenant', q))
def testc_find_ofc_item(self):
"""test find OFC item."""
def test_find_ofc_item(self):
o, q, n = self.get_ofc_item_random_params()
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
tenant = ndb.find_ofc_item(self.session, 'ofc_tenant', o)
self.assertEqual(tenant.ofc_id, o)
self.assertEqual(tenant.quantum_id, q)
tenant_none = ndb.find_ofc_item(self.session, 'ofc_tenant', n)
self.assertIsNone(tenant_none)
def test_find_ofc_item_for_nonexisting_entry(self):
self.assertIsNone(
ndb.find_ofc_item(self.session, 'ofc_tenant', 'non-existi-id'))
def testc_del_ofc_item(self):
"""test delete OFC item."""
def test_del_ofc_item(self):
o, q, n = self.get_ofc_item_random_params()
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
ndb.del_ofc_item(self.session, 'ofc_tenant', q)
self.assertTrue(ndb.del_ofc_item(self.session, 'ofc_tenant', q))
tenant_none = ndb.get_ofc_item(self.session,
'ofc_tenant', q)
self.assertIsNone(tenant_none)
tenant_none = ndb.find_ofc_item(self.session,
'ofc_tenant', o)
self.assertIsNone(tenant_none)
self.assertIsNone(ndb.get_ofc_item(self.session, 'ofc_tenant', q))
self.assertIsNone(ndb.find_ofc_item(self.session, 'ofc_tenant', o))
def test_del_ofc_item_for_nonexisting_entry(self):
self.assertFalse(
ndb.del_ofc_item(self.session, 'ofc_tenant', 'non-existi-id'))
class NECPluginV2DBPortInfoTest(NECPluginV2DBTestBase):
def _compare_portinfo(self, portinfo, expected):
self.assertEqual(portinfo.id, expected['port_id'])
@ -168,110 +174,3 @@ class NECPluginV2DBTest(NECPluginV2DBTestBase):
ndb.del_portinfo(self.session, params['port_id'])
portinfo_none = ndb.get_portinfo(self.session, params['port_id'])
self.assertIsNone(portinfo_none)
class NECPluginV2DBOldMappingTest(NECPluginV2DBTestBase):
"""Test related to old ID mapping."""
# Mapping Table mode
OLD = True
NEW = False
def test_add_ofc_item_new(self):
o, q, n = self.get_ofc_item_random_params()
ret = ndb.add_ofc_item(self.session, 'ofc_tenant', q, o, self.NEW)
self.assertEqual(ret.ofc_id, o)
self.assertEqual(ret.quantum_id, q)
ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, self.NEW)
self.assertEqual(ret.ofc_id, o)
self.assertEqual(ret.quantum_id, q)
ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, self.OLD)
self.assertIsNone(ret)
def test_add_ofc_item_old(self):
o, q, n = self.get_ofc_item_random_params()
ret = ndb.add_ofc_item(self.session, 'ofc_tenant', q, o, self.OLD)
self.assertEqual(ret.id, o)
self.assertEqual(ret.quantum_id, q)
ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, self.NEW)
self.assertIsNone(ret)
ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, self.OLD)
self.assertEqual(ret.id, o)
self.assertEqual(ret.quantum_id, q)
def _check_new_old_item(self, method, q_id, exp_new, exp_old):
ret = method(self.session, 'ofc_tenant', q_id, self.NEW)
self.assertEqual(ret, exp_new)
ret = method(self.session, 'ofc_tenant', q_id, self.OLD)
self.assertEqual(ret, exp_old)
def test_get_ofc_id_new(self):
o, q, n = self.get_ofc_item_random_params()
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o, self.NEW)
self._check_new_old_item(ndb.get_ofc_id, q, o, None)
ret = ndb.get_ofc_id_lookup_both(self.session, 'ofc_tenant', q)
self.assertEqual(ret, o)
def test_get_ofc_id_old(self):
o, q, n = self.get_ofc_item_random_params()
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o, self.OLD)
self._check_new_old_item(ndb.get_ofc_id, q, None, o)
ret = ndb.get_ofc_id_lookup_both(self.session, 'ofc_tenant', q)
self.assertEqual(ret, o)
def _check_exists_ofc_item(self, mode, exp_new, exp_old):
o, q, n = self.get_ofc_item_random_params()
self._check_new_old_item(ndb.exists_ofc_item, q, False, False)
self.assertFalse(ndb.exists_ofc_item_lookup_both(
self.session, 'ofc_tenant', q))
ndb.add_ofc_item(self.session, 'ofc_tenant', q, o, mode)
self._check_new_old_item(ndb.exists_ofc_item, q, exp_new, exp_old)
self.assertTrue(ndb.exists_ofc_item_lookup_both(
self.session, 'ofc_tenant', q))
ndb.del_ofc_item(self.session, 'ofc_tenant', q, mode)
self._check_new_old_item(ndb.exists_ofc_item, q, False, False)
self.assertFalse(ndb.exists_ofc_item_lookup_both(
self.session, 'ofc_tenant', q))
def test_exists_ofc_item_new(self):
self._check_exists_ofc_item(self.NEW, True, False)
def test_exists_ofc_item_old(self):
self._check_exists_ofc_item(self.OLD, False, True)
def _check_delete_ofc_item(self, mode, detect_mode=False):
o, q, n = self.get_ofc_item_random_params()
ret = ndb.add_ofc_item(self.session, 'ofc_tenant', q, o, mode)
ofc_id = ret.ofc_id if mode == self.NEW else ret.id
self.assertEqual(ofc_id, o)
self.assertEqual(ret.quantum_id, q)
ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, mode)
ofc_id = ret.ofc_id if mode == self.NEW else ret.id
self.assertEqual(ofc_id, o)
self.assertEqual(ret.quantum_id, q)
if detect_mode:
ndb.del_ofc_item_lookup_both(self.session, 'ofc_tenant', q)
else:
ndb.del_ofc_item(self.session, 'ofc_tenant', q, mode)
ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, self.NEW)
self.assertIsNone(ret)
ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, self.OLD)
self.assertIsNone(ret)
def test_delete_ofc_item_new(self):
self._check_delete_ofc_item(self.NEW)
def test_delete_ofc_item_old(self):
self._check_delete_ofc_item(self.OLD)
def test_delete_ofc_item_with_auto_detect_new(self):
self._check_delete_ofc_item(self.NEW, detect_mode=True)
def test_delete_ofc_item_old_auto_detect_new(self):
self._check_delete_ofc_item(self.OLD, detect_mode=True)

105
neutron/tests/unit/nec/test_ofc_manager.py
View File

@ -296,108 +296,3 @@ class OFCManagerRouterTest(OFCManagerTestBase):
routes = []
self.ofc.update_ofc_router_route(self.ctx, r, routes)
self.assertEqual(len(self.ofc.driver.ofc_router_route_dict), 0)
class OFCManagerTestWithOldMapping(OFCManagerTestBase):
def setUp(self):
super(OFCManagerTestWithOldMapping, self).setUp()
# NOTE(amotoki): In OldMapping tests, DB entries are directly modified
# to create a case where the old mapping tables are used intentionally.
self.ofc.driver.disable_autocheck()
def test_exists_ofc_tenant(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_tenant(self.ctx, t))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_tenant', t, ofc_t, old_style=True)
self.assertTrue(self.ofc.exists_ofc_tenant(self.ctx, t))
def test_delete_ofc_tenant(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_tenant(self.ctx, t))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_tenant', t, ofc_t, old_style=True)
self.assertTrue(self.ofc.exists_ofc_tenant(self.ctx, t))
self.ofc.delete_ofc_tenant(self.ctx, t)
self.assertFalse(self.ofc.exists_ofc_tenant(self.ctx, t))
def test_exists_ofc_network(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_network(self.ctx, n))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_network', n, ofc_n, old_style=True)
self.assertTrue(self.ofc.exists_ofc_network(self.ctx, n))
def test_delete_ofc_network(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_network(self.ctx, n))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_network', n, ofc_n, old_style=True)
self.assertTrue(self.ofc.exists_ofc_network(self.ctx, n))
net = {'tenant_id': t}
self.ofc.delete_ofc_network(self.ctx, n, net)
self.assertFalse(self.ofc.exists_ofc_network(self.ctx, n))
def test_exists_ofc_port(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_port(self.ctx, p))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_port', p, ofc_p, old_style=True)
self.assertTrue(self.ofc.exists_ofc_port(self.ctx, p))
def test_delete_ofc_port(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_port(self.ctx, p))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_port', p, ofc_p, old_style=True)
self.assertTrue(self.ofc.exists_ofc_port(self.ctx, p))
port = {'tenant_id': t, 'network_id': n}
self.ofc.delete_ofc_port(self.ctx, p, port)
self.assertFalse(self.ofc.exists_ofc_port(self.ctx, p))
def test_exists_ofc_packet_filter(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_packet_filter(self.ctx, f))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_packet_filter', f, ofc_f,
old_style=True)
self.assertTrue(self.ofc.exists_ofc_packet_filter(self.ctx, f))
def test_delete_ofc_packet_filter(self):
t, n, p, f, none = self.get_random_params()
ofc_t, ofc_n, ofc_p, ofc_f, ofc_none = self.get_random_params()
self.assertFalse(self.ofc.exists_ofc_packet_filter(self.ctx, f))
session = self.ctx.session
ndb.add_ofc_item(session, 'ofc_packet_filter', f, ofc_f,
old_style=True)
self.assertTrue(self.ofc.exists_ofc_packet_filter(self.ctx, f))
self.ofc.delete_ofc_packet_filter(self.ctx, f)
self.assertFalse(self.ofc.exists_ofc_packet_filter(self.ctx, f))

114
neutron/tests/unit/nec/test_pfc_driver.py
View File

@ -25,7 +25,6 @@ import netaddr
from neutron.common import constants
from neutron.openstack.common import uuidutils
from neutron.plugins.nec.common import ofc_client as ofc
from neutron.plugins.nec.db import api as ndb
from neutron.plugins.nec.db import models as nmodels
from neutron.plugins.nec import drivers
from neutron.plugins.nec.drivers import pfc
@ -704,116 +703,3 @@ class PFCDriverStringTest(base.BaseTestCase):
ret_str = self.driver._generate_pfc_description(random_str)
self.assertEqual(exp_str, ret_str)
class PFCIdConvertTest(base.BaseTestCase):
driver = 'neutron.plugins.nec.drivers.pfc.PFCDriverBase'
def setUp(self):
super(PFCIdConvertTest, self).setUp()
self.driver = drivers.get_driver(self.driver)(TestConfig)
self.ctx = mock.Mock()
self.ctx.session = "session"
self.get_ofc_id_lookup_both = mock.patch.object(
ndb, 'get_ofc_id_lookup_both').start()
self.addCleanup(mock.patch.stopall)
def generate_random_ids(self, count=1):
if count == 1:
return uuidutils.generate_uuid()
else:
return [uuidutils.generate_uuid() for _ in xrange(count)]
def test_convert_tenant_id(self):
ofc_t_id = self.generate_random_ids(1)
ret = self.driver.convert_ofc_tenant_id(self.ctx, ofc_t_id)
self.assertEqual(ret, '/tenants/%s' % ofc_t_id)
def test_convert_tenant_id_noconv(self):
ofc_t_id = '/tenants/%s' % self.generate_random_ids(1)
ret = self.driver.convert_ofc_tenant_id(self.ctx, ofc_t_id)
self.assertEqual(ret, ofc_t_id)
def test_convert_network_id(self):
t_id, ofc_t_id, ofc_n_id = self.generate_random_ids(3)
self.get_ofc_id_lookup_both.return_value = ofc_t_id
ret = self.driver.convert_ofc_network_id(self.ctx, ofc_n_id, t_id)
self.assertEqual(ret, ('/tenants/%(tenant)s/networks/%(network)s' %
{'tenant': ofc_t_id, 'network': ofc_n_id}))
self.get_ofc_id_lookup_both.assert_called_once_with(
self.ctx.session, 'ofc_tenant', t_id)
def test_convert_network_id_with_new_tenant_id(self):
t_id, ofc_t_id, ofc_n_id = self.generate_random_ids(3)
ofc_t_path = '/tenants/%s' % ofc_t_id
self.get_ofc_id_lookup_both.return_value = ofc_t_path
ret = self.driver.convert_ofc_network_id(self.ctx, ofc_n_id, t_id)
self.assertEqual(ret, ('/tenants/%(tenant)s/networks/%(network)s' %
{'tenant': ofc_t_id, 'network': ofc_n_id}))
self.get_ofc_id_lookup_both.assert_called_once_with(
self.ctx.session, 'ofc_tenant', t_id)
def test_convert_network_id_noconv(self):
t_id = 'dummy'
ofc_t_id, ofc_n_id = self.generate_random_ids(2)
ofc_n_id = ('/tenants/%(tenant)s/networks/%(network)s' %
{'tenant': ofc_t_id, 'network': ofc_n_id})
ret = self.driver.convert_ofc_network_id(self.ctx, ofc_n_id, t_id)
self.assertEqual(ret, ofc_n_id)
def test_convert_port_id(self):
t_id, n_id = self.generate_random_ids(2)
ofc_t_id, ofc_n_id, ofc_p_id = self.generate_random_ids(3)
self.get_ofc_id_lookup_both.side_effect = [ofc_n_id, ofc_t_id]
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
exp = ('/tenants/%(tenant)s/networks/%(network)s/ports/%(port)s' %
{'tenant': ofc_t_id, 'network': ofc_n_id, 'port': ofc_p_id})
self.assertEqual(ret, exp)
self.get_ofc_id_lookup_both.assert_has_calls([
mock.call(self.ctx.session, 'ofc_network', n_id),
mock.call(self.ctx.session, 'ofc_tenant', t_id),
])
def test_convert_port_id_with_new_tenant_id(self):
t_id, n_id = self.generate_random_ids(2)
ofc_t_id, ofc_n_id, ofc_p_id = self.generate_random_ids(3)
ofc_t_path = '/tenants/%s' % ofc_t_id
self.get_ofc_id_lookup_both.side_effect = [ofc_n_id, ofc_t_path]
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
exp = ('/tenants/%(tenant)s/networks/%(network)s/ports/%(port)s' %
{'tenant': ofc_t_id, 'network': ofc_n_id, 'port': ofc_p_id})
self.assertEqual(ret, exp)
self.get_ofc_id_lookup_both.assert_has_calls([
mock.call(self.ctx.session, 'ofc_network', n_id),
mock.call(self.ctx.session, 'ofc_tenant', t_id),
])
def test_convert_port_id_with_new_network_id(self):
t_id, n_id = self.generate_random_ids(2)
ofc_t_id, ofc_n_id, ofc_p_id = self.generate_random_ids(3)
ofc_n_path = ('/tenants/%(tenant)s/networks/%(network)s' %
{'tenant': ofc_t_id, 'network': ofc_n_id})
self.get_ofc_id_lookup_both.return_value = ofc_n_path
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
exp = ('/tenants/%(tenant)s/networks/%(network)s/ports/%(port)s' %
{'tenant': ofc_t_id, 'network': ofc_n_id, 'port': ofc_p_id})
self.assertEqual(ret, exp)
self.get_ofc_id_lookup_both.assert_called_once_with(
self.ctx.session, 'ofc_network', n_id)
def test_convert_port_id_noconv(self):
t_id = n_id = 'dummy'
ofc_t_id, ofc_n_id, ofc_p_id = self.generate_random_ids(3)
ofc_p_id = ('/tenants/%(tenant)s/networs/%(network)s/ports/%(port)s'
% {'tenant': ofc_t_id, 'network': ofc_n_id,
'port': ofc_p_id})
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
self.assertEqual(ret, ofc_p_id)

140
neutron/tests/unit/nec/test_trema_driver.py
View File

@ -21,7 +21,6 @@ import mock
from neutron.openstack.common import uuidutils
from neutron.plugins.nec.common import ofc_client
from neutron.plugins.nec.db import api as ndb
from neutron.plugins.nec.db import models as nmodels
from neutron.plugins.nec import drivers
from neutron.tests import base
@ -349,142 +348,3 @@ class TremaFilterDriverTest(TremaDriverTestBase):
f_path = "/filters/%s" % uuidutils.generate_uuid()
self.driver.delete_filter(f_path)
self.do_request.assert_called_once_with("DELETE", f_path)
def generate_random_ids(count=1):
if count == 1:
return uuidutils.generate_uuid()
else:
return [uuidutils.generate_uuid() for i in xrange(count)]
class TremaIdConvertTest(base.BaseTestCase):
driver_name = 'trema'
def setUp(self):
super(TremaIdConvertTest, self).setUp()
self.driver = drivers.get_driver(self.driver_name)(TestConfig)
self.ctx = mock.Mock()
def test_convert_tenant_id(self):
ofc_t_id = generate_random_ids(1)
ret = self.driver.convert_ofc_tenant_id(self.ctx, ofc_t_id)
self.assertEqual(ret, '/tenants/%s' % ofc_t_id)
def test_convert_tenant_id_noconv(self):
ofc_t_id = '/tenants/%s' % generate_random_ids(1)
ret = self.driver.convert_ofc_tenant_id(self.ctx, ofc_t_id)
self.assertEqual(ret, ofc_t_id)
def test_convert_network_id(self):
t_id, ofc_t_id, ofc_n_id = generate_random_ids(3)
ret = self.driver.convert_ofc_network_id(self.ctx, ofc_n_id, t_id)
self.assertEqual(ret, ('/networks/%s' % ofc_n_id))
def test_convert_network_id_noconv(self):
t_id = 'dummy'
ofc_t_id, ofc_n_id = generate_random_ids(2)
ofc_n_id = '/networks/%s' % ofc_n_id
self.driver.convert_ofc_network_id(self.ctx, ofc_n_id, t_id)
def test_convert_filter_id(self):
ofc_f_id = generate_random_ids(1)
ret = self.driver.convert_ofc_filter_id(self.ctx, ofc_f_id)
self.assertEqual(ret, '/filters/%s' % ofc_f_id)
def test_convert_filter_id_noconv(self):
ofc_f_id = '/filters/%s' % generate_random_ids(1)
ret = self.driver.convert_ofc_filter_id(self.ctx, ofc_f_id)
self.assertEqual(ret, ofc_f_id)
class TremaIdConvertTestBase(base.BaseTestCase):
def setUp(self):
super(TremaIdConvertTestBase, self).setUp()
self.driver = drivers.get_driver(self.driver_name)(TestConfig)
self.ctx = mock.Mock()
self.ctx.session = "session"
self.get_ofc_id_lookup_both = mock.patch.object(
ndb, 'get_ofc_id_lookup_both').start()
self.addCleanup(mock.patch.stopall)
def _test_convert_port_id(self, port_path_template):
t_id, n_id = generate_random_ids(2)
ofc_n_id, ofc_p_id = generate_random_ids(2)
self.get_ofc_id_lookup_both.return_value = ofc_n_id
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
exp = port_path_template % {'network': ofc_n_id, 'port': ofc_p_id}
self.assertEqual(ret, exp)
self.get_ofc_id_lookup_both.assert_called_once_with(
self.ctx.session, 'ofc_network', n_id)
def _test_convert_port_id_with_new_network_id(self, port_path_template):
t_id, n_id = generate_random_ids(2)
ofc_n_id, ofc_p_id = generate_random_ids(2)
ofc_n_path = '/networks/%s' % ofc_n_id
self.get_ofc_id_lookup_both.return_value = ofc_n_path
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
exp = port_path_template % {'network': ofc_n_id, 'port': ofc_p_id}
self.assertEqual(ret, exp)
self.get_ofc_id_lookup_both.assert_called_once_with(
self.ctx.session, 'ofc_network', n_id)
def _test_convert_port_id_noconv(self, port_path_template):
t_id = n_id = 'dummy'
ofc_n_id, ofc_p_id = generate_random_ids(2)
ofc_p_id = port_path_template % {'network': ofc_n_id, 'port': ofc_p_id}
ret = self.driver.convert_ofc_port_id(self.ctx, ofc_p_id, t_id, n_id)
self.assertEqual(ret, ofc_p_id)
class TremaIdConvertPortBaseTest(TremaIdConvertTestBase):
driver_name = "trema_port"
def test_convert_port_id(self):
self._test_convert_port_id('/networks/%(network)s/ports/%(port)s')
def test_convert_port_id_with_new_network_id(self):
self._test_convert_port_id_with_new_network_id(
'/networks/%(network)s/ports/%(port)s')
def test_convert_port_id_noconv(self):
self._test_convert_port_id_noconv(
'/networs/%(network)s/ports/%(port)s')
class TremaIdConvertPortMACBaseTest(TremaIdConvertTestBase):
driver_name = "trema_portmac"
def test_convert_port_id(self):
self._test_convert_port_id(
'/networks/%(network)s/ports/dummy-%(port)s/attachments/%(port)s')
def test_convert_port_id_with_new_network_id(self):
self._test_convert_port_id_with_new_network_id(
'/networks/%(network)s/ports/dummy-%(port)s/attachments/%(port)s')
def test_convert_port_id_noconv(self):
self._test_convert_port_id_noconv(
'/networs/%(network)s/ports/dummy-%(port)s/attachments/%(port)s')
class TremaIdConvertMACBaseTest(TremaIdConvertTestBase):
driver_name = "trema_mac"
def test_convert_port_id(self):
self._test_convert_port_id(
'/networks/%(network)s/attachments/%(port)s')
def test_convert_port_id_with_new_network_id(self):
self._test_convert_port_id_with_new_network_id(
'/networks/%(network)s/attachments/%(port)s')
def test_convert_port_id_noconv(self):
self._test_convert_port_id_noconv(
'/networs/%(network)s/attachments/%(port)s')