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

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

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)

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."""