Browse Source

Don't call get_session in vlan type driver

In stein the get_session function was removed from the
upstream type driver helper. This change replaces that
with calls to neutron-lib's db_api.

This also re-enables the type driver unit tests which have
been disabled for a while (missing __init__.py) and adds
coverage for a upstream test cases for the upstream VLAN
type driver against the Arista VLAN type driver.

Change-Id: I5b52abafadf236278ca707c89b69883501992219
changes/15/761515/1 2020.1.1
Mitchell Jameson 11 months ago
parent
commit
44560756cc
  1. 2
      lower-constraints.txt
  2. 8
      networking_arista/ml2/arista_resources.py
  3. 13
      networking_arista/ml2/type_drivers/type_arista_vlan.py
  4. 0
      networking_arista/tests/unit/ml2/type_drivers/__init__.py
  5. 34
      networking_arista/tests/unit/ml2/type_drivers/test_arista_type_driver.py

2
lower-constraints.txt

@ -30,7 +30,7 @@ future==0.16.0
futurist==1.2.0
gitdb==0.6.4
GitPython==1.0.1
greenlet==0.4.10
greenlet==0.4.14
hacking==1.1.0
httplib2==0.9.1
imagesize==0.7.1

8
networking_arista/ml2/arista_resources.py

@ -101,8 +101,8 @@ class AristaResourcesBase(object):
def update_neutron_resource(self, id, action):
LOG.debug("%(pid)s Requesting %(action)s %(class)s resource %(id)s",
{'action': action, 'class': self.__class__.__name__, 'id': id,
'pid': os.getpid()})
{'action': action, 'class': self.__class__.__name__,
'id': id, 'pid': os.getpid()})
resource = self.get_db_resources(id)
if action == a_const.FULL_SYNC:
self.clear_neutron_data()
@ -119,8 +119,8 @@ class AristaResourcesBase(object):
# cvx resources
self.force_resource_update(id)
LOG.debug("%(pid)s Resource %(class)s %(id)s found, creating",
{'class': self.__class__.__name__, 'id': id,
'pid': os.getpid()})
{'class': self.__class__.__name__, 'id': id,
'pid': os.getpid()})
self._add_neutron_resource(resource[0])
else:
LOG.debug("%(pid)s Resource %(class)s %(id)s not found, deleting",

13
networking_arista/ml2/type_drivers/type_arista_vlan.py

@ -85,11 +85,12 @@ class AristaVlanTypeDriver(type_vlan.VlanTypeDriver):
self).release_segment(context, segment)
def allocate_fully_specified_segment(self, context, **raw_segment):
session, ctx_manager = self._get_session(context)
with ctx_manager:
alloc = session.query(self.model).filter_by(**raw_segment).first()
with db_api.CONTEXT_READER.using(context):
alloc = (
context.session.query(self.model).filter_by(**raw_segment).
first())
if not alloc:
raise exc.VlanUnavailable(**raw_segment)
return super(AristaVlanTypeDriver,
self).allocate_fully_specified_segment(
context, **raw_segment)
return super(AristaVlanTypeDriver,
self).allocate_fully_specified_segment(
context, **raw_segment)

0
networking_arista/tests/unit/ml2/type_drivers/__init__.py

34
networking_arista/tests/unit/ml2/type_drivers/test_arista_type_driver.py

@ -21,11 +21,14 @@ from neutron_lib.db import api as db_api
from oslo_config import cfg
from neutron.db.models.plugins.ml2 import vlanallocation
from neutron.tests.unit.plugins.ml2.drivers import test_helpers
from neutron.tests.unit import testlib_api
from networking_arista.ml2.drivers.driver_helpers import VlanSyncService
from networking_arista.ml2.drivers.type_arista_vlan import AristaVlanTypeDriver
import networking_arista.tests.unit.ml2.utils as utils
import networking_arista.common.exceptions as exc
from networking_arista.ml2.type_drivers.driver_helpers import VlanSyncService
from networking_arista.ml2.type_drivers.type_arista_vlan \
import AristaVlanTypeDriver
import networking_arista.tests.unit.utils as utils
EAPI_SEND_FUNC = ('networking_arista.ml2.rpc.arista_eapi.AristaRPCWrapperEapi'
@ -49,9 +52,7 @@ class AristaTypeDriverTest(testlib_api.SqlTestCase):
type_driver.initialize()
cmds = ['show openstack agent uuid',
'show openstack instances',
'show openstack agent uuid',
'show openstack features']
'show openstack resource-pool vlan region RegionOne uuid']
calls = [mock.call(cmds=[cmd], commands_to_log=[cmd])
for cmd in cmds]
@ -59,6 +60,27 @@ class AristaTypeDriverTest(testlib_api.SqlTestCase):
type_driver.timer.cancel()
class AristaTypeDriverHelpersTest(test_helpers.HelpersTest):
def setUp(self):
utils.setup_arista_wrapper_config(cfg)
super(AristaTypeDriverHelpersTest, self).setUp()
self.driver = AristaVlanTypeDriver()
self.driver.network_vlan_ranges = test_helpers.NETWORK_VLAN_RANGES
def test_allocate_specific_allocated_segment_outside_pools(self):
# Invalid test case for Arista type driver because the first
# allocate fails with VlanUnavailable
pass
def test_allocate_specific_unallocated_segment_outside_pools(self):
expected = dict(physical_network=test_helpers.TENANT_NET,
vlan_id=test_helpers.VLAN_OUTSIDE)
self.assertRaises(exc.VlanUnavailable,
self.driver.allocate_fully_specified_segment,
self.context, **expected)
class VlanSyncServiceTest(testlib_api.SqlTestCase):
"""Test that VLANs are synchronized between EOS and Neutron."""

Loading…
Cancel
Save