Add a common test case for Port Binding Extension

This commit merges the test cases for Port Binding Extension in
Linux Bridge plugin and Open vSwitch plugin into a common one.
It makes it easier for each plugin to support the port binding extension.

This commit also adds binding:capabilities attribute to OVS plugin.
Note that binding:capabilities is under discussion and it may be changed
in the future. This change is done just to make OVS plugin support
same as Linux Bridge plugin and pass the common unit test for the extension.

blueprint vif-plugging-improvements

Change-Id: Idc38ea65fe2059e23814d94cd0a0f11a09d1f43e

diff --git a/quantum/plugins/openvswitch/ovs_quantum_plugin.py b/quantum/plugins/openvswitch/ovs_quantum_plugin.py
index 642b861d18e031738746125088f0ea566bfe5ee8..28cdb16a4c65f1789d7a5dcd35e6d6b38f833951 100644 (file)
--- a/quantum/plugins/openvswitch/ovs_quantum_plugin.py
+++ b/quantum/plugins/openvswitch/ovs_quantum_plugin.py
@@ -496,6 +496,9 @@ class OVSQuantumPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
     def _extend_port_dict_binding(self, context, port):
         if self._check_view_auth(context, port, self.binding_view):
             port[portbindings.VIF_TYPE] = portbindings.VIF_TYPE_OVS
+            port[portbindings.CAPABILITIES] = {
+                portbindings.CAP_PORT_FILTER:
+                'security-group' in self.supported_extension_aliases}
         return port
 
     def create_port(self, context, port):

diff --git a/quantum/tests/unit/_test_extension_portbindings.py b/quantum/tests/unit/_test_extension_portbindings.py
new file mode 100644 (file)
index 0000000..b6891c1
--- /dev/null
+++ b/quantum/tests/unit/_test_extension_portbindings.py
@@ -0,0 +1,84 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 NEC Corporation
+# All rights reserved.
+#
+#    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.
+#
+# @author: Akihiro Motoki, NEC Corporation
+#
+
+import contextlib
+
+from quantum import context
+from quantum.extensions import portbindings
+from quantum.manager import QuantumManager
+from quantum.openstack.common import cfg
+from quantum.tests.unit import test_db_plugin
+
+
+class PortBindingsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
+
+    # VIF_TYPE must be overridden according to plugin vif_type
+    VIF_TYPE = portbindings.VIF_TYPE_OTHER
+    # The plugin supports the port security feature such as
+    # security groups and anti spoofing.
+    HAS_PORT_FILTER = False
+
+    def _check_response_portbindings(self, port):
+        self.assertEqual(port['binding:vif_type'], self.VIF_TYPE)
+        port_cap = port[portbindings.CAPABILITIES]
+        self.assertEqual(port_cap[portbindings.CAP_PORT_FILTER],
+                         self.HAS_PORT_FILTER)
+
+    def _check_response_no_portbindings(self, port):
+        self.assertTrue('status' in port)
+        self.assertFalse(portbindings.VIF_TYPE in port)
+        self.assertFalse(portbindings.CAPABILITIES in port)
+
+    def test_port_vif_details(self):
+        plugin = QuantumManager.get_plugin()
+        with self.port(name='name') as port:
+            port_id = port['port']['id']
+            # Check a response of create_port
+            self._check_response_portbindings(port['port'])
+            # Check a response of get_port
+            ctx = context.get_admin_context()
+            port = plugin.get_port(ctx, port_id)
+            self._check_response_portbindings(port)
+            # By default user is admin - now test non admin user
+            ctx = context.Context(user_id=None,
+                                  tenant_id=self._tenant_id,
+                                  is_admin=False,
+                                  read_deleted="no")
+            non_admin_port = plugin.get_port(ctx, port_id)
+            self._check_response_no_portbindings(non_admin_port)
+
+    def test_ports_vif_details(self):
+        plugin = QuantumManager.get_plugin()
+        cfg.CONF.set_default('allow_overlapping_ips', True)
+        with contextlib.nested(self.port(), self.port()):
+            ctx = context.get_admin_context()
+            ports = plugin.get_ports(ctx)
+            self.assertEqual(len(ports), 2)
+            for port in ports:
+                self._check_response_portbindings(port)
+            # By default user is admin - now test non admin user
+            ctx = context.Context(user_id=None,
+                                  tenant_id=self._tenant_id,
+                                  is_admin=False,
+                                  read_deleted="no")
+            ports = plugin.get_ports(ctx)
+            self.assertEqual(len(ports), 2)
+            for non_admin_port in ports:
+                self._check_response_no_portbindings(non_admin_port)

diff --git a/quantum/tests/unit/linuxbridge/test_linuxbridge_plugin.py b/quantum/tests/unit/linuxbridge/test_linuxbridge_plugin.py
index 3eed40fde48163876956e6eab5b20306a362585f..1112a0c91b2aa26e967b5e61b07987a1929b8332 100644 (file)
--- a/quantum/tests/unit/linuxbridge/test_linuxbridge_plugin.py
+++ b/quantum/tests/unit/linuxbridge/test_linuxbridge_plugin.py
@@ -13,12 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import contextlib
-
-from quantum import context
 from quantum.extensions import portbindings
