L2Gateway DB Models for OVSDB Hardware_vtep Schema

These MYSQL DB models are introduced in order to
maintain a replica of OVSDB tables in Neutron DB

Change-Id: I013dc63925c77eec7c4499a2fb8d1484ca261f35
Author: Manjunath Patil <mpatil@hp.com>
Co-Authored-By: Phani Pawan <ppawan@hp.com>
stable/ocata
Manjunath Patil 8 years ago committed by armando-migliaccio
parent 30065ff090
commit fce1ea47d6
  1. 0
      networking_l2gw/db/__init__.py
  2. 0
      networking_l2gw/db/migration/alembic_migrations/__init__.py
  3. 111
      networking_l2gw/db/migration/alembic_migrations/versions/54c9c8fe22bf_db_models_for_ovsdb_hardware_vtep_schema.py
  4. 1
      networking_l2gw/db/migration/alembic_migrations/versions/HEAD
  5. 0
      networking_l2gw/services/l2gateway/ovsdb/__init__.py
  6. 368
      networking_l2gw/services/l2gateway/ovsdb/lib.py
  7. 100
      networking_l2gw/services/l2gateway/ovsdb/models.py
  8. 0
      networking_l2gw/tests/unit/services/l2gateway/ovsdb/__init__.py
  9. 455
      networking_l2gw/tests/unit/services/l2gateway/ovsdb/test_lib.py

@ -0,0 +1,111 @@
# Copyright 2015 OpenStack Foundation
#
# 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.
#
"""DB_Models_for_OVSDB_Hardware_VTEP_Schema
Revision ID: 54c9c8fe22bf
Revises: 42438454c556
Create Date: 2015-01-27 02:05:21.599215
"""
# revision identifiers, used by Alembic.
revision = '54c9c8fe22bf'
down_revision = '42438454c556'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.create_table('physical_locators',
sa.Column('dst_ip', sa.String(length=64), nullable=True),
sa.Column('uuid', sa.String(length=36), nullable=False),
sa.Column('ovsdb_identifier', sa.String(length=64),
nullable=False),
sa.PrimaryKeyConstraint('uuid', 'ovsdb_identifier'))
op.create_table('physical_switches',
sa.Column('uuid', sa.String(length=36), nullable=False),
sa.Column('name', sa.String(length=255), nullable=True),
sa.Column('tunnel_ip', sa.String(length=64),
nullable=True),
sa.Column('ovsdb_identifier', sa.String(length=64),
nullable=False),
sa.PrimaryKeyConstraint('uuid', 'ovsdb_identifier'))
op.create_table('physical_ports',
sa.Column('name', sa.String(length=255), nullable=True),
sa.Column('uuid', sa.String(length=36), nullable=False),
sa.Column('physical_switch_id', sa.String(length=36),
nullable=True),
sa.Column('ovsdb_identifier', sa.String(length=64),
nullable=False),
sa.PrimaryKeyConstraint('uuid', 'ovsdb_identifier'))
op.create_table('logical_switches',
sa.Column('uuid', sa.String(length=36), nullable=False),
sa.Column('name', sa.String(length=255), nullable=True),
sa.Column('key', sa.Integer(), nullable=True),
sa.Column('ovsdb_identifier', sa.String(length=64),
nullable=False),
sa.PrimaryKeyConstraint('uuid', 'ovsdb_identifier'))
op.create_table('ucast_macs_locals',
sa.Column('uuid', sa.String(length=36), nullable=False),
sa.Column('mac', sa.String(length=32), nullable=True),
sa.Column('logical_switch_id', sa.String(length=36),
nullable=True),
sa.Column('physical_locator_id', sa.String(length=36),
nullable=True),
sa.Column('ip_address', sa.String(length=64),
nullable=True),
sa.Column('ovsdb_identifier', sa.String(length=64),
nullable=False),
sa.PrimaryKeyConstraint('uuid', 'ovsdb_identifier'))
op.create_table('ucast_macs_remotes',
sa.Column('uuid', sa.String(length=36), nullable=False),
sa.Column('mac', sa.String(length=32), nullable=True),
sa.Column('logical_switch_id', sa.String(length=36),
nullable=True),
sa.Column('physical_locator_id', sa.String(length=36),
nullable=True),
sa.Column('ip_address', sa.String(length=64),
nullable=True),
sa.Column('ovsdb_identifier', sa.String(length=64),
nullable=False),
sa.PrimaryKeyConstraint('uuid', 'ovsdb_identifier'))
op.create_table('vlan_bindings',
sa.Column('port_uuid', sa.String(length=36),
nullable=False),
sa.Column('vlan', sa.Integer(), nullable=False),
sa.Column('logical_switch_uuid', sa.String(length=36),
nullable=False),
sa.Column('ovsdb_identifier', sa.String(length=64),
nullable=False),
sa.PrimaryKeyConstraint('port_uuid', 'ovsdb_identifier',
'vlan', 'logical_switch_uuid'))
def downgrade():
op.drop_table('physical_locators')
op.drop_table('physical_switches')
op.drop_table('physical_ports')
op.drop_table('logical_switches')
op.drop_table('ucast_macs_locals')
op.drop_table('ucast_macs_remotes')
op.drop_table('vlan_bindings')

