--- /dev/null
+# 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')
+ )
'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)
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).
"""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),
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
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."""
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
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."""
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.
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.
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.
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}
: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
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):
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)
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)
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:
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)
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))
+
+ 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'))
- 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)
+
+class NECPluginV2DBPortInfoTest(NECPluginV2DBTestBase):
def _compare_portinfo(self, portinfo, expected):
self.assertEqual(portinfo.id, expected['port_id'])
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)
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))
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
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)
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
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')
Code Review / openstack-build / neutron-build.git / commitdiff