-from quantum.manager import QuantumManager
-from quantum.openstack.common import cfg
+from quantum.tests.unit import _test_extension_portbindings as test_bindings
 from quantum.tests.unit import test_db_plugin as test_plugin
 
 PLUGIN_NAME = ('quantum.plugins.linuxbridge.'
@@ -43,49 +39,11 @@ class TestLinuxBridgeV2HTTPResponse(test_plugin.TestV2HTTPResponse,
 
 
 class TestLinuxBridgePortsV2(test_plugin.TestPortsV2,
-                             LinuxBridgePluginV2TestCase):
-    def test_port_vif_details(self):
-        plugin = QuantumManager.get_plugin()
-        with self.port(name='name') as port:
-            port_id = port['port']['id']
-            self.assertEqual(port['port'][portbindings.VIF_TYPE],
-                             portbindings.VIF_TYPE_BRIDGE)
-            port_cap = port['port'][portbindings.CAPABILITIES]
-            self.assertEqual(port_cap[portbindings.CAP_PORT_FILTER], True)
-            # By default user is admin - now test non admin user
-            ctx = context.Context(user_id=None,
-                                  tenant_id=self._tenant_id,
-                                  is_admin=False,
-                                  read_deleted="no")
-            non_admin_port = plugin.get_port(ctx, port_id)
-            self.assertTrue('status' in non_admin_port)
-            self.assertFalse(portbindings.VIF_TYPE in non_admin_port)
-            self.assertFalse(portbindings.CAPABILITIES in non_admin_port)
+                             LinuxBridgePluginV2TestCase,
+                             test_bindings.PortBindingsTestCase):
 
-    def test_ports_vif_details(self):
-        cfg.CONF.set_default('allow_overlapping_ips', True)
-        plugin = QuantumManager.get_plugin()
-        with contextlib.nested(self.port(), self.port()) as (port1, port2):
-            ctx = context.get_admin_context()
-            ports = plugin.get_ports(ctx)
-            self.assertEqual(len(ports), 2)
-            for port in ports:
-                self.assertEqual(port[portbindings.VIF_TYPE],
-                                 portbindings.VIF_TYPE_BRIDGE)
-                port_cap = port[portbindings.CAPABILITIES]
-                self.assertEqual(port_cap[portbindings.CAP_PORT_FILTER], True)
-            # By default user is admin - now test non admin user
-            ctx = context.Context(user_id=None,
-                                  tenant_id=self._tenant_id,
-                                  is_admin=False,
-                                  read_deleted="no")
-            ports = plugin.get_ports(ctx)
-            self.assertEqual(len(ports), 2)
-            for non_admin_port in ports:
-                self.assertTrue('status' in non_admin_port)
-                self.assertFalse(portbindings.VIF_TYPE in non_admin_port)
-                self.assertFalse(portbindings.CAP_PORT_FILTER
-                                 in non_admin_port)
+    VIF_TYPE = portbindings.VIF_TYPE_BRIDGE
+    HAS_PORT_FILTER = True
 
 
 class TestLinuxBridgeNetworksV2(test_plugin.TestNetworksV2,

diff --git a/quantum/tests/unit/openvswitch/test_openvswitch_plugin.py b/quantum/tests/unit/openvswitch/test_openvswitch_plugin.py
index 49a10918196c4c2da4afbd282982435bd6050f51..d7244b53f8abe89b8f848eb814985e2acb8f8330 100644 (file)
--- a/quantum/tests/unit/openvswitch/test_openvswitch_plugin.py
+++ b/quantum/tests/unit/openvswitch/test_openvswitch_plugin.py
@@ -13,12 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import contextlib
-
-from quantum import context
 from quantum.extensions import portbindings
-from quantum.manager import QuantumManager
-from quantum.openstack.common import cfg
+from quantum.tests.unit import _test_extension_portbindings as test_bindings
 from quantum.tests.unit import test_db_plugin as test_plugin
 
 
@@ -42,42 +38,11 @@ class TestOpenvswitchV2HTTPResponse(test_plugin.TestV2HTTPResponse,
 
 
 class TestOpenvswitchPortsV2(test_plugin.TestPortsV2,
-                             OpenvswitchPluginV2TestCase):
-    def test_port_vif_details(self):
-        plugin = QuantumManager.get_plugin()
-        with self.port(name='name') as port:
-            port_id = port['port']['id']
-            self.assertEqual(port['port']['binding:vif_type'],
-                             portbindings.VIF_TYPE_OVS)
-            # By default user is admin - now test non admin user
-            ctx = context.Context(user_id=None,
-                                  tenant_id=self._tenant_id,
-                                  is_admin=False,
-                                  read_deleted="no")
-            non_admin_port = plugin.get_port(ctx, port_id)
-            self.assertTrue('status' in non_admin_port)
-            self.assertFalse('binding:vif_type' in non_admin_port)
+                             OpenvswitchPluginV2TestCase,
+                             test_bindings.PortBindingsTestCase):
 
-    def test_ports_vif_details(self):
-        cfg.CONF.set_default('allow_overlapping_ips', True)
-        plugin = QuantumManager.get_plugin()
-        with contextlib.nested(self.port(), self.port()) as (port1, port2):
-            ctx = context.get_admin_context()
-            ports = plugin.get_ports(ctx)
-            self.assertEqual(len(ports), 2)
-            for port in ports:
-                self.assertEqual(port['binding:vif_type'],
-                                 portbindings.VIF_TYPE_OVS)
-            # By default user is admin - now test non admin user
-            ctx = context.Context(user_id=None,
-                                  tenant_id=self._tenant_id,
-                                  is_admin=False,
-                                  read_deleted="no")
-            ports = plugin.get_ports(ctx)
-            self.assertEqual(len(ports), 2)
-            for non_admin_port in ports:
-                self.assertTrue('status' in non_admin_port)
-                self.assertFalse('binding:vif_type' in non_admin_port)
+    VIF_TYPE = portbindings.VIF_TYPE_OVS
+    HAS_PORT_FILTER = False
 
 
 class TestOpenvswitchNetworksV2(test_plugin.TestNetworksV2,