@ -0,0 +1,368 @@
# Copyright (c) 2015 OpenStack Foundation
#
# 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.
from sqlalchemy.orm import exc
from neutron.openstack.common import log as logging
from networking_l2gw.services.l2gateway.ovsdb import models
LOG = logging.getLogger(__name__)
def add_vlan_binding(context, record_dict):
"""Insert a vlan binding of a given physical port."""
session = context.session
with session.begin(subtransactions=True):
binding = models.VlanBindings(
port_uuid=record_dict['port_uuid'],
vlan=record_dict['vlan'],
logical_switch_uuid=record_dict['logical_switch_uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
session.add(binding)
def delete_vlan_binding(context, record_dict):
"""Delete vlan bindings of a given physical port."""
session = context.session
with session.begin(subtransactions=True):
if(record_dict['vlan'] and record_dict['logical_switch_uuid']):
session.query(models.VlanBindings).filter_by(
port_uuid=record_dict['port_uuid'], vlan=record_dict['vlan'],
logical_switch_uuid=record_dict['logical_switch_uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).delete()
def add_physical_locator(context, record_dict):
"""Insert a new physical locator."""
session = context.session
with session.begin(subtransactions=True):
locator = models.PhysicalLocators(
uuid=record_dict['uuid'],
dst_ip=record_dict['dst_ip'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
session.add(locator)
def delete_physical_locator(context, record_dict):
"""Delete physical locator that matches the supplied uuid."""
session = context.session
with session.begin(subtransactions=True):
if(record_dict['uuid']):
session.query(models.PhysicalLocators).filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).delete()
def add_physical_switch(context, record_dict):
"""Insert a new physical switch."""
session = context.session
with session.begin(subtransactions=True):
physical_switch = models.PhysicalSwitches(
uuid=record_dict['uuid'],
name=record_dict['name'],
tunnel_ip=record_dict['tunnel_ip'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
session.add(physical_switch)
def delete_physical_switch(context, record_dict):
"""Delete physical switch that matches the supplied uuid."""
session = context.session
with session.begin(subtransactions=True):
if(record_dict['uuid']):
session.query(models.PhysicalSwitches).filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).delete()
def add_logical_switch(context, record_dict):
"""Insert a new logical switch."""
session = context.session
with session.begin(subtransactions=True):
logical_switch = models.LogicalSwitches(
uuid=record_dict['uuid'],
name=record_dict['name'],
key=record_dict['key'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
session.add(logical_switch)
def delete_logical_switch(context, record_dict):
"""delete logical switch that matches the supplied uuid."""
session = context.session
with session.begin(subtransactions=True):
if(record_dict['uuid']):
session.query(models.LogicalSwitches).filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).delete()
def add_physical_port(context, record_dict):
"""Insert a new physical port."""
session = context.session
with session.begin(subtransactions=True):
physical_port = models.PhysicalPorts(
uuid=record_dict['uuid'],
name=record_dict['name'],
physical_switch_id=record_dict['physical_switch_id'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
session.add(physical_port)
def delete_physical_port(context, record_dict):
"""Delete physical port that matches the supplied uuid."""
session = context.session
with session.begin(subtransactions=True):
if(record_dict['uuid']):
session.query(models.PhysicalPorts).filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).delete()
def add_ucast_mac_local(context, record_dict):
"""Insert a new ucast mac local."""
session = context.session
with session.begin(subtransactions=True):
ucast_mac_local = models.UcastMacsLocals(
uuid=record_dict['uuid'],
mac=record_dict['mac'],
logical_switch_id=record_dict['logical_switch_id'],
physical_locator_id=record_dict['physical_locator_id'],
ip_address=record_dict['ip_address'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
session.add(ucast_mac_local)
def delete_ucast_mac_local(context, record_dict):
"""Delete ucast mac local that matches the supplied uuid."""
session = context.session
with session.begin(subtransactions=True):
if(record_dict['uuid']):
session.query(models.UcastMacsLocals).filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).delete()
def add_ucast_mac_remote(context, record_dict):
"""Insert a new ucast mac remote."""
session = context.session
with session.begin(subtransactions=True):
ucast_mac_remote = models.UcastMacsRemotes(
uuid=record_dict['uuid'],
mac=record_dict['mac'],
logical_switch_id=record_dict['logical_switch_id'],
physical_locator_id=record_dict['physical_locator_id'],
ip_address=record_dict['ip_address'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
session.add(ucast_mac_remote)
def delete_ucast_mac_remote(context, record_dict):
"""Delete ucast mac remote that matches the supplied uuid."""
session = context.session
with session.begin(subtransactions=True):
if(record_dict['uuid']):
session.query(models.UcastMacsRemotes).filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).delete()
def get_physical_port(context, record_dict):
"""Get physical port that matches the uuid and ovsdb_identifier."""
try:
query = context.session.query(models.PhysicalPorts)
physical_port = query.filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no physical port found for %s and %s',
record_dict['uuid'],
record_dict['ovsdb_identifier'])
return
return physical_port
def get_logical_switch(context, record_dict):
"""Get logical switch that matches the uuid and ovsdb_identifier."""
try:
query = context.session.query(models.LogicalSwitches)
logical_switch = query.filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no logical switch found for %s and %s',
record_dict['uuid'],
record_dict['ovsdb_identifier'])
return
return logical_switch
def get_all_logical_switch_by_name(context, name):
"""Get logical switch that matches the supplied name."""
query = context.session.query(models.LogicalSwitches)
return query.filter_by(name=name).all()
def get_ucast_mac_remote(context, record_dict):
"""Get ucast macs remote that matches the uuid and ovsdb_identifier."""
try:
query = context.session.query(models.UcastMacsRemotes)
remote_mac = query.filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no Remote mac found for %s and %s',
record_dict['uuid'],
record_dict['ovsdb_identifier'])
return
return remote_mac
def get_ucast_mac_local(context, record_dict):
"""Get ucast macs local that matches the uuid and ovsdb_identifier."""
try:
query = context.session.query(models.UcastMacsLocals)
local_mac = query.filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no Local mac found for %s and %s',
record_dict['uuid'],
record_dict['ovsdb_identifier'])
return
return local_mac
def get_ucast_mac_remote_by_mac_and_ls(context, record_dict):
"""Get ucast macs remote that matches the uuid and ovsdb_identifier."""
try:
query = context.session.query(models.UcastMacsRemotes)
remote_mac = query.filter_by(
mac=record_dict['mac'],
ovsdb_identifier=record_dict['ovsdb_identifier'],
logical_switch_id=record_dict['logical_switch_uuid']).one()
except exc.NoResultFound:
LOG.debug('no Remote mac found for %s and %s',
record_dict['uuid'],
record_dict['logical_switch_uuid'])
return
return remote_mac
def get_physical_switch(context, record_dict):
"""Get physical switch that matches the uuid and ovsdb_identifier."""
try:
query = context.session.query(models.PhysicalSwitches)
physical_switch = query.filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no physical switch found for %s and %s',
record_dict['uuid'],
record_dict['ovsdb_identifier'])
return
return physical_switch
def get_physical_locator(context, record_dict):
"""Get physical locator that matches the supplied uuid."""
try:
query = context.session.query(models.PhysicalLocators)
physical_locator = query.filter_by(
uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no physical locator found for %s and %s',
record_dict['uuid'],
record_dict['ovsdb_identifier'])
return
return physical_locator
def get_physical_locator_by_dst_ip(context, record_dict):
"""Get physical locator that matches the supplied destination IP."""
try:
query = context.session.query(models.PhysicalLocators)
physical_locator = query.filter_by(
uuid=record_dict['dst_ip'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no physical locator found for %s and %s',
record_dict['dst_ip'],
record_dict['ovsdb_identifier'])
return
return physical_locator
def get_logical_switch_by_name(context, record_dict):
"""Get logical switch that matches the supplied name."""
try:
query = context.session.query(models.LogicalSwitches)
logical_switch = query.filter_by(
name=record_dict['logical_switch_name'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no logical switch found for %s and %s',
record_dict['logical_switch_name'],
record_dict['ovsdb_identifier'])
return
return logical_switch
def get_all_vlan_bindings_by_physical_port(context, record_dict):
"""Get vlan bindings that matches the supplied physical port."""
query = context.session.query(models.VlanBindings)
return query.filter_by(
port_uuid=record_dict['uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).all()
def get_vlan_binding(context, record_dict):
"""Get vlan bindings that matches the supplied physical port."""
try:
query = context.session.query(models.VlanBindings)
vlan_binding = query.filter_by(
port_uuid=record_dict['port_uuid'],
vlan=record_dict['vlan'],
logical_switch_uuid=record_dict['logical_switch_uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no vlan binding found for %s and %s',
record_dict['port_uuid'],
record_dict['ovsdb_identifier'])
return
return vlan_binding
def get_physical_switch_by_name(context, name):
"""Get logical switch that matches the supplied name."""
query = context.session.query(models.PhysicalSwitches)
return query.filter_by(name=name).first()
def get_physical_port_by_name_and_ps(context, record_dict):
"""Get vlan bindings that matches the supplied physical port."""
try:
query = context.session.query(models.PhysicalPorts)
physical_port = query.filter_by(
name=record_dict['interface_name'],
physical_switch_id=record_dict['physical_switch_id'],
ovsdb_identifier=record_dict['ovsdb_identifier']).one()
except exc.NoResultFound:
LOG.debug('no physical switch found for %s and %s',
record_dict['name'])
return
return physical_port

@ -0,0 +1,100 @@
# Copyright (c) 2015 OpenStack Foundation
#
# 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.
import sqlalchemy as sa
from neutron.db import model_base
class PhysicalLocators(model_base.BASEV2):
__tablename__ = 'physical_locators'
uuid = sa.Column(sa.String(36), nullable=False, primary_key=True)
dst_ip = sa.Column(sa.String(64), nullable=False)
ovsdb_identifier = sa.Column(sa.String(64), nullable=False,
primary_key=True)
__table_args__ = (sa.UniqueConstraint(uuid,
ovsdb_identifier),)
class PhysicalSwitches(model_base.BASEV2):
__tablename__ = 'physical_switches'
uuid = sa.Column(sa.String(36), nullable=False, primary_key=True)
name = sa.Column(sa.String(255), nullable=False)
tunnel_ip = sa.Column(sa.String(64), nullable=False)
ovsdb_identifier = sa.Column(sa.String(64), nullable=False,
primary_key=True)
__table_args__ = (sa.UniqueConstraint(uuid,
ovsdb_identifier),)
class PhysicalPorts(model_base.BASEV2):
__tablename__ = 'physical_ports'
uuid = sa.Column(sa.String(36), nullable=False, primary_key=True)
name = sa.Column(sa.String(255), nullable=False)
physical_switch_id = sa.Column(sa.String(36), nullable=False)
ovsdb_identifier = sa.Column(sa.String(64), nullable=False,
primary_key=True)
__table_args__ = (sa.UniqueConstraint(uuid,
ovsdb_identifier),)
class LogicalSwitches(model_base.BASEV2):
__tablename__ = 'logical_switches'
uuid = sa.Column(sa.String(36), nullable=False, primary_key=True)
name = sa.Column(sa.String(255), nullable=False)
key = sa.Column(sa.Integer, nullable=False)
ovsdb_identifier = sa.Column(sa.String(64), nullable=False,
primary_key=True)
__table_args__ = (sa.UniqueConstraint(uuid,
ovsdb_identifier),)
class UcastMacsLocals(model_base.BASEV2):
__tablename__ = 'ucast_macs_locals'
uuid = sa.Column(sa.String(36), nullable=False, primary_key=True)
mac = sa.Column(sa.String(32), nullable=False)
logical_switch_id = sa.Column(sa.String(36), nullable=False)
physical_locator_id = sa.Column(sa.String(36), nullable=False)
ip_address = sa.Column(sa.String(64), nullable=False)
ovsdb_identifier = sa.Column(sa.String(64), nullable=False,
primary_key=True)
__table_args__ = (sa.UniqueConstraint(uuid,
ovsdb_identifier),)
class UcastMacsRemotes(model_base.BASEV2):
__tablename__ = 'ucast_macs_remotes'
uuid = sa.Column(sa.String(36), nullable=False, primary_key=True)
mac = sa.Column(sa.String(32), nullable=False)
logical_switch_id = sa.Column(sa.String(36), nullable=False)
physical_locator_id = sa.Column(sa.String(36), nullable=False)
ip_address = sa.Column(sa.String(64), nullable=False)
ovsdb_identifier = sa.Column(sa.String(64), nullable=False,
primary_key=True)
__table_args__ = (sa.UniqueConstraint(uuid,
ovsdb_identifier),)
class VlanBindings(model_base.BASEV2):
__tablename__ = 'vlan_bindings'
port_uuid = sa.Column(sa.String(36), nullable=False, primary_key=True)
vlan = sa.Column(sa.Integer, nullable=False, primary_key=True)
logical_switch_uuid = sa.Column(sa.String(36), nullable=False,
primary_key=True)
ovsdb_identifier = sa.Column(sa.String(64), nullable=False,
primary_key=True)
__table_args__ = (sa.UniqueConstraint(port_uuid, vlan,
logical_switch_uuid,
ovsdb_identifier),)

@ -0,0 +1,455 @@
# Copyright (c) 2015 OpenStack Foundation
#
# 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.
from oslo.db import exception as d_exc
from neutron import context
from neutron.openstack.common import uuidutils
from neutron.tests.unit import testlib_api
from networking_l2gw.services.l2gateway.ovsdb import lib
from networking_l2gw.services.l2gateway.ovsdb import models
_uuid = uuidutils.generate_uuid
class OvsdbLibTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(OvsdbLibTestCase, self).setUp()
self.ctx = context.get_admin_context()
def _get_logical_switch_dict(self):
uuid = _uuid()
record_dict = {'uuid': uuid,
'name': 'logical_switch1',
'key': '100',
'ovsdb_identifier': "host1"}
return record_dict
def _create_logical_switch(self, record_dict, name=None):
if name:
record_dict['name'] = name
with self.ctx.session.begin(subtransactions=True):
entry = models.LogicalSwitches(
uuid=record_dict['uuid'],
name=record_dict['name'],
key=record_dict['key'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
self.ctx.session.add(entry)
return entry
def test_get_logical_switch(self):
record_dict = self._get_logical_switch_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_logical_switch(record_dict)
result = lib.get_logical_switch(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_get_logical_switch_return_none(self):
record_dict = {'uuid': 'foo_uuid', 'ovsdb_identifier': 'foo_ovsdb_id'}
result = lib.get_logical_switch(self.ctx, record_dict)
self.assertIsNone(result)
def test_add_logical_switch(self):
record_dict = self._get_logical_switch_dict()
self._create_logical_switch(record_dict)
count = self.ctx.session.query(models.LogicalSwitches).count()
self.assertEqual(1, count)
def test_delete_logical_switch(self):
record_dict = self._get_logical_switch_dict()
self._create_logical_switch(record_dict)
lib.delete_logical_switch(self.ctx, record_dict)
count = self.ctx.session.query(models.LogicalSwitches).count()
self.assertEqual(count, 0)
def _get_physical_locator_dict(self):
uuid = _uuid()
record_dict = {'uuid': uuid,
'dst_ip': '10.0.0.1',
'ovsdb_identifier': 'host1'}
return record_dict
def _create_physical_locator(self, record_dict, dst_ip=None):
if dst_ip:
record_dict['dst_ip'] = dst_ip
with self.ctx.session.begin(subtransactions=True):
entry = models.PhysicalLocators(
uuid=record_dict['uuid'],
dst_ip=record_dict['dst_ip'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
self.ctx.session.add(entry)
return entry
def test_get_physical_locator(self):
record_dict = self._get_physical_locator_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_physical_locator(record_dict)
result = lib.get_physical_locator(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_add_physical_locator(self):
record_dict = self._get_physical_locator_dict()
self._create_physical_locator(record_dict)
count = self.ctx.session.query(models.PhysicalLocators).count()
self.assertEqual(1, count)
def test_delete_physical_locator(self):
record_dict = self._get_physical_locator_dict()
self._create_physical_locator(record_dict)
lib.delete_physical_locator(self.ctx, record_dict)
count = self.ctx.session.query(models.PhysicalLocators).count()
self.assertEqual(count, 0)
def _get_physical_switch_dict(self):
uuid = _uuid()
record_dict = {'uuid': uuid,
'name': 'physical_switch1',
'tunnel_ip': '10.0.0.1',
'ovsdb_identifier': 'host1'}
return record_dict
def _create_physical_switch(self, record_dict, name=None):
if name:
record_dict['name'] = name
with self.ctx.session.begin(subtransactions=True):
entry = models.PhysicalSwitches(
uuid=record_dict['uuid'],
name=record_dict['name'],
tunnel_ip=record_dict['tunnel_ip'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
self.ctx.session.add(entry)
return entry
def test_get_physical_switch(self):
record_dict = self._get_physical_switch_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_physical_switch(record_dict)
result = lib.get_physical_switch(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_add_physical_switch(self):
record_dict = self._get_physical_switch_dict()
self._create_physical_switch(record_dict)
count = self.ctx.session.query(models.PhysicalSwitches).count()
self.assertEqual(1, count)
def test_add_physical_switch_raise_on_duplicate_constraint(self):
record_dict = self._get_physical_switch_dict()
self._create_physical_switch(record_dict)
# Call the method twice to trigger a db duplicate constraint error,
# this time with a different switch name!
self.assertRaises(d_exc.DBDuplicateEntry,
self._create_physical_switch,
record_dict, 'physical_switch2')
def test_delete_physical_switch(self):
record_dict = self._get_physical_switch_dict()
self._create_physical_switch(record_dict)
lib.delete_physical_switch(self.ctx, record_dict)
count = self.ctx.session.query(models.PhysicalSwitches).count()
self.assertEqual(count, 0)
def _get_physical_port_dict(self):
uuid = _uuid()
ps_id = _uuid()
record_dict = {'uuid': uuid,
'name': 'physical_port1',
'physical_switch_id': ps_id,
'ovsdb_identifier': 'host1'}
return record_dict
def _create_physical_port(self,
record_dict,
name=None,
physical_switch_id=None):
if name and physical_switch_id:
record_dict['name'] = name
record_dict['physical_switch_id'] = physical_switch_id
with self.ctx.session.begin(subtransactions=True):
entry = models.PhysicalPorts(
uuid=record_dict['uuid'],
name=record_dict['name'],
physical_switch_id=record_dict['physical_switch_id'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
self.ctx.session.add(entry)
return entry
def test_get_physical_port(self):
record_dict = self._get_physical_port_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_physical_port(record_dict)
result = lib.get_physical_port(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_add_physical_port(self):
record_dict = self._get_physical_port_dict()
self._create_physical_port(record_dict)
count = self.ctx.session.query(models.PhysicalPorts).count()
self.assertEqual(1, count)
def test_add_physical_port_raise_on_duplicate_constraint(self):
record_dict = self._get_physical_port_dict()
self._create_physical_port(record_dict)
# Call the method twice to trigger a db duplicate constraint error,
# this time with a different switch name and physical switch id!
self.assertRaises(d_exc.DBDuplicateEntry,
self._create_physical_port,
record_dict, 'physical_port2', _uuid())
def test_delete_physical_port(self):
record_dict = self._get_physical_port_dict()
self._create_physical_port(record_dict)
lib.delete_physical_port(self.ctx, record_dict)
count = self.ctx.session.query(models.PhysicalPorts).count()
self.assertEqual(count, 0)
def _get_ucast_mac_local_dict(self):
uuid = _uuid()
ls_id = _uuid()
pl_id = _uuid()
record_dict = {'uuid': uuid,
'mac': '12:34:56:78:90:aa:bb',
'logical_switch_id': ls_id,
'physical_locator_id': pl_id,
'ip_address': '10.0.0.1',
'ovsdb_identifier': 'host1'}
return record_dict
def _create_ucast_mac_local(self, record_dict):
with self.ctx.session.begin(subtransactions=True):
entry = models.UcastMacsLocals(
uuid=record_dict['uuid'],
mac=record_dict['mac'],
logical_switch_id=record_dict['logical_switch_id'],
physical_locator_id=record_dict['physical_locator_id'],
ip_address=record_dict['ip_address'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
self.ctx.session.add(entry)
return entry
def test_get_ucast_mac_local(self):
record_dict = self._get_ucast_mac_local_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_ucast_mac_local(record_dict)
result = lib.get_ucast_mac_local(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_add_ucast_mac_local(self):
record_dict = self._get_ucast_mac_local_dict()
self._create_ucast_mac_local(record_dict)
count = self.ctx.session.query(models.UcastMacsLocals).count()
self.assertEqual(1, count)
def test_add_ucast_mac_local_raise_on_duplicate_constraint(self):
record_dict = self._get_ucast_mac_local_dict()
self._create_ucast_mac_local(record_dict)
# Call the method twice to trigger a db duplicate constraint error,
# this time with a different mac and logical switch id!
record_dict['mac'] = '11:22:33:44:55:66:77'
record_dict['logical_switch_id'] = _uuid()
self.assertRaises(d_exc.DBDuplicateEntry,
self._create_ucast_mac_local,
record_dict)
def test_delete_ucast_mac_local(self):
record_dict = self._get_ucast_mac_local_dict()
self._create_ucast_mac_local(record_dict)
lib.delete_ucast_mac_local(self.ctx, record_dict)
count = self.ctx.session.query(models.UcastMacsLocals).count()
self.assertEqual(count, 0)
def _get_ucast_mac_remote_dict(self):
uuid = _uuid()
ls_id = _uuid()
pl_id = _uuid()
record_dict = {'uuid': uuid,
'mac': '12:34:56:78:90:aa:bb',
'logical_switch_id': ls_id,
'physical_locator_id': pl_id,
'ip_address': '10.0.0.1',
'ovsdb_identifier': 'host1'}
return record_dict
def _create_ucast_mac_remote(self,
record_dict,
mac=None,
logical_switch_uuid=None):
if mac and logical_switch_uuid:
record_dict['mac'] = mac
record_dict['logical_switch_id'] = logical_switch_uuid
with self.ctx.session.begin(subtransactions=True):
entry = models.UcastMacsRemotes(
uuid=record_dict['uuid'],
mac=record_dict['mac'],
logical_switch_id=record_dict['logical_switch_id'],
physical_locator_id=record_dict['physical_locator_id'],
ip_address=record_dict['ip_address'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
self.ctx.session.add(entry)
return entry
def test_get_ucast_mac_remote(self):
record_dict = self._get_ucast_mac_remote_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_ucast_mac_remote(record_dict)
result = lib.get_ucast_mac_remote(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_add_ucast_mac_remote(self):
record_dict = self._get_ucast_mac_remote_dict()
self._create_ucast_mac_remote(record_dict)
count = self.ctx.session.query(models.UcastMacsRemotes).count()
self.assertEqual(1, count)
def test_add_ucast_mac_remote_raise_on_duplicate_constraint(self):
record_dict = self._get_ucast_mac_remote_dict()
self._create_ucast_mac_remote(record_dict)
# Call the method twice to trigger a db duplicate constraint error,
# this time with a different mac and logical switch id!
self.assertRaises(d_exc.DBDuplicateEntry,
self._create_ucast_mac_remote,
record_dict, '11:22:33:44:55:66:77', _uuid())
def test_delete_ucast_mac_remote(self):
record_dict = self._get_ucast_mac_remote_dict()
self._create_ucast_mac_remote(record_dict)
lib.delete_ucast_mac_remote(self.ctx, record_dict)
count = self.ctx.session.query(models.UcastMacsRemotes).count()
self.assertEqual(count, 0)
def _get_vlan_binding_dict(self):
port_uuid = _uuid()
ls_uuid = _uuid()
record_dict = {'port_uuid': port_uuid,
'vlan': 200,
'logical_switch_uuid': ls_uuid,
'ovsdb_identifier': 'host1'}
return record_dict
def _create_vlan_binding(self, record_dict, port_uuid=None):
if port_uuid:
record_dict['port_uuid'] = port_uuid
with self.ctx.session.begin(subtransactions=True):
entry = models.VlanBindings(
port_uuid=record_dict['port_uuid'],
vlan=record_dict['vlan'],
logical_switch_uuid=record_dict['logical_switch_uuid'],
ovsdb_identifier=record_dict['ovsdb_identifier'])
self.ctx.session.add(entry)
return entry
def test_get_vlan_binding(self):
record_dict = self._get_vlan_binding_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_vlan_binding(record_dict)
result = lib.get_vlan_binding(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_add_vlan_binding(self):
record_dict = self._get_vlan_binding_dict()
self._create_vlan_binding(record_dict)
count = self.ctx.session.query(models.VlanBindings).count()
self.assertEqual(1, count)
def test_add_vlan_binding_raise_on_duplicate_constraint(self):
record_dict = self._get_vlan_binding_dict()
self._create_vlan_binding(record_dict)
# Call the method twice to trigger a db duplicate constraint error,
# this time with a same entries
self.assertRaises(d_exc.DBDuplicateEntry,
self._create_vlan_binding,
record_dict)
def test_delete_vlan_binding(self):
record_dict = self._get_vlan_binding_dict()
self._create_vlan_binding(record_dict)
lib.delete_vlan_binding(self.ctx, record_dict)
count = self.ctx.session.query(models.VlanBindings).count()
self.assertEqual(count, 0)
def test_get_logical_switch_by_name(self):
record_dict = self._get_logical_switch_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_logical_switch(record_dict,
'logical_switch2')
record_dict['logical_switch_name'] = 'logical_switch2'
result = lib.get_logical_switch_by_name(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_get_all_logical_switch_by_name(self):
record_dict1 = self._get_logical_switch_dict()
self._create_logical_switch(record_dict1, 'logical_switch2')
record_dict2 = self._get_logical_switch_dict()
self._create_logical_switch(record_dict2, 'logical_switch2')
ls_list = lib.get_all_logical_switch_by_name(self.ctx,
'logical_switch2')
self.assertEqual(2, len(ls_list))
def test_get_physical_locator_by_dst_ip(self):
record_dict = self._get_physical_locator_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_physical_locator(record_dict, '20.0.0.1')
record_dict['dst_ip'] = '20.0.0.1'
result = lib.get_physical_locator(self.ctx, record_dict)
self.assertEqual(entry, result)
def test_get_physical_switch_by_name(self):
record_dict = self._get_physical_switch_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_physical_switch(record_dict,
'physical_switch2')
result = lib.get_physical_switch_by_name(self.ctx,
'physical_switch2')
self.assertEqual(entry, result)
def test_get_all_vlan_bindings_by_physical_port(self):
record_dict1 = {'port_uuid': 'ps123',
'vlan': 200,
'logical_switch_uuid': 'ls123',
'ovsdb_identifier': 'host1'}
self._create_vlan_binding(record_dict1)
record_dict1['vlan'] = 300
record_dict1['logical_switch_uuid'] = 'ls456'
self._create_vlan_binding(record_dict1)
record_dict1['uuid'] = record_dict1.get('port_uuid')
vlan_list = lib.get_all_vlan_bindings_by_physical_port(self.ctx,
record_dict1)
self.assertEqual(2, len(vlan_list))
def test_get_ucast_mac_remote_by_mac_and_ls(self):
record_dict = self._get_ucast_mac_remote_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_ucast_mac_remote(record_dict,
'00:11:22:33:44:55:66',
'ls123')
record_dict['mac'] = '00:11:22:33:44:55:66'
record_dict['logical_switch_uuid'] = 'ls123'
result = lib.get_ucast_mac_remote_by_mac_and_ls(self.ctx,
record_dict)
self.assertEqual(entry, result)
def test_get_physical_port_by_name_and_ps(self):
record_dict = self._get_physical_port_dict()
with self.ctx.session.begin(subtransactions=True):
entry = self._create_physical_port(record_dict,
'port1',
'ps123')
record_dict['interface_name'] = 'port1'
record_dict['physical_switch_id'] = 'ps123'
result = lib.get_physical_port_by_name_and_ps(self.ctx,
record_dict)
self.assertEqual(entry, result)
Loading…
Cancel
Save