--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright 2013 OpenStack LLC
+#
+# 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 plugin sharednet
+
+Revision ID: 3b54bf9e29f7
+Revises: 54c2c487e913
+Create Date: 2013-02-17 09:21:48.287134
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '3b54bf9e29f7'
+down_revision = '54c2c487e913'
+
+# Change to ['*'] if this migration applies to all plugins
+
+migration_for_plugins = [
+ 'quantum.plugins.nec.nec_plugin.NECPluginV2'
+]
+
+from alembic import op
+import sqlalchemy as sa
+
+
+from quantum.db import migration
+
+
+def upgrade(active_plugin=None, options=None):
+ if not migration.should_run(active_plugin, migration_for_plugins):
+ return
+
+ op.create_table(
+ 'ofctenantmappings',
+ sa.Column('ofc_id', sa.String(length=255), nullable=False),
+ sa.Column('quantum_id', sa.String(length=36), nullable=False),
+ sa.PrimaryKeyConstraint('quantum_id'),
+ sa.UniqueConstraint('ofc_id')
+ )
+ op.create_table(
+ 'ofcnetworkmappings',
+ sa.Column('ofc_id', sa.String(length=255), nullable=False),
+ sa.Column('quantum_id', sa.String(length=36), nullable=False),
+ sa.PrimaryKeyConstraint('quantum_id'),
+ sa.UniqueConstraint('ofc_id')
+ )
+ op.create_table(
+ 'ofcportmappings',
+ sa.Column('ofc_id', sa.String(length=255), nullable=False),
+ sa.Column('quantum_id', sa.String(length=36), nullable=False),
+ sa.PrimaryKeyConstraint('quantum_id'),
+ sa.UniqueConstraint('ofc_id')
+ )
+ op.create_table(
+ 'ofcfiltermappings',
+ sa.Column('ofc_id', sa.String(length=255), nullable=False),
+ sa.Column('quantum_id', sa.String(length=36), nullable=False),
+ sa.PrimaryKeyConstraint('quantum_id'),
+ sa.UniqueConstraint('ofc_id')
+ )
+
+
+def downgrade(active_plugin=None, options=None):
+ if not migration.should_run(active_plugin, migration_for_plugins):
+ return
+
+ op.drop_table('ofcfiltermappings')
+ op.drop_table('ofcportmappings')
+ op.drop_table('ofcnetworkmappings')
+ op.drop_table('ofctenantmappings')
return httplib.HTTPConnection
def do_request(self, method, action, body=None):
- LOG.debug(_("Client request: %(method)s %(action)s [%(body)s]"),
- locals())
-
+ LOG.debug(_("Client request: %(host)s:%(port)s "
+ "%(method)s %(action)s [%(body)s]"),
+ {'host': self.host, 'port': self.port,
+ 'method': method, 'action': action, 'body': body})
if type(body) is dict:
body = json.dumps(body)
try:
OFP_VLAN_NONE = 0xffff
+resource_map = {'ofc_tenant': nmodels.OFCTenantMapping,
+ 'ofc_network': nmodels.OFCNetworkMapping,
+ 'ofc_port': nmodels.OFCPortMapping,
+ '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:
+ return old_resource_map[resource]
+ else:
+ return resource_map[resource]
+
+
def initialize():
db.configure_db()
db.clear_db(base)
-def get_ofc_item(model, id):
- session = db.get_session()
+def get_ofc_item(session, resource, quantum_id, old_style=False):
try:
- return (session.query(model).
- filter_by(id=id).
- one())
+ model = _get_resource_model(resource, old_style)
+ return session.query(model).filter_by(quantum_id=quantum_id).one()
except sa.orm.exc.NoResultFound:
return None
-def find_ofc_item(model, quantum_id):
- session = db.get_session()
+def get_ofc_id(session, resource, quantum_id, old_style=False):
+ ofc_item = get_ofc_item(session, resource, quantum_id, old_style)
+ if ofc_item:
+ if old_style:
+ return ofc_item.id
+ else:
+ return ofc_item.ofc_id
+ else:
+ return None
+
+
+def exists_ofc_item(session, resource, quantum_id, old_style=False):
+ if get_ofc_item(session, resource, quantum_id, old_style):
+ return True
+ else:
+ return False
+
+
+def find_ofc_item(session, resource, ofc_id, old_style=False):
try:
- return (session.query(model).
- filter_by(quantum_id=quantum_id).
- one())
+ model = _get_resource_model(resource, old_style)
+ if old_style:
+ params = dict(id=ofc_id)
+ else:
+ 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(model, id, quantum_id):
- session = db.get_session()
+def add_ofc_item(session, resource, quantum_id, ofc_id, old_style=False):
try:
- item = model(id=id, quantum_id=quantum_id)
+ model = _get_resource_model(resource, old_style)
+ if old_style:
+ params = dict(quantum_id=quantum_id, id=ofc_id)
+ else:
+ params = dict(quantum_id=quantum_id, ofc_id=ofc_id)
+ item = model(**params)
session.add(item)
session.flush()
except Exception as exc:
return item
-def del_ofc_item(model, id):
- session = db.get_session()
+def del_ofc_item(session, resource, quantum_id, old_style=False,
+ warning=True):
try:
- item = (session.query(model).
- filter_by(id=id).
- one())
+ model = _get_resource_model(resource, old_style)
+ item = session.query(model).filter_by(quantum_id=quantum_id).one()
session.delete(item)
session.flush()
+ return True
except sa.orm.exc.NoResultFound:
- LOG.warning(_("_del_ofc_item(): NotFound item "
- "(model=%(model)s, id=%(id)s) "), locals())
-
-
-def get_portinfo(id):
- session = db.get_session()
+ if warning:
+ LOG.warning(_("_del_ofc_item(): NotFound item "
+ "(model=%(model)s, id=%(id)s) "),
+ {'model': model, 'id': quantum_id})
+ return False
+
+
+def get_ofc_id_lookup_both(session, resource, quantum_id):
+ ofc_id = get_ofc_id(session, resource, quantum_id)
+ # Lookup old style of OFC mapping table
+ if not ofc_id:
+ ofc_id = get_ofc_id(session, resource, quantum_id,
+ old_style=True)
+ if not ofc_id:
+ reason = (_("NotFound %(resource)s for quantum_id=%(id)s.")
+ % {'resource': resource, 'id': quantum_id})
+ raise nexc.OFCConsistencyBroken(reason=reason)
+ return ofc_id
+
+
+def exists_ofc_item_lookup_both(session, resource, quantum_id):
+ if exists_ofc_item(session, resource, quantum_id):
+ return True
+ # Check old style of OFC mapping table
+ if exists_ofc_item(session, resource, quantum_id,
+ old_style=True):
+ return True
+ return False
+
+
+def del_ofc_item_lookup_both(session, resource, quantum_id):
+ # Delete the mapping from new style of OFC mapping table
+ if del_ofc_item(session, resource, quantum_id,
+ old_style=False, warning=False):
+ return
+ # Delete old style of OFC mapping table
+ if del_ofc_item(session, resource, quantum_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': quantum_id})
+
+
+def get_portinfo(session, id):
try:
return (session.query(nmodels.PortInfo).
filter_by(id=id).
return None
-def add_portinfo(id, datapath_id='', port_no=0, vlan_id=OFP_VLAN_NONE, mac=''):
- session = db.get_session()
+def add_portinfo(session, id, datapath_id='', port_no=0,
+ vlan_id=OFP_VLAN_NONE, mac=''):
try:
portinfo = nmodels.PortInfo(id=id, datapath_id=datapath_id,
port_no=port_no, vlan_id=vlan_id, mac=mac)
return portinfo
-def del_portinfo(id):
- session = db.get_session()
+def del_portinfo(session, id):
try:
- portinfo = (session.query(nmodels.PortInfo).
- filter_by(id=id).
- one())
+ portinfo = session.query(nmodels.PortInfo).filter_by(id=id).one()
session.delete(portinfo)
session.flush()
except sa.orm.exc.NoResultFound:
from quantum.db import models_v2
+"""New mapping tables"""
+
+
+class OFCId(object):
+ """Resource ID on OpenFlow Controller"""
+ ofc_id = sa.Column(sa.String(255), unique=True, nullable=False)
+
+
+class QuantumId(object):
+ """Logical ID on Quantum"""
+ quantum_id = sa.Column(sa.String(36), primary_key=True)
+
+
+class OFCTenantMapping(model_base.BASEV2, QuantumId, OFCId):
+ """Represents a Tenant on OpenFlow Network/Controller."""
+
+
+class OFCNetworkMapping(model_base.BASEV2, QuantumId, OFCId):
+ """Represents a Network on OpenFlow Network/Controller."""
+
+
+class OFCPortMapping(model_base.BASEV2, QuantumId, OFCId):
+ """Represents a Port on OpenFlow Network/Controller."""
+
+
+class OFCFilterMapping(model_base.BASEV2, QuantumId, OFCId):
+ """Represents a Filter on OpenFlow Network/Controller."""
+
+
+"""Old mapping tables"""
+
+
class HasQuantumId(object):
"""Logical ID on Quantum"""
quantum_id = sa.Column(sa.String(36), nullable=False)
'trema_port': DRIVER_PATH % "trema.TremaPortBaseDriver",
'trema_portmac': DRIVER_PATH % "trema.TremaPortMACBaseDriver",
'trema_mac': DRIVER_PATH % "trema.TremaMACBaseDriver",
- 'pfc': DRIVER_PATH % "pfc.PFCDriver"}
+ 'pfc': DRIVER_PATH % "pfc.PFCV4Driver",
+ 'pfc_v3': DRIVER_PATH % "pfc.PFCV3Driver",
+ 'pfc_v4': DRIVER_PATH % "pfc.PFCV4Driver"}
def get_driver(driver_name):
# License for the specific language governing permissions and limitations
# under the License.
# @author: Ryota MIBU
+# @author: Akihiro MOTOKI
+import re
+import uuid
+
+from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec.common import ofc_client
from quantum.plugins.nec import ofc_driver_base
-TENANTS_PATH = "/tenants"
-TENANT_PATH = "/tenants/%s"
-NETWORKS_PATH = "/tenants/%s/networks"
-NETWORK_PATH = "/tenants/%s/networks/%s"
-PORTS_PATH = "/tenants/%s/networks/%s/ports"
-PORT_PATH = "/tenants/%s/networks/%s/ports/%s"
+class PFCDriverBase(ofc_driver_base.OFCDriverBase):
+ """Base Class for PDC Drivers
+ PFCDriverBase provides methods to handle PFC resources through REST API.
+ This uses ofc resource path instead of ofc resource ID.
-class PFCDriver(ofc_driver_base.OFCDriverBase):
+ The class implements the API for PFC V4.0 or later.
+ """
def __init__(self, conf_ofc):
self.client = ofc_client.OFCClient(host=conf_ofc.host,
def filter_supported(cls):
return False
+ def _generate_pfc_str(self, raw_str):
+ """Generate PFC acceptable String"""
+ return re.sub(r'[^0-9a-zA-Z]', '_', raw_str)
+
+ def _generate_pfc_id(self, id_str):
+ """Generate ID on PFC
+
+ Currently, PFC ID must be less than 32.
+ Shorten UUID string length from 36 to 31 by follows:
+ * delete UUID Version and hyphen (see RFC4122)
+ * ensure str length
+ """
+ try:
+ # openstack.common.uuidutils.is_uuid_like() returns
+ # False for KeyStone tenant_id, so uuid.UUID is used
+ # directly here to accept tenant_id as UUID string
+ uuid_str = str(uuid.UUID(id_str)).replace('-', '')
+ uuid_no_version = uuid_str[:12] + uuid_str[13:]
+ return uuid_no_version[:31]
+ except:
+ return self._generate_pfc_str(id_str)[:31]
+
+ def _generate_pfc_description(self, desc):
+ """Generate Description on PFC
+
+ Currently, PFC Description must be less than 128.
+ """
+ return self._generate_pfc_str(desc)[:127]
+
def create_tenant(self, description, tenant_id=None):
- body = {'description': description}
- if tenant_id:
- body.update({'id': tenant_id})
- res = self.client.post(TENANTS_PATH, body=body)
- ofc_tenant_id = res['id']
- return ofc_tenant_id
-
- def update_tenant(self, ofc_tenant_id, description):
- path = TENANT_PATH % ofc_tenant_id
- body = {'description': description}
- res = self.client.put(path, body=body)
+ ofc_tenant_id = self._generate_pfc_id(tenant_id)
+ body = {'id': ofc_tenant_id}
+ res = self.client.post('/tenants', body=body)
+ return '/tenants/' + ofc_tenant_id
def delete_tenant(self, ofc_tenant_id):
- path = TENANT_PATH % ofc_tenant_id
- return self.client.delete(path)
+ return self.client.delete(ofc_tenant_id)
def create_network(self, ofc_tenant_id, description, network_id=None):
- path = NETWORKS_PATH % ofc_tenant_id
- body = {'description': description}
- if network_id:
- body.update({'id': network_id})
+ path = "%s/networks" % ofc_tenant_id
+ pfc_desc = self._generate_pfc_description(description)
+ body = {'description': pfc_desc}
res = self.client.post(path, body=body)
ofc_network_id = res['id']
- return ofc_network_id
-
- def update_network(self, ofc_tenant_id, ofc_network_id, description):
- path = NETWORK_PATH % (ofc_tenant_id, ofc_network_id)
- body = {'description': description}
- return self.client.put(path, body=body)
+ return path + '/' + ofc_network_id
- def delete_network(self, ofc_tenant_id, ofc_network_id):
- path = NETWORK_PATH % (ofc_tenant_id, ofc_network_id)
- return self.client.delete(path)
+ def delete_network(self, ofc_network_id):
+ return self.client.delete(ofc_network_id)
- def create_port(self, ofc_tenant_id, ofc_network_id, portinfo,
+ def create_port(self, ofc_network_id, portinfo,
port_id=None):
- path = PORTS_PATH % (ofc_tenant_id, ofc_network_id)
+ path = "%s/ports" % ofc_network_id
body = {'datapath_id': portinfo.datapath_id,
'port': str(portinfo.port_no),
'vid': str(portinfo.vlan_id)}
- if port_id:
- body.update({'id': port_id})
res = self.client.post(path, body=body)
ofc_port_id = res['id']
- return ofc_port_id
+ return path + '/' + ofc_port_id
+
+ 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 PFCV3Driver(PFCDriverBase):
+
+ def create_tenant(self, description, tenant_id):
+ ofc_tenant_id = self._generate_pfc_id(tenant_id)
+ return "/tenants/" + ofc_tenant_id
+
+ def delete_tenant(self, ofc_tenant_id):
+ pass
- def update_port(self, ofc_tenant_id, ofc_network_id, portinfo, port_id):
- path = PORT_PATH % (ofc_tenant_id, ofc_network_id, ofc_port_id)
- body = {'datapath_id': portinfo.datapath_id,
- 'port': str(portinfo.port_no),
- 'vid': str(portinfo.vlan_id)}
- res = self.client.put(path, body=body)
- ofc_port_id = res['id']
- return ofc_port_id
- def delete_port(self, ofc_tenant_id, ofc_network_id, ofc_port_id):
- path = PORT_PATH % (ofc_tenant_id, ofc_network_id, ofc_port_id)
- return self.client.delete(path)
+class PFCV4Driver(PFCDriverBase):
+ pass
# License for the specific language governing permissions and limitations
# under the License.
# @author: Ryota MIBU
+# @author: Akihiro MOTOKI
from quantum.openstack.common import uuidutils
+from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec.common import ofc_client
from quantum.plugins.nec import ofc_driver_base
self.client = ofc_client.OFCClient(host=conf_ofc.host,
port=conf_ofc.port)
+ def _get_network_id(self, ofc_network_id):
+ # ofc_network_id : /networks/<network-id>
+ return ofc_network_id.split('/')[2]
+
+ def _get_tenant_id(self, tenant_id):
+ # Trema does not use tenant_id, but it returns
+ # /tenants/<tenant_id> format to keep consistency with PFC driver.
+ return '/tenants/' + tenant_id
+
def create_tenant(self, description, tenant_id=None):
- return tenant_id or uuidutils.generate_uuid()
+ return self._get_tenant_id(tenant_id or uuidutils.generate_uuid())
def update_tenant(self, ofc_tenant_id, description):
pass
ofc_network_id = network_id or uuidutils.generate_uuid()
body = {'id': ofc_network_id, 'description': description}
self.client.post(self.networks_path, body=body)
- return ofc_network_id
+ return self.network_path % ofc_network_id
- def update_network(self, ofc_tenant_id, ofc_network_id, description):
- path = self.network_path % ofc_network_id
- body = {'description': description}
- return self.client.put(path, body=body)
+ def delete_network(self, ofc_network_id):
+ return self.client.delete(ofc_network_id)
- def delete_network(self, ofc_tenant_id, ofc_network_id):
- path = self.network_path % ofc_network_id
- return self.client.delete(path)
+ 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 update_port(self, ofc_tenant_id, ofc_network_id, ofc_port_id,
- portinfo):
- self.delete_port(ofc_tenant_id, ofc_network_id, ofc_port_id)
- self.create_port(ofc_tenant_id, ofc_network_id, portinfo, ofc_port_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 TremaFilterDriver(object):
- """Trema (Sliceable Switch) PacketFilter Driver"""
+ """Trema (Sliceable Switch) PacketFilter Driver Mixin"""
filters_path = "/filters"
filter_path = "/filters/%s"
def filter_supported(cls):
return True
- def create_filter(self, ofc_tenant_id, ofc_network_id, filter_dict,
+ def create_filter(self, ofc_network_id, filter_dict,
portinfo=None, filter_id=None):
if filter_dict['action'].upper() in ["ACCEPT", "ALLOW"]:
ofc_action = "ALLOW"
ofc_action = "DENY"
body = {'priority': filter_dict['priority'],
- 'slice': ofc_network_id,
+ 'slice': self._get_network_id(ofc_network_id),
'action': ofc_action}
ofp_wildcards = ["dl_vlan", "dl_vlan_pcp", "nw_tos"]
body['ofp_wildcards'] = ','.join(ofp_wildcards)
self.client.post(self.filters_path, body=body)
- return ofc_filter_id
+ return self.filter_path % ofc_filter_id
+
+ def delete_filter(self, ofc_filter_id):
+ return self.client.delete(ofc_filter_id)
- def delete_filter(self, ofc_tenant_id, ofc_network_id, ofc_filter_id):
- path = self.filter_path % ofc_filter_id
- return self.client.delete(path)
+ 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, TremaFilterDriver):
TremaPortBaseDriver uses port base binding.
Ports are identified by datapath_id, port_no and vlan_id.
"""
- ports_path = "/networks/%s/ports"
- port_path = "/networks/%s/ports/%s"
+ ports_path = "%(network)s/ports"
+ port_path = "%(network)s/ports/%(port)s"
- def create_port(self, ofc_tenant_id, ofc_network_id, portinfo,
+ def create_port(self, ofc_network_id, portinfo,
port_id=None):
ofc_port_id = port_id or uuidutils.generate_uuid()
- path = self.ports_path % ofc_network_id
+ path = self.ports_path % {'network': ofc_network_id}
body = {'id': ofc_port_id,
'datapath_id': portinfo.datapath_id,
'port': str(portinfo.port_no),
'vid': str(portinfo.vlan_id)}
self.client.post(path, body=body)
- return ofc_port_id
+ return self.port_path % {'network': ofc_network_id,
+ 'port': ofc_port_id}
- def delete_port(self, ofc_tenant_id, ofc_network_id, ofc_port_id):
- path = self.port_path % (ofc_network_id, ofc_port_id)
- return self.client.delete(path)
+ 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, TremaFilterDriver):
TremaPortBaseDriver uses port-mac base binding.
Ports are identified by datapath_id, port_no, vlan_id and mac.
"""
- ports_path = "/networks/%s/ports"
- port_path = "/networks/%s/ports/%s"
- attachments_path = "/networks/%s/ports/%s/attachments"
- attachment_path = "/networks/%s/ports/%s/attachments/%s"
+ ports_path = "%(network)s/ports"
+ port_path = "%(network)s/ports/%(port)s"
+ attachments_path = "%(network)s/ports/%(port)s/attachments"
+ attachment_path = "%(network)s/ports/%(port)s/attachments/%(attachment)s"
- def create_port(self, ofc_tenant_id, ofc_network_id, portinfo,
- port_id=None):
+ def create_port(self, ofc_network_id, portinfo, port_id=None):
#NOTE: This Driver create slices with Port-MAC Based bindings on Trema
# Sliceable. It's REST API requires Port Based binding before you
# define Port-MAC Based binding.
ofc_port_id = port_id or uuidutils.generate_uuid()
dummy_port_id = "dummy-%s" % ofc_port_id
- path = self.ports_path % ofc_network_id
+ path = self.ports_path % {'network': ofc_network_id}
body = {'id': dummy_port_id,
'datapath_id': portinfo.datapath_id,
'port': str(portinfo.port_no),
'vid': str(portinfo.vlan_id)}
self.client.post(path, body=body)
- path = self.attachments_path % (ofc_network_id, dummy_port_id)
+ path = self.attachments_path % {'network': ofc_network_id,
+ 'port': dummy_port_id}
body = {'id': ofc_port_id, 'mac': portinfo.mac}
self.client.post(path, body=body)
- path = self.port_path % (ofc_network_id, dummy_port_id)
+ path = self.port_path % {'network': ofc_network_id,
+ 'port': dummy_port_id}
self.client.delete(path)
- return ofc_port_id
+ return self.attachment_path % {'network': ofc_network_id,
+ 'port': dummy_port_id,
+ 'attachment': ofc_port_id}
- def delete_port(self, ofc_tenant_id, ofc_network_id, ofc_port_id):
- dummy_port_id = "dummy-%s" % ofc_port_id
- path = self.attachment_path % (ofc_network_id, dummy_port_id,
- ofc_port_id)
- return self.client.delete(path)
+ 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):
TremaPortBaseDriver uses mac base binding.
Ports are identified by mac.
"""
- attachments_path = "/networks/%s/attachments"
- attachment_path = "/networks/%s/attachments/%s"
+ attachments_path = "%(network)s/attachments"
+ attachment_path = "%(network)s/attachments/%(attachment)s"
@classmethod
def filter_supported(cls):
return False
- def create_port(self, ofc_tenant_id, ofc_network_id, portinfo,
- port_id=None):
+ def create_port(self, ofc_network_id, portinfo, port_id=None):
ofc_port_id = port_id or uuidutils.generate_uuid()
- path = self.attachments_path % ofc_network_id
+ path = self.attachments_path % {'network': ofc_network_id}
body = {'id': ofc_port_id, 'mac': portinfo.mac}
self.client.post(path, body=body)
- return ofc_port_id
-
- def delete_port(self, ofc_tenant_id, ofc_network_id, ofc_port_id):
- path = self.attachment_path % (ofc_network_id, ofc_port_id)
- return self.client.delete(path)
+ return self.attachment_path % {'network': ofc_network_id,
+ 'attachment': ofc_port_id}
+
+ 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}
# License for the specific language governing permissions and limitations
# under the License.
# @author: Ryota MIBU
+# @author: Akihiro MOTOKI
from quantum.agent import securitygroups_rpc as sg_rpc
from quantum.common import constants as q_const
LOG.debug(_("activate_port_if_ready(): skip, "
"network.admin_state_up is False."))
port_status = OperationalStatus.DOWN
- elif not ndb.get_portinfo(port['id']):
+ elif not ndb.get_portinfo(context.session, port['id']):
LOG.debug(_("activate_port_if_ready(): skip, "
"no portinfo for this port."))
port_status = OperationalStatus.DOWN
in_port=port)
if port_status in [OperationalStatus.ACTIVE]:
- if self.ofc.exists_ofc_port(port['id']):
+ if self.ofc.exists_ofc_port(context, port['id']):
LOG.debug(_("activate_port_if_ready(): skip, "
"ofc_port already exists."))
else:
try:
- self.ofc.create_ofc_port(port['tenant_id'],
- port['network_id'],
- port['id'])
+ self.ofc.create_ofc_port(context, port['id'], port)
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = _("create_ofc_port() failed due to %s") % exc
LOG.error(reason)
Deactivate port and packet_filters associated with the port.
"""
port_status = OperationalStatus.DOWN
- if self.ofc.exists_ofc_port(port['id']):
+ if self.ofc.exists_ofc_port(context, port['id']):
try:
- self.ofc.delete_ofc_port(port['tenant_id'],
- port['network_id'],
- port['id'])
+ self.ofc.delete_ofc_port(context, port['id'], port)
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = _("delete_ofc_port() failed due to %s") % exc
LOG.error(reason)
OperationalStatus.BUILD)
try:
- if not self.ofc.exists_ofc_tenant(new_net['tenant_id']):
- self.ofc.create_ofc_tenant(new_net['tenant_id'])
- self.ofc.create_ofc_network(new_net['tenant_id'], new_net['id'],
- new_net['name'])
+ if not self.ofc.exists_ofc_tenant(context, new_net['tenant_id']):
+ self.ofc.create_ofc_tenant(context, new_net['tenant_id'])
+ self.ofc.create_ofc_network(context, new_net['tenant_id'],
+ new_net['id'], new_net['name'])
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = _("create_network() failed due to %s") % exc
LOG.error(reason)
or deactivate ports and packetfilters associated with the network.
"""
LOG.debug(_("NECPluginV2.update_network() called, "
- "id=%(id)s network=%(network)s ."), locals())
+ "id=%(id)s network=%(network)s ."),
+ {'id': id, 'network': network})
session = context.session
with session.begin(subtransactions=True):
old_net = super(NECPluginV2, self).get_network(context, id)
super(NECPluginV2, self).delete_network(context, id)
try:
- self.ofc.delete_ofc_network(tenant_id, id)
+ # 'net' parameter is required to lookup old OFC mapping
+ self.ofc.delete_ofc_network(context, id, net)
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = _("delete_network() failed due to %s") % exc
# NOTE: The OFC configuration of this network could be remained
nets = super(NECPluginV2, self).get_networks(context, filters=filters)
if len(nets) == 0:
try:
- self.ofc.delete_ofc_tenant(tenant_id)
+ self.ofc.delete_ofc_tenant(context, tenant_id)
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = _("delete_ofc_tenant() failed due to %s") % exc
LOG.warn(reason)
or deactivate the port and packetfilters associated with it.
"""
LOG.debug(_("NECPluginV2.update_port() called, "
- "id=%(id)s port=%(port)s ."), locals())
+ "id=%(id)s port=%(port)s ."),
+ {'id': id, 'port': port})
need_port_update_notify = False
with context.session.begin(subtransactions=True):
old_port = super(NECPluginV2, self).get_port(context, id)
LOG.debug(_("_activate_packet_filter_if_ready(): skip, "
"invalid in_port_id."))
pf_status = OperationalStatus.DOWN
- elif in_port_id and not ndb.get_portinfo(in_port_id):
+ elif in_port_id and not ndb.get_portinfo(context.session, in_port_id):
LOG.debug(_("_activate_packet_filter_if_ready(): skip, "
"no portinfo for in_port."))
pf_status = OperationalStatus.DOWN
if pf_status in [OperationalStatus.ACTIVE]:
- if self.ofc.exists_ofc_packet_filter(packet_filter['id']):
+ if self.ofc.exists_ofc_packet_filter(context, packet_filter['id']):
LOG.debug(_("_activate_packet_filter_if_ready(): skip, "
"ofc_packet_filter already exists."))
else:
try:
(self.ofc.
- create_ofc_packet_filter(packet_filter['tenant_id'],
- packet_filter['network_id'],
+ create_ofc_packet_filter(context,
packet_filter['id'],
packet_filter))
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
def _deactivate_packet_filter(self, context, packet_filter):
"""Deactivate packet_filter by deleting filter from OFC if exixts."""
pf_status = OperationalStatus.DOWN
- if not self.ofc.exists_ofc_packet_filter(packet_filter['id']):
+ if not self.ofc.exists_ofc_packet_filter(context, packet_filter['id']):
LOG.debug(_("_deactivate_packet_filter(): skip, "
"ofc_packet_filter does not exist."))
else:
try:
- self.ofc.delete_ofc_packet_filter(packet_filter['tenant_id'],
- packet_filter['network_id'],
- packet_filter['id'])
+ self.ofc.delete_ofc_packet_filter(context, packet_filter['id'])
except (nexc.OFCException, nexc.OFCConsistencyBroken) as exc:
reason = _("delete_ofc_packet_filter() failed due to "
"%s") % exc
"""
LOG.debug(_("NECPluginV2.update_packet_filter() called, "
"id=%(id)s packet_filter=%(packet_filter)s ."),
- locals())
+ {'id': id, 'packet_filter': packet_filter})
old_pf = super(NECPluginV2, self).get_packet_filter(context, id)
new_pf = super(NECPluginV2, self).update_packet_filter(context, id,
packet_filter)
"kwargs=%s ."), kwargs)
topic = kwargs['topic']
datapath_id = kwargs['datapath_id']
+ session = rpc_context.session
for p in kwargs.get('port_added', []):
id = p['id']
port = self.plugin.get_port(rpc_context, id)
- if port and ndb.get_portinfo(id):
- ndb.del_portinfo(id)
+ if port and ndb.get_portinfo(session, id):
+ ndb.del_portinfo(session, id)
self.plugin.deactivate_port(rpc_context, port)
- ndb.add_portinfo(id, datapath_id, p['port_no'],
+ ndb.add_portinfo(session, id, datapath_id, p['port_no'],
mac=p.get('mac', ''))
self.plugin.activate_port_if_ready(rpc_context, port)
for id in kwargs.get('port_removed', []):
port = self.plugin.get_port(rpc_context, id)
- if port and ndb.get_portinfo(id):
- ndb.del_portinfo(id)
+ if port and ndb.get_portinfo(session, id):
+ ndb.del_portinfo(session, id)
self.plugin.deactivate_port(rpc_context, port)
# License for the specific language governing permissions and limitations
# under the License.
# @author: Ryota MIBU
+# @author: Akihiro MOTOKI
from abc import ABCMeta, abstractmethod
:param ofc_tenant_id: a OFC tenant ID in which a new network belongs.
:param description: A description of this network.
- :param network_id: A hint of a OFC network ID.
+ :param network_id: A hint of an ID of OFC network.
:returns: ID of the network created at OpenFlow Controller.
+ ID returned must be unique in the OpenFlow Controller.
+ If a network is identified in conjunction with other information
+ such as a tenant ID, such information should be included in the ID.
:raises: quantum.plugin.nec.common.exceptions.OFCException
"""
pass
@abstractmethod
- def update_network(self, ofc_tenant_id, ofc_network_id, description):
- """Update description of specified network.
-
- :raises: quantum.plugin.nec.common.exceptions.OFCException
- """
- pass
-
- @abstractmethod
- def delete_network(self, ofc_tenant_id, ofc_network_id):
+ def delete_network(self, ofc_network_id):
"""Delete a netwrok at OpenFlow Controller.
:raises: quantum.plugin.nec.common.exceptions.OFCException
pass
@abstractmethod
- def create_port(self, ofc_tenant_id, ofc_network_id, portinfo,
+ def create_port(self, ofc_network_id, portinfo,
port_id=None):
- """Create a new port on specified tenant and network at OFC.
+ """Create a new port on specified network at OFC.
:param ofc_network_id: a OFC tenant ID in which a new port belongs.
:param portinfo: An OpenFlow information of this port.
'vlan_id': VLAN ID that a port tagging.
'mac': Mac address.
}
- :param port_id: A hint of a OFC port ID.
+ :param port_id: A hint of an ID of OFC port.
+ ID returned must be unique in the OpenFlow Controller.
+
+ If a port is identified in combination with a network or
+ a tenant, such information should be included in the ID.
:returns: ID of the port created at OpenFlow Controller.
:raises: quantum.plugin.nec.common.exceptions.OFCException
pass
@abstractmethod
- def delete_port(self, ofc_tenant_id, ofc_network_id, ofc_port_id):
+ def delete_port(self, ofc_port_id):
"""Delete a port at OpenFlow Controller.
:raises: quantum.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: quantum 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: quantum context object
+ :param ofc_network_id: ofc_network_id to be converted
+ :param tenant_id: quantum 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: quantum context object
+ :param ofc_port_id: ofc_port_id to be converted
+ :param tenant_id: quantum tenant_id of the port
+ :param network_id: quantum network_id of the port
+ """
+ pass
# License for the specific language governing permissions and limitations
# under the License.
# @author: Ryota MIBU
+# @author: Akihiro MOTOKI
from quantum.plugins.nec.common import config
from quantum.plugins.nec.common import exceptions as nexc
from quantum.plugins.nec.db import api as ndb
-from quantum.plugins.nec.db import models as nmodels
from quantum.plugins.nec import drivers
OFC is identified by a switch ID 'datapath_id' and a port number 'port_no'
of the switch. An ID named as 'ofc_*' is used to identify resource on OFC.
"""
- resource_map = {'ofc_tenant': nmodels.OFCTenant,
- 'ofc_network': nmodels.OFCNetwork,
- 'ofc_port': nmodels.OFCPort,
- 'ofc_packet_filter': nmodels.OFCFilter}
def __init__(self):
self.driver = drivers.get_driver(config.OFC.driver)(config.OFC)
- def _get_ofc_id(self, resource, quantum_id):
- model = self.resource_map[resource]
- ofc_item = ndb.find_ofc_item(model, quantum_id)
- if not ofc_item:
- reason = _("NotFound %(resource)s for "
- "quantum_id=%(quantum_id)s.") % locals()
- raise nexc.OFCConsistencyBroken(reason=reason)
- return ofc_item.id
-
- def _exists_ofc_item(self, resource, quantum_id):
- model = self.resource_map[resource]
- if ndb.find_ofc_item(model, quantum_id):
- return True
- else:
- return False
-
- # Tenant
-
- def create_ofc_tenant(self, tenant_id):
+ def _get_ofc_id(self, context, resource, quantum_id):
+ return ndb.get_ofc_id_lookup_both(context.session,
+ resource, quantum_id)
+
+ def _exists_ofc_item(self, context, resource, quantum_id):
+ return ndb.exists_ofc_item_lookup_both(context.session,
+ resource, quantum_id)
+
+ def _add_ofc_item(self, context, resource, quantum_id, ofc_id):
+ # Ensure a new item is added to the new mapping table
+ ndb.add_ofc_item(context.session, resource, quantum_id, ofc_id)
+
+ def _del_ofc_item(self, context, resource, quantum_id):
+ ndb.del_ofc_item_lookup_both(context.session, resource, quantum_id)
+
+ def create_ofc_tenant(self, context, tenant_id):
desc = "ID=%s at OpenStack." % tenant_id
ofc_tenant_id = self.driver.create_tenant(desc, tenant_id)
- ndb.add_ofc_item(nmodels.OFCTenant, ofc_tenant_id, tenant_id)
+ self._add_ofc_item(context, "ofc_tenant", tenant_id, ofc_tenant_id)
- def exists_ofc_tenant(self, tenant_id):
- return self._exists_ofc_item("ofc_tenant", tenant_id)
+ def exists_ofc_tenant(self, context, tenant_id):
+ return self._exists_ofc_item(context, "ofc_tenant", tenant_id)
- def delete_ofc_tenant(self, tenant_id):
- ofc_tenant_id = self._get_ofc_id("ofc_tenant", 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)
- ndb.del_ofc_item(nmodels.OFCTenant, ofc_tenant_id)
+ self._del_ofc_item(context, "ofc_tenant", tenant_id)
- # Network
-
- def create_ofc_network(self, tenant_id, network_id, network_name=None):
- ofc_tenant_id = self._get_ofc_id("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 Quantum." % (network_id, network_name)
ofc_net_id = self.driver.create_network(ofc_tenant_id, desc,
network_id)
- ndb.add_ofc_item(nmodels.OFCNetwork, ofc_net_id, network_id)
-
- def update_ofc_network(self, tenant_id, network_id, network_name):
- ofc_tenant_id = self._get_ofc_id("ofc_tenant", tenant_id)
- ofc_net_id = self._get_ofc_id("ofc_network", network_id)
-
- desc = "ID=%s Name=%s at Quantum." % (network_id, network_name)
- self.driver.update_network(ofc_tenant_id, ofc_net_id, desc)
-
- def exists_ofc_network(self, network_id):
- return self._exists_ofc_item("ofc_network", network_id)
-
- def delete_ofc_network(self, tenant_id, network_id):
- ofc_tenant_id = self._get_ofc_id("ofc_tenant", tenant_id)
- ofc_net_id = self._get_ofc_id("ofc_network", network_id)
-
- self.driver.delete_network(ofc_tenant_id, ofc_net_id)
- ndb.del_ofc_item(nmodels.OFCNetwork, ofc_net_id)
-
- # Port
-
- def create_ofc_port(self, tenant_id, network_id, port_id):
- ofc_tenant_id = self._get_ofc_id("ofc_tenant", tenant_id)
- ofc_net_id = self._get_ofc_id("ofc_network", network_id)
- portinfo = ndb.get_portinfo(port_id)
+ self._add_ofc_item(context, "ofc_network", network_id, ofc_net_id)
+
+ def exists_ofc_network(self, context, network_id):
+ return self._exists_ofc_item(context, "ofc_network", 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)
- ofc_port_id = self.driver.create_port(ofc_tenant_id, ofc_net_id,
- portinfo, port_id)
- ndb.add_ofc_item(nmodels.OFCPort, ofc_port_id, port_id)
-
- def exists_ofc_port(self, port_id):
- return self._exists_ofc_item("ofc_port", port_id)
-
- def delete_ofc_port(self, tenant_id, network_id, port_id):
- ofc_tenant_id = self._get_ofc_id("ofc_tenant", tenant_id)
- ofc_net_id = self._get_ofc_id("ofc_network", network_id)
- ofc_port_id = self._get_ofc_id("ofc_port", port_id)
+ ofc_port_id = self.driver.create_port(ofc_net_id, portinfo, port_id)
+ self._add_ofc_item(context, "ofc_port", port_id, ofc_port_id)
- self.driver.delete_port(ofc_tenant_id, ofc_net_id, ofc_port_id)
- ndb.del_ofc_item(nmodels.OFCPort, ofc_port_id)
+ def exists_ofc_port(self, context, port_id):
+ return self._exists_ofc_item(context, "ofc_port", port_id)
- # PacketFilter
+ 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, tenant_id, network_id, filter_id,
- filter_dict):
- ofc_tenant_id = self._get_ofc_id("ofc_tenant", tenant_id)
- ofc_net_id = self._get_ofc_id("ofc_network", network_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:
- portinfo = ndb.get_portinfo(in_port_id)
+ portinfo = ndb.get_portinfo(context.session, in_port_id)
if not portinfo:
raise nexc.PortInfoNotFound(id=in_port_id)
- ofc_pf_id = self.driver.create_filter(ofc_tenant_id, ofc_net_id,
+ ofc_pf_id = self.driver.create_filter(ofc_net_id,
filter_dict, portinfo, filter_id)
- ndb.add_ofc_item(nmodels.OFCFilter, ofc_pf_id, filter_id)
+ self._add_ofc_item(context, "ofc_packet_filter", filter_id, ofc_pf_id)
- def exists_ofc_packet_filter(self, filter_id):
- return self._exists_ofc_item("ofc_packet_filter", filter_id)
+ def exists_ofc_packet_filter(self, context, filter_id):
+ return self._exists_ofc_item(context, "ofc_packet_filter", filter_id)
- def delete_ofc_packet_filter(self, tenant_id, network_id, filter_id):
- ofc_tenant_id = self._get_ofc_id("ofc_tenant", tenant_id)
- ofc_net_id = self._get_ofc_id("ofc_network", network_id)
- ofc_pf_id = self._get_ofc_id("ofc_packet_filter", filter_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)
- res = self.driver.delete_filter(ofc_tenant_id, ofc_net_id, ofc_pf_id)
- ndb.del_ofc_item(nmodels.OFCFilter, ofc_pf_id)
+ res = self.driver.delete_filter(ofc_pf_id)
+ self._del_ofc_item(context, "ofc_packet_filter", filter_id)
def create_network(self, ofc_tenant_id, description, network_id=None):
return "ofc-" + network_id[:-4]
- def update_network(self, ofc_tenant_id, ofc_network_id, description):
+ def update_network(self, ofc_network_id, description):
pass
- def delete_network(self, ofc_tenant_id, ofc_network_id):
+ def delete_network(self, ofc_network_id):
pass
- def create_port(self, ofc_tenant_id, ofc_network_id, info, port_id=None):
+ def create_port(self, ofc_network_id, info, port_id=None):
return "ofc-" + port_id[:-4]
- def delete_port(self, ofc_tenant_id, ofc_network_id, ofc_port_id):
+ def delete_port(self, ofc_port_id):
pass
@classmethod
def filter_supported(cls):
return True
- def create_filter(self, ofc_tenant_id, ofc_network_id, filter_dict,
+ def create_filter(self, ofc_network_id, filter_dict,
portinfo=None, filter_id=None):
return "ofc-" + filter_id[:-4]
- def delete_filter(self, ofc_tenant_id, ofc_network_id, ofc_filter_id):
+ def delete_filter(self, ofc_filter_id):
pass
+
+ def convert_ofc_tenant_id(self, context, ofc_tenant_id):
+ return ofc_tenant_id
+
+ def convert_ofc_network_id(self, context, ofc_network_id, tenant_id):
+ return ofc_network_id
+
+ def convert_ofc_port_id(self, context, ofc_port_id, tenant_id, network_id):
+ return ofc_port_id
+
+ def convert_ofc_filter_id(self, context, ofc_filter_id):
+ return ofc_filter_id
import unittest
from quantum.openstack.common import uuidutils
+from quantum.db import api as db_api
from quantum.plugins.nec.common import exceptions as nexc
from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec.db import models as nmodels
-class NECPluginV2DBTest(unittest.TestCase):
+class NECPluginV2DBTestBase(object):
"""Class conisting of NECPluginV2 DB unit tests"""
def setUp(self):
"""Setup for tests"""
ndb.initialize()
+ self.session = db_api.get_session()
def tearDown(self):
"""Tear Down"""
none = uuidutils.generate_uuid()
return ofc_id, quantum_id, none
+ def get_portinfo_random_params(self):
+ """create random parameters for portinfo test"""
+ port_id = uuidutils.generate_uuid()
+ datapath_id = hex(random.randint(0, 0xffffffff))
+ port_no = random.randint(1, 100)
+ vlan_id = random.randint(0, 4095)
+ mac = ':'.join(["%02x" % random.randint(0, 0xff) for x in range(6)])
+ none = uuidutils.generate_uuid()
+ return port_id, datapath_id, port_no, vlan_id, mac, none
+
+
+class NECPluginV2DBTest(NECPluginV2DBTestBase,
+ unittest.TestCase):
+
def testa_add_ofc_item(self):
"""test add OFC item"""
o, q, n = self.get_ofc_item_random_params()
- tenant = ndb.add_ofc_item(nmodels.OFCTenant, o, q)
- self.assertEqual(tenant.id, o)
+ tenant = ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
+ self.assertEqual(tenant.ofc_id, o)
self.assertEqual(tenant.quantum_id, q)
- exception_raised = False
- try:
- ndb.add_ofc_item(nmodels.OFCTenant, o, q)
- except nexc.NECDBException:
- exception_raised = True
- self.assertTrue(exception_raised)
+ self.assertRaises(nexc.NECDBException,
+ ndb.add_ofc_item,
+ self.session, 'ofc_tenant', q, o)
def testb_get_ofc_item(self):
"""test get OFC item"""
o, q, n = self.get_ofc_item_random_params()
- ndb.add_ofc_item(nmodels.OFCTenant, o, q)
- tenant = ndb.get_ofc_item(nmodels.OFCTenant, o)
- self.assertEqual(tenant.id, o)
+ 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(nmodels.OFCTenant, n)
+ tenant_none = ndb.get_ofc_item(self.session, 'ofc_tenant', n)
+ self.assertEqual(None, tenant_none)
+
+ def testb_get_ofc_id(self):
+ """test get OFC d"""
+ 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.assertEqual(None, tenant_none)
+
+ def testb_exists_ofc_item(self):
+ """test get OFC d"""
+ o, q, n = self.get_ofc_item_random_params()
+ ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
+ ret = ndb.exists_ofc_item(self.session, 'ofc_tenant', q)
+ self.assertTrue(ret)
+
+ tenant_none = ndb.get_ofc_item(self.session, 'ofc_tenant', n)
self.assertEqual(None, tenant_none)
def testc_find_ofc_item(self):
"""test find OFC item"""
o, q, n = self.get_ofc_item_random_params()
- ndb.add_ofc_item(nmodels.OFCTenant, o, q)
- tenant = ndb.find_ofc_item(nmodels.OFCTenant, q)
- self.assertEqual(tenant.id, o)
+ 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(nmodels.OFCTenant, n)
+ tenant_none = ndb.find_ofc_item(self.session, 'ofc_tenant', n)
self.assertEqual(None, tenant_none)
def testc_del_ofc_item(self):
"""test delete OFC item"""
o, q, n = self.get_ofc_item_random_params()
- ndb.add_ofc_item(nmodels.OFCTenant, o, q)
- ndb.del_ofc_item(nmodels.OFCTenant, o)
+ ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
+ ndb.del_ofc_item(self.session, 'ofc_tenant', q)
- tenant_none = ndb.get_ofc_item(nmodels.OFCTenant, q)
+ tenant_none = ndb.get_ofc_item(self.session,
+ 'ofc_tenant', q)
self.assertEqual(None, tenant_none)
- tenant_none = ndb.find_ofc_item(nmodels.OFCTenant, q)
+ tenant_none = ndb.find_ofc_item(self.session,
+ 'ofc_tenant', o)
self.assertEqual(None, tenant_none)
- def get_portinfo_random_params(self):
- """create random parameters for portinfo test"""
- port_id = uuidutils.generate_uuid()
- datapath_id = hex(random.randint(0, 0xffffffff))
- port_no = random.randint(1, 100)
- vlan_id = random.randint(0, 4095)
- mac = ':'.join(["%02x" % random.randint(0, 0xff) for x in range(6)])
- none = uuidutils.generate_uuid()
- return port_id, datapath_id, port_no, vlan_id, mac, none
-
def testd_add_portinfo(self):
"""test add portinfo"""
i, d, p, v, m, n = self.get_portinfo_random_params()
- portinfo = ndb.add_portinfo(i, d, p, v, m)
+ portinfo = ndb.add_portinfo(self.session, i, d, p, v, m)
self.assertEqual(portinfo.id, i)
self.assertEqual(portinfo.datapath_id, d)
self.assertEqual(portinfo.port_no, p)
exception_raised = False
try:
- ndb.add_portinfo(i, d, p, v, m)
+ ndb.add_portinfo(self.session, i, d, p, v, m)
except nexc.NECDBException:
exception_raised = True
self.assertTrue(exception_raised)
def teste_get_portinfo(self):
"""test get portinfo"""
i, d, p, v, m, n = self.get_portinfo_random_params()
- ndb.add_portinfo(i, d, p, v, m)
- portinfo = ndb.get_portinfo(i)
+ ndb.add_portinfo(self.session, i, d, p, v, m)
+ portinfo = ndb.get_portinfo(self.session, i)
self.assertEqual(portinfo.id, i)
self.assertEqual(portinfo.datapath_id, d)
self.assertEqual(portinfo.port_no, p)
self.assertEqual(portinfo.vlan_id, v)
self.assertEqual(portinfo.mac, m)
- portinfo_none = ndb.get_portinfo(n)
+ portinfo_none = ndb.get_portinfo(self.session, n)
self.assertEqual(None, portinfo_none)
def testf_del_portinfo(self):
"""test delete portinfo"""
i, d, p, v, m, n = self.get_portinfo_random_params()
- ndb.add_portinfo(i, d, p, v, m)
- portinfo = ndb.get_portinfo(i)
+ ndb.add_portinfo(self.session, i, d, p, v, m)
+ portinfo = ndb.get_portinfo(self.session, i)
self.assertEqual(portinfo.id, i)
- ndb.del_portinfo(i)
- portinfo_none = ndb.get_portinfo(i)
+ ndb.del_portinfo(self.session, i)
+ portinfo_none = ndb.get_portinfo(self.session, i)
self.assertEqual(None, portinfo_none)
+
+
+class NECPluginV2DBOldMappingTest(NECPluginV2DBTestBase,
+ unittest.TestCase):
+ """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.assertEqual(ret, None)
+
+ 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.assertEqual(ret, None)
+ 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.assertEqual(ret, None)
+ ret = ndb.get_ofc_item(self.session, 'ofc_tenant', q, self.OLD)
+ self.assertEqual(ret, None)
+
+ 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)
import unittest
+from quantum import context
from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import config
from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec import ofc_manager
-class OFCManagerTest(unittest.TestCase):
+class OFCManagerTestBase(object):
"""Class conisting of OFCManager unit tests"""
def setUp(self):
config.CONF.set_override('driver', driver, 'OFC')
ndb.initialize()
self.ofc = ofc_manager.OFCManager()
+ self.ctx = context.get_admin_context()
def tearDown(self):
ndb.clear_db()
none = uuidutils.generate_uuid()
return tenant, network, port, _filter, none
+
+class OFCManagerTest(OFCManagerTestBase, unittest.TestCase):
def testa_create_ofc_tenant(self):
"""test create ofc_tenant"""
t, n, p, f, none = self.get_random_params()
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCTenant, t))
- self.ofc.create_ofc_tenant(t)
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCTenant, t))
- tenant = ndb.find_ofc_item(nmodels.OFCTenant, t)
- self.assertEqual(tenant.id, "ofc-" + t[:-4])
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session, 'ofc_tenant', t))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session, 'ofc_tenant', t))
+ tenant = ndb.get_ofc_item(self.ctx.session, 'ofc_tenant', t)
+ self.assertEqual(tenant.ofc_id, "ofc-" + t[:-4])
def testb_exists_ofc_tenant(self):
"""test exists_ofc_tenant"""
t, n, p, f, none = self.get_random_params()
- self.assertFalse(self.ofc.exists_ofc_tenant(t))
- self.ofc.create_ofc_tenant(t)
- self.assertTrue(self.ofc.exists_ofc_tenant(t))
+ self.assertFalse(self.ofc.exists_ofc_tenant(self.ctx, t))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.assertTrue(self.ofc.exists_ofc_tenant(self.ctx, t))
def testc_delete_ofc_tenant(self):
"""test delete ofc_tenant"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCTenant, t))
- self.ofc.delete_ofc_tenant(t)
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCTenant, t))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session, 'ofc_tenant', t))
+ self.ofc.delete_ofc_tenant(self.ctx, t)
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session, 'ofc_tenant', t))
def testd_create_ofc_network(self):
"""test create ofc_network"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCNetwork, n))
- self.ofc.create_ofc_network(t, n)
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCNetwork, n))
- network = ndb.find_ofc_item(nmodels.OFCNetwork, n)
- self.assertEqual(network.id, "ofc-" + n[:-4])
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session, 'ofc_network', n))
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session, 'ofc_network', n))
+ network = ndb.get_ofc_item(self.ctx.session, 'ofc_network', n)
+ self.assertEqual(network.ofc_id, "ofc-" + n[:-4])
def teste_exists_ofc_network(self):
"""test exists_ofc_network"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.assertFalse(self.ofc.exists_ofc_network(n))
- self.ofc.create_ofc_network(t, n)
- self.assertTrue(self.ofc.exists_ofc_network(n))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.assertFalse(self.ofc.exists_ofc_network(self.ctx, n))
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ self.assertTrue(self.ofc.exists_ofc_network(self.ctx, n))
def testf_delete_ofc_network(self):
"""test delete ofc_network"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.ofc.create_ofc_network(t, n)
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCNetwork, n))
- self.ofc.delete_ofc_network(t, n)
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCNetwork, n))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session, 'ofc_network', n))
+ self.ofc.delete_ofc_network(self.ctx, n, {'tenant_id': t})
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session, 'ofc_network', n))
def testg_create_ofc_port(self):
"""test create ofc_port"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.ofc.create_ofc_network(t, n)
- ndb.add_portinfo(p, "0xabc", 1, 65535, "00:11:22:33:44:55")
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCPort, p))
- self.ofc.create_ofc_port(t, n, p)
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCPort, p))
- port = ndb.find_ofc_item(nmodels.OFCPort, p)
- self.assertEqual(port.id, "ofc-" + p[:-4])
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ ndb.add_portinfo(self.ctx.session, p, "0xabc", 1, 65535,
+ "00:11:22:33:44:55")
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session, 'ofc_port', p))
+ port = {'tenant_id': t, 'network_id': n}
+ self.ofc.create_ofc_port(self.ctx, p, port)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session, 'ofc_port', p))
+ port = ndb.get_ofc_item(self.ctx.session, 'ofc_port', p)
+ self.assertEqual(port.ofc_id, "ofc-" + p[:-4])
def testh_exists_ofc_port(self):
"""test exists_ofc_port"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.ofc.create_ofc_network(t, n)
- ndb.add_portinfo(p, "0xabc", 2, 65535, "00:12:22:33:44:55")
- self.assertFalse(self.ofc.exists_ofc_port(p))
- self.ofc.create_ofc_port(t, n, p)
- self.assertTrue(self.ofc.exists_ofc_port(p))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ ndb.add_portinfo(self.ctx.session, p, "0xabc", 2, 65535,
+ "00:12:22:33:44:55")
+ self.assertFalse(self.ofc.exists_ofc_port(self.ctx, p))
+ port = {'tenant_id': t, 'network_id': n}
+ self.ofc.create_ofc_port(self.ctx, p, port)
+ self.assertTrue(self.ofc.exists_ofc_port(self.ctx, p))
def testi_delete_ofc_port(self):
"""test delete ofc_port"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.ofc.create_ofc_network(t, n)
- ndb.add_portinfo(p, "0xabc", 3, 65535, "00:13:22:33:44:55")
- self.ofc.create_ofc_port(t, n, p)
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCPort, p))
- self.ofc.delete_ofc_port(t, n, p)
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCPort, p))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ ndb.add_portinfo(self.ctx.session, p, "0xabc", 3, 65535,
+ "00:13:22:33:44:55")
+ port = {'tenant_id': t, 'network_id': n}
+ self.ofc.create_ofc_port(self.ctx, p, port)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session, 'ofc_port', p))
+ self.ofc.delete_ofc_port(self.ctx, p, port)
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session, 'ofc_port', p))
def testj_create_ofc_packet_filter(self):
"""test create ofc_filter"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.ofc.create_ofc_network(t, n)
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCFilter, f))
- self.ofc.create_ofc_packet_filter(t, n, f, {})
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCFilter, f))
- _filter = ndb.find_ofc_item(nmodels.OFCFilter, f)
- self.assertEqual(_filter.id, "ofc-" + f[:-4])
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session,
+ 'ofc_packet_filter', f))
+ pf = {'tenant_id': t, 'network_id': n}
+ self.ofc.create_ofc_packet_filter(self.ctx, f, pf)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session,
+ 'ofc_packet_filter', f))
+ _filter = ndb.get_ofc_item(self.ctx.session, 'ofc_packet_filter', f)
+ self.assertEqual(_filter.ofc_id, "ofc-" + f[:-4])
def testk_exists_ofc_packet_filter(self):
"""test exists_ofc_packet_filter"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.ofc.create_ofc_network(t, n)
- self.assertFalse(self.ofc.exists_ofc_packet_filter(f))
- self.ofc.create_ofc_packet_filter(t, n, f, {})
- self.assertTrue(self.ofc.exists_ofc_packet_filter(f))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ self.assertFalse(self.ofc.exists_ofc_packet_filter(self.ctx, f))
+ pf = {'tenant_id': t, 'network_id': n}
+ self.ofc.create_ofc_packet_filter(self.ctx, f, pf)
+ self.assertTrue(self.ofc.exists_ofc_packet_filter(self.ctx, f))
def testl_delete_ofc_packet_filter(self):
"""test delete ofc_filter"""
t, n, p, f, none = self.get_random_params()
- self.ofc.create_ofc_tenant(t)
- self.ofc.create_ofc_network(t, n)
- self.ofc.create_ofc_packet_filter(t, n, f, {})
- self.assertTrue(ndb.find_ofc_item(nmodels.OFCFilter, f))
- self.ofc.delete_ofc_packet_filter(t, n, f)
- self.assertFalse(ndb.find_ofc_item(nmodels.OFCFilter, f))
+ self.ofc.create_ofc_tenant(self.ctx, t)
+ self.ofc.create_ofc_network(self.ctx, t, n)
+ pf = {'tenant_id': t, 'network_id': n}
+ self.ofc.create_ofc_packet_filter(self.ctx, f, pf)
+ self.assertTrue(ndb.get_ofc_item(self.ctx.session,
+ 'ofc_packet_filter', f))
+ self.ofc.delete_ofc_packet_filter(self.ctx, f)
+ self.assertFalse(ndb.get_ofc_item(self.ctx.session,
+ 'ofc_packet_filter', f))
+
+
+class OFCManagerTestWithOldMapping(OFCManagerTestBase, unittest.TestCase):
+
+ 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))
# under the License.
# @author: Ryota MIBU
+import random
+import string
+
import mox
import unittest
+from quantum import context
from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import ofc_client as ofc
+from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec.db import models as nmodels
from quantum.plugins.nec import drivers
return "ofc-%s" % id
-class PFCDriverTestBase(unittest.TestCase):
+class PFCDriverTestBase():
+
+ driver = 'quantum.plugins.nec.drivers.pfc.PFCDriverBase'
def setUp(self):
self.mox = mox.Mox()
- self.driver = drivers.get_driver("pfc")(TestConfig)
+ self.driver = drivers.get_driver(self.driver)(TestConfig)
self.mox.StubOutWithMock(ofc.OFCClient, 'do_request')
def tearDown(self):
mac="11:22:33:44:55:66")
return tenant_id, network_id, portinfo
- def testa_create_tenant(self):
- t, n, p = self.get_ofc_item_random_params()
- description = "desc of %s" % t
+ def _generate_ofc_tenant_id(self, tenant_id):
+ fields = tenant_id.split('-')
+ # Strip 1st character (UUID version) of 3rd field
+ fields[2] = fields[2][1:]
+ return ''.join(fields)
+
+ def get_ofc_description(self, desc):
+ """OFC description consists of [A-Za-z0-9_]."""
+ return desc.replace('-', '_').replace(' ', '_')
+ def _create_tenant(self, t, ofc_t, post_id=False, post_desc=False):
+ tenant_path = '/tenants/%s' % ofc_t
path = "/tenants"
- body = {'id': t, 'description': description}
- tenant = {'id': _ofc(t)}
- ofc.OFCClient.do_request("POST", path, body=body).AndReturn(tenant)
+ description = "desc of %s" % t
+ body = {}
+ if post_desc:
+ ofc_description = self.get_ofc_description(description)
+ body['description'] = ofc_description
+ if post_id:
+ body['id'] = ofc_t
+ ofc.OFCClient.do_request("POST", path, body=body)
+ else:
+ ofc.OFCClient.do_request("POST", path, body=body).\
+ AndReturn({'id': ofc_t})
self.mox.ReplayAll()
ret = self.driver.create_tenant(description, t)
self.mox.VerifyAll()
- self.assertEqual(ret, _ofc(t))
+ self.assertEqual(ret, tenant_path)
- def testb_update_tenant(self):
+ def testa_create_tenant(self):
t, n, p = self.get_ofc_item_random_params()
- description = "new desc of %s" % t
-
- path = "/tenants/%s" % _ofc(t)
- body = {'description': description}
- ofc.OFCClient.do_request("PUT", path, body=body)
- self.mox.ReplayAll()
-
- self.driver.update_tenant(_ofc(t), description)
- self.mox.VerifyAll()
+ ofc_t = self._generate_ofc_tenant_id(t)
+ self._create_tenant(t, ofc_t, post_id=True)
def testc_delete_tenant(self):
t, n, p = self.get_ofc_item_random_params()
ofc.OFCClient.do_request("DELETE", path)
self.mox.ReplayAll()
- self.driver.delete_tenant(_ofc(t))
+ self.driver.delete_tenant(path)
self.mox.VerifyAll()
def testd_create_network(self):
t, n, p = self.get_ofc_item_random_params()
description = "desc of %s" % n
+ ofc_description = self.get_ofc_description(description)
- path = "/tenants/%s/networks" % _ofc(t)
- body = {'id': n, 'description': description}
+ tenant_path = "/tenants/%s" % _ofc(t)
+ post_path = "%s/networks" % tenant_path
+ body = {'description': ofc_description}
network = {'id': _ofc(n)}
- ofc.OFCClient.do_request("POST", path, body=body).AndReturn(network)
- self.mox.ReplayAll()
-
- ret = self.driver.create_network(_ofc(t), description, n)
- self.mox.VerifyAll()
- self.assertEqual(ret, _ofc(n))
-
- def teste_update_network(self):
- t, n, p = self.get_ofc_item_random_params()
- description = "desc of %s" % n
-
- path = "/tenants/%s/networks/%s" % (_ofc(t), _ofc(n))
- body = {'description': description}
- ofc.OFCClient.do_request("PUT", path, body=body)
+ ofc.OFCClient.do_request("POST", post_path, body=body).\
+ AndReturn(network)
self.mox.ReplayAll()
- self.driver.update_network(_ofc(t), _ofc(n), description)
+ ret = self.driver.create_network(tenant_path, description, n)
self.mox.VerifyAll()
+ net_path = "/tenants/%s/networks/%s" % (_ofc(t), _ofc(n))
+ self.assertEqual(ret, net_path)
def testf_delete_network(self):
t, n, p = self.get_ofc_item_random_params()
- path = "/tenants/%s/networks/%s" % (_ofc(t), _ofc(n))
- ofc.OFCClient.do_request("DELETE", path)
+ net_path = "/tenants/%s/networks/%s" % (_ofc(t), _ofc(n))
+ ofc.OFCClient.do_request("DELETE", net_path)
self.mox.ReplayAll()
- self.driver.delete_network(_ofc(t), _ofc(n))
+ self.driver.delete_network(net_path)
self.mox.VerifyAll()
def testg_create_port(self):
t, n, p = self.get_ofc_item_random_params()
- path = "/tenants/%s/networks/%s/ports" % (_ofc(t), _ofc(n))
- body = {'id': p.id,
- 'datapath_id': p.datapath_id,
+ net_path = "/tenants/%s/networks/%s" % (_ofc(t), _ofc(n))
+ post_path = "%s/ports" % net_path
+ port_path = "/tenants/%s/networks/%s/ports/%s" % (_ofc(t), _ofc(n),
+ _ofc(p.id))
+ body = {'datapath_id': p.datapath_id,
'port': str(p.port_no),
'vid': str(p.vlan_id)}
port = {'id': _ofc(p.id)}
- ofc.OFCClient.do_request("POST", path, body=body).AndReturn(port)
+ ofc.OFCClient.do_request("POST", post_path, body=body).AndReturn(port)
self.mox.ReplayAll()
- ret = self.driver.create_port(_ofc(t), _ofc(n), p, p.id)
+ ret = self.driver.create_port(net_path, p, p.id)
self.mox.VerifyAll()
- self.assertEqual(ret, _ofc(p.id))
+ self.assertEqual(ret, port_path)
def testh_delete_port(self):
t, n, p = self.get_ofc_item_random_params()
- path = "/tenants/%s/networks/%s/ports/%s" % (_ofc(t), _ofc(n),
- _ofc(p.id))
- ofc.OFCClient.do_request("DELETE", path)
+ port_path = "/tenants/%s/networks/%s/ports/%s" % (_ofc(t), _ofc(n),
+ _ofc(p.id))
+ ofc.OFCClient.do_request("DELETE", port_path)
+ self.mox.ReplayAll()
+
+ self.driver.delete_port(port_path)
+ self.mox.VerifyAll()
+
+
+class PFCDriverBaseTest(PFCDriverTestBase, unittest.TestCase):
+ pass
+
+
+class PFCV3DriverTest(PFCDriverTestBase, unittest.TestCase):
+ driver = 'pfc_v3'
+
+ def testa_create_tenant(self):
+ t, n, p = self.get_ofc_item_random_params()
+ self.mox.ReplayAll()
+
+ ret = self.driver.create_tenant('dummy_desc', t)
+ self.mox.VerifyAll()
+
+ ofc_t_path = "/tenants/" + self._generate_ofc_tenant_id(t)
+ self.assertEqual(ofc_t_path, ret)
+
+ def testc_delete_tenant(self):
+ pass
+
+
+class PFCV4DriverTest(PFCDriverTestBase, unittest.TestCase):
+ driver = 'pfc_v4'
+
+
+class PFCDriverStringTest(unittest.TestCase):
+
+ driver = 'quantum.plugins.nec.drivers.pfc.PFCDriverBase'
+
+ def setUp(self):
+ self.driver = drivers.get_driver(self.driver)(TestConfig)
+
+ def tearDown(self):
+ pass
+
+ def test_generate_pfc_id_uuid(self):
+ id_str = uuidutils.generate_uuid()
+ exp_str = (id_str[:14] + id_str[15:]).replace('-', '')[:31]
+
+ ret_str = self.driver._generate_pfc_id(id_str)
+ self.assertEqual(exp_str, ret_str)
+
+ def test_generate_pfc_id_uuid_no_hyphen(self):
+ # Keystone tenant_id style uuid
+ id_str = uuidutils.generate_uuid()
+ id_no_hyphen = id_str.replace('-', '')
+ exp_str = (id_str[:14] + id_str[15:]).replace('-', '')[:31]
+
+ ret_str = self.driver._generate_pfc_id(id_no_hyphen)
+ self.assertEqual(exp_str, ret_str)
+
+ def test_generate_pfc_id_string(self):
+ id_str = uuidutils.generate_uuid() + 'x'
+ exp_str = id_str[:31].replace('-', '_')
+
+ ret_str = self.driver._generate_pfc_id(id_str)
+ self.assertEqual(exp_str, ret_str)
+
+ def test_generate_pfc_desc(self):
+ random_list = [random.choice(string.printable) for x in range(128)]
+ random_str = ''.join(random_list)
+
+ accept_letters = string.letters + string.digits
+ exp_list = [x if x in accept_letters else '_' for x in random_list]
+ exp_str = ''.join(exp_list)[:127]
+
+ ret_str = self.driver._generate_pfc_description(random_str)
+ self.assertEqual(exp_str, ret_str)
+
+
+class PFCIdConvertTest(unittest.TestCase):
+ driver = 'quantum.plugins.nec.drivers.pfc.PFCDriverBase'
+
+ def setUp(self):
+ self.mox = mox.Mox()
+ self.driver = drivers.get_driver(self.driver)(TestConfig)
+ self.ctx = self.mox.CreateMock(context.Context)
+ self.ctx.session = "session"
+ self.mox.StubOutWithMock(ndb, 'get_ofc_id_lookup_both')
+
+ def tearDown(self):
+ self.mox.UnsetStubs()
+
+ 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)
+ print ofc_t_id
+ 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)
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_id)
+ self.mox.ReplayAll()
+
+ 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.mox.VerifyAll()
+
+ 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
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_path)
self.mox.ReplayAll()
- self.driver.delete_port(_ofc(t), _ofc(n), _ofc(p.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.mox.VerifyAll()
+
+ 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)
+
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_id)
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_id)
+ self.mox.ReplayAll()
+
+ 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.mox.VerifyAll()
+
+ 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
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_id)
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_tenant', t_id).AndReturn(ofc_t_path)
+ self.mox.ReplayAll()
+
+ 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.mox.VerifyAll()
+
+ 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})
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_path)
+ self.mox.ReplayAll()
+
+ 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.mox.VerifyAll()
+
+ 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)
import mox
import unittest
+from quantum import context
from quantum.openstack.common import uuidutils
from quantum.plugins.nec.common import ofc_client
+from quantum.plugins.nec.db import api as ndb
from quantum.plugins.nec.db import models as nmodels
from quantum.plugins.nec import drivers
ofc_client.OFCClient.do_request("POST", "/networks", body=body)
self.mox.ReplayAll()
- self.driver.create_network(t, description, n)
- self.mox.VerifyAll()
-
- def testb_update_network(self):
- t, n, p = self.get_ofc_item_random_params()
- description = "desc of %s" % n
-
- body = {'description': description}
- ofc_client.OFCClient.do_request("PUT", "/networks/%s" % n, body=body)
- self.mox.ReplayAll()
-
- self.driver.update_network(t, n, description)
+ ret = self.driver.create_network(t, description, n)
self.mox.VerifyAll()
+ self.assertEqual(ret, '/networks/%s' % n)
def testc_delete_network(self):
t, n, p = self.get_ofc_item_random_params()
- ofc_client.OFCClient.do_request("DELETE", "/networks/%s" % n)
+ net_path = "/networks/%s" % n
+ ofc_client.OFCClient.do_request("DELETE", net_path)
self.mox.ReplayAll()
- self.driver.delete_network(t, n)
+ self.driver.delete_network(net_path)
self.mox.VerifyAll()
driver_name = "trema_port"
def testd_create_port(self):
- t, n, p = self.get_ofc_item_random_params()
+ _t, n, p = self.get_ofc_item_random_params()
+ net_path = "/networks/%s" % n
body = {'id': p.id,
'datapath_id': p.datapath_id,
'port': str(p.port_no),
"/networks/%s/ports" % n, body=body)
self.mox.ReplayAll()
- self.driver.create_port(t, n, p, p.id)
+ ret = self.driver.create_port(net_path, p, p.id)
self.mox.VerifyAll()
+ self.assertEqual(ret, '/networks/%s/ports/%s' % (n, p.id))
def testd_delete_port(self):
t, n, p = self.get_ofc_item_random_params()
- ofc_client.OFCClient.do_request("DELETE",
- "/networks/%s/ports/%s" % (n, p.id))
+ p_path = "/networks/%s/ports/%s" % (n, p.id)
+ ofc_client.OFCClient.do_request("DELETE", p_path)
self.mox.ReplayAll()
- self.driver.delete_port(t, n, p.id)
+ self.driver.delete_port(p_path)
self.mox.VerifyAll()
t, n, p = self.get_ofc_item_random_params()
dummy_port = "dummy-%s" % p.id
+ net_path = "/networks/%s" % n
path_1 = "/networks/%s/ports" % n
body_1 = {'id': dummy_port,
'datapath_id': p.datapath_id,
ofc_client.OFCClient.do_request("DELETE", path_3)
self.mox.ReplayAll()
- self.driver.create_port(t, n, p, p.id)
+ ret = self.driver.create_port(net_path, p, p.id)
self.mox.VerifyAll()
+ port_path = "/networks/%s/ports/%s/attachments/%s" % (n, dummy_port,
+ p.id)
+ self.assertEqual(ret, port_path)
def testd_delete_port(self):
t, n, p = self.get_ofc_item_random_params()
ofc_client.OFCClient.do_request("DELETE", path)
self.mox.ReplayAll()
- self.driver.delete_port(t, n, p.id)
+ self.driver.delete_port(path)
self.mox.VerifyAll()
def testd_create_port(self):
t, n, p = self.get_ofc_item_random_params()
+ net_path = "/networks/%s" % n
path = "/networks/%s/attachments" % n
body = {'id': p.id, 'mac': p.mac}
ofc_client.OFCClient.do_request("POST", path, body=body)
self.mox.ReplayAll()
- self.driver.create_port(t, n, p, p.id)
+ ret = self.driver.create_port(net_path, p, p.id)
self.mox.VerifyAll()
+ self.assertEqual(ret, '/networks/%s/attachments/%s' % (n, p.id))
def testd_delete_port(self):
t, n, p = self.get_ofc_item_random_params()
ofc_client.OFCClient.do_request("DELETE", path)
self.mox.ReplayAll()
- self.driver.delete_port(t, n, p.id)
+ self.driver.delete_port(path)
self.mox.VerifyAll()
def testa_create_filter(self):
t, n, p, f = self.get_ofc_item_random_params()
+ net_path = "/networks/%s" % n
ofp_wildcards = 'dl_vlan,dl_vlan_pcp,nw_tos,dl_dst,' + \
'nw_src:32,nw_dst:32,tp_src,tp_dst'
body = {'id': f.id,
ofc_client.OFCClient.do_request("POST", "/filters", body=body)
self.mox.ReplayAll()
- self.driver.create_filter(t, n, f, p, f.id)
+ ret = self.driver.create_filter(net_path, f, p, f.id)
self.mox.VerifyAll()
+ self.assertEqual(ret, '/filters/%s' % f.id)
def testb_delete_filter(self):
t, n, p, f = self.get_ofc_item_random_params()
- ofc_client.OFCClient.do_request("DELETE", "/filters/%s" % f.id)
+ f_path = "/filters/%s" % f.id
+ ofc_client.OFCClient.do_request("DELETE", f_path)
self.mox.ReplayAll()
- self.driver.delete_filter(t, n, f.id)
+ self.driver.delete_filter(f_path)
self.mox.VerifyAll()
+
+
+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(unittest.TestCase):
+ driver_name = 'trema'
+
+ def setUp(self):
+ self.driver = drivers.get_driver(self.driver_name)(TestConfig)
+ self.mox = mox.Mox()
+ self.ctx = self.mox.CreateMock(context.Context)
+
+ def tearDown(self):
+ self.mox.UnsetStubs()
+
+ 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
+ ret = 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(object):
+ def setUp(self):
+ self.mox = mox.Mox()
+ self.driver = drivers.get_driver(self.driver_name)(TestConfig)
+ self.ctx = self.mox.CreateMock(context.Context)
+ self.ctx.session = "session"
+ self.mox.StubOutWithMock(ndb, 'get_ofc_id_lookup_both')
+
+ def tearDown(self):
+ self.mox.UnsetStubs()
+
+ 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)
+
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_id)
+ self.mox.ReplayAll()
+
+ 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.mox.VerifyAll()
+
+ 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
+ ndb.get_ofc_id_lookup_both(
+ self.ctx.session, 'ofc_network', n_id).AndReturn(ofc_n_path)
+ self.mox.ReplayAll()
+
+ 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}
+ print 'exp=', exp
+ print 'ret=', ret
+ self.assertEqual(ret, exp)
+ self.mox.VerifyAll()
+
+ 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, unittest.TestCase):
+ 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, unittest.TestCase):
+ 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, unittest.TestCase):
+ 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