Merge "Add functional test for tags in Neutron resources"

openstack/network/v2/_proxy.py

@@ -5408,6 +5408,17 @@ class Proxy(proxy.Proxy):
                 % resource.__class__.__name__
             )
 
+    def get_tags(self, resource):
+        """Retrieve the tags of a specified resource
+
+        :param resource: :class:`~openstack.resource.Resource` instance.
+
+        :returns: The resource tags list
+        :rtype: "list"
+        """
+        self._check_tag_support(resource)
+        return resource.fetch_tags(self).tags
+
     def set_tags(self, resource, tags):
         """Replace tags of a specified resource with specified tags
 
@@ -5422,6 +5433,56 @@ class Proxy(proxy.Proxy):
         self._check_tag_support(resource)
         return resource.set_tags(self, tags)
 
+    def add_tag(self, resource, tag):
+        """Add one single tag to a specified resource
+
+        :param resource: :class:`~openstack.resource.Resource` instance.
+        :param tag: New tag to be set.
+        :type tag: "str"
+
+        :returns: The updated resource
+        :rtype: :class:`~openstack.resource.Resource`
+        """
+        self._check_tag_support(resource)
+        return resource.add_tag(self, tag)
+
+    def remove_tag(self, resource, tag):
+        """Remove one single tag of a specified resource
+
+        :param resource: :class:`~openstack.resource.Resource` instance.
+        :param tag: New tag to be set.
+        :type tag: "str"
+
+        :returns: The updated resource
+        :rtype: :class:`~openstack.resource.Resource`
+        """
+        self._check_tag_support(resource)
+        return resource.remove_tag(self, tag)
+
+    def remove_all_tags(self, resource):
+        """Remove all tags of a specified resource
+
+        :param resource: :class:`~openstack.resource.Resource` instance.
+
+        :returns: The updated resource
+        :rtype: :class:`~openstack.resource.Resource`
+        """
+        self._check_tag_support(resource)
+        return resource.remove_all_tags(self)
+
+    def check_tag(self, resource, tag):
+        """Checks if tag exists on the specified resource
+
+        :param resource: :class:`~openstack.resource.Resource` instance.
+        :param tag: Tag to be tested
+        :type tags: "string"
+
+        :returns: If the tag exists in the specified resource
+        :rtype: bool
+        """
+        self._check_tag_support(resource)
+        return resource.check_tag(self, tag)
+
     def create_trunk(self, **attrs):
         """Create a new trunk from attributes
 

openstack/tests/functional/network/v2/common.py

@@ -0,0 +1,90 @@
+# 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 unittest import mock
+
+from openstack import exceptions
+from openstack.tests.functional import base
+
+
+# NOTE: method to make mypy happy.
+def _get_command(*args):
+    return mock.Mock()
+
+
+class TestTagNeutron(base.BaseFunctionalTest):
+    get_command = _get_command
+
+    def test_set_tags(self):
+        sot = self.get_command(self.ID)
+        self.assertEqual([], sot.tags)
+
+        self.user_cloud.network.set_tags(sot, ["blue"])
+        sot = self.get_command(self.ID)
+        self.assertEqual(["blue"], sot.tags)
+
+        self.user_cloud.network.set_tags(sot, [])
+        sot = self.get_command(self.ID)
+        self.assertEqual([], sot.tags)
+
+    def test_get_tags(self):
+        sot = self.get_command(self.ID)
+        self.assertEqual([], sot.tags)
+
+        self.user_cloud.network.set_tags(sot, ["blue", "red"])
+        tags = self.user_cloud.network.get_tags(sot)
+        self.assertEqual(["blue", "red"], tags)
+
+    def test_add_tag(self):
+        sot = self.get_command(self.ID)
+        self.assertEqual([], sot.tags)
+
+        self.user_cloud.network.add_tag(sot, "blue")
+        tags = self.user_cloud.network.get_tags(sot)
+        self.assertEqual(["blue"], tags)
+
+        # The operation is idempotent.
+        self.user_cloud.network.add_tag(sot, "blue")
+        tags = self.user_cloud.network.get_tags(sot)
+        self.assertEqual(["blue"], tags)
+
+    def test_remove_tag(self):
+        sot = self.get_command(self.ID)
+        self.assertEqual([], sot.tags)
+
+        self.user_cloud.network.set_tags(sot, ["blue"])
+        tags = self.user_cloud.network.get_tags(sot)
+        self.assertEqual(["blue"], tags)
+
+        self.user_cloud.network.remove_tag(sot, "blue")
+        tags = self.user_cloud.network.get_tags(sot)
+        self.assertEqual([], tags)
+
+        # The operation is not idempotent.
+        self.assertRaises(
+            exceptions.NotFoundException,
+            self.user_cloud.network.remove_tag,
+            sot,
+            "blue",
+        )
+
+    def test_remove_all_tags(self):
+        sot = self.get_command(self.ID)
+        self.assertEqual([], sot.tags)
+
+        self.user_cloud.network.set_tags(sot, ["blue", "red"])
+        sot = self.get_command(self.ID)
+        self.assertEqual(["blue", "red"], sot.tags)
+
+        self.user_cloud.network.remove_all_tags(sot)
+        sot = self.get_command(self.ID)
+        self.assertEqual([], sot.tags)

openstack/tests/functional/network/v2/test_floating_ip.py

@@ -9,17 +9,17 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-
+# mypy: disable-error-code="method-assign"
 
 from openstack.network.v2 import floating_ip
 from openstack.network.v2 import network
 from openstack.network.v2 import port
 from openstack.network.v2 import router
 from openstack.network.v2 import subnet
-from openstack.tests.functional import base
+from openstack.tests.functional.network.v2 import common
 
 
-class TestFloatingIP(base.BaseFunctionalTest):
+class TestFloatingIP(common.TestTagNeutron):
     IPV4 = 4
     EXT_CIDR = "10.100.0.0/24"
     INT_CIDR = "10.101.0.0/24"
@@ -105,6 +105,8 @@ class TestFloatingIP(base.BaseFunctionalTest):
         fip = self.user_cloud.network.create_ip(**fip_args)
         assert isinstance(fip, floating_ip.FloatingIP)
         self.FIP = fip
+        self.ID = self.FIP.id
+        self.get_command = self.user_cloud.network.get_ip
 
     def tearDown(self):
         sot = self.user_cloud.network.delete_ip(
@@ -194,18 +196,6 @@ class TestFloatingIP(base.BaseFunctionalTest):
         self._assert_port_details(self.PORT, sot.port_details)
         self.assertEqual(self.FIP.id, sot.id)
 
-    def test_set_tags(self):
-        sot = self.user_cloud.network.get_ip(self.FIP.id)
-        self.assertEqual([], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, ["blue"])
-        sot = self.user_cloud.network.get_ip(self.FIP.id)
-        self.assertEqual(["blue"], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, [])
-        sot = self.user_cloud.network.get_ip(self.FIP.id)
-        self.assertEqual([], sot.tags)
-
     def _assert_port_details(self, port, port_details):
         self.assertEqual(port.name, port_details["name"])
         self.assertEqual(port.network_id, port_details["network_id"])

openstack/tests/functional/network/v2/test_network.py

@@ -9,10 +9,10 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-
+# mypy: disable-error-code="method-assign"
 
 from openstack.network.v2 import network
-from openstack.tests.functional import base
+from openstack.tests.functional.network.v2 import common
 
 
 def create_network(conn, name, cidr):
@@ -35,7 +35,7 @@ def delete_network(conn, network, subnet):
         conn.network.delete_network(network)
 
 
-class TestNetwork(base.BaseFunctionalTest):
+class TestNetwork(common.TestTagNeutron):
     ID = None
 
     def setUp(self):
@@ -45,6 +45,7 @@ class TestNetwork(base.BaseFunctionalTest):
         assert isinstance(sot, network.Network)
         self.assertEqual(self.NAME, sot.name)
         self.ID = sot.id
+        self.get_command = self.user_cloud.network.get_network
 
     def tearDown(self):
         sot = self.user_cloud.network.delete_network(
@@ -83,15 +84,3 @@ class TestNetwork(base.BaseFunctionalTest):
     def test_list(self):
         names = [o.name for o in self.user_cloud.network.networks()]
         self.assertIn(self.NAME, names)
-
-    def test_set_tags(self):
-        sot = self.user_cloud.network.get_network(self.ID)
-        self.assertEqual([], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, ["blue"])
-        sot = self.user_cloud.network.get_network(self.ID)
-        self.assertEqual(["blue"], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, [])
-        sot = self.user_cloud.network.get_network(self.ID)
-        self.assertEqual([], sot.tags)

openstack/tests/functional/network/v2/test_port.py

@@ -9,15 +9,15 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-
+# mypy: disable-error-code="method-assign"
 
 from openstack.network.v2 import network
 from openstack.network.v2 import port
 from openstack.network.v2 import subnet
-from openstack.tests.functional import base
+from openstack.tests.functional.network.v2 import common
 
 
-class TestPort(base.BaseFunctionalTest):
+class TestPort(common.TestTagNeutron):
     IPV4 = 4
     CIDR = "10.100.0.0/24"
     NET_ID = None
@@ -48,7 +48,8 @@ class TestPort(base.BaseFunctionalTest):
         )
         assert isinstance(prt, port.Port)
         self.assertEqual(self.PORT_NAME, prt.name)
-        self.PORT_ID = prt.id
+        self.PORT_ID = self.ID = prt.id
+        self.get_command = self.user_cloud.network.get_port
 
     def tearDown(self):
         sot = self.user_cloud.network.delete_port(
@@ -84,15 +85,3 @@ class TestPort(base.BaseFunctionalTest):
             self.PORT_ID, name=self.UPDATE_NAME
         )
         self.assertEqual(self.UPDATE_NAME, sot.name)
-
-    def test_set_tags(self):
-        sot = self.user_cloud.network.get_port(self.PORT_ID)
-        self.assertEqual([], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, ["blue"])
-        sot = self.user_cloud.network.get_port(self.PORT_ID)
-        self.assertEqual(["blue"], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, [])
-        sot = self.user_cloud.network.get_port(self.PORT_ID)
-        self.assertEqual([], sot.tags)

openstack/tests/functional/network/v2/test_router.py

@@ -9,13 +9,13 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-
+# mypy: disable-error-code="method-assign"
 
 from openstack.network.v2 import router
-from openstack.tests.functional import base
+from openstack.tests.functional.network.v2 import common
 
 
-class TestRouter(base.BaseFunctionalTest):
+class TestRouter(common.TestTagNeutron):
     ID = None
 
     def setUp(self):
@@ -26,6 +26,7 @@ class TestRouter(base.BaseFunctionalTest):
         assert isinstance(sot, router.Router)
         self.assertEqual(self.NAME, sot.name)
         self.ID = sot.id
+        self.get_command = self.user_cloud.network.get_router
 
     def tearDown(self):
         sot = self.user_cloud.network.delete_router(
@@ -57,15 +58,3 @@ class TestRouter(base.BaseFunctionalTest):
             self.ID, name=self.UPDATE_NAME
         )
         self.assertEqual(self.UPDATE_NAME, sot.name)
-
-    def test_set_tags(self):
-        sot = self.user_cloud.network.get_router(self.ID)
-        self.assertEqual([], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, ["blue"])
-        sot = self.user_cloud.network.get_router(self.ID)
-        self.assertEqual(["blue"], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, [])
-        sot = self.user_cloud.network.get_router(self.ID)
-        self.assertEqual([], sot.tags)

openstack/tests/functional/network/v2/test_security_group.py

@@ -9,13 +9,13 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-
+# mypy: disable-error-code="method-assign"
 
 from openstack.network.v2 import security_group
-from openstack.tests.functional import base
+from openstack.tests.functional.network.v2 import common
 
 
-class TestSecurityGroup(base.BaseFunctionalTest):
+class TestSecurityGroup(common.TestTagNeutron):
     ID = None
 
     def setUp(self):
@@ -25,6 +25,7 @@ class TestSecurityGroup(base.BaseFunctionalTest):
         assert isinstance(sot, security_group.SecurityGroup)
         self.assertEqual(self.NAME, sot.name)
         self.ID = sot.id
+        self.get_command = self.user_cloud.network.get_security_group
 
     def tearDown(self):
         sot = self.user_cloud.network.delete_security_group(
@@ -51,15 +52,3 @@ class TestSecurityGroup(base.BaseFunctionalTest):
             o.id for o in self.user_cloud.network.security_groups(id=[self.ID])
         ]
         self.assertIn(self.ID, ids)
-
-    def test_set_tags(self):
-        sot = self.user_cloud.network.get_security_group(self.ID)
-        self.assertEqual([], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, ["blue"])
-        sot = self.user_cloud.network.get_security_group(self.ID)
-        self.assertEqual(["blue"], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, [])
-        sot = self.user_cloud.network.get_security_group(self.ID)
-        self.assertEqual([], sot.tags)

openstack/tests/functional/network/v2/test_subnet.py

@@ -9,14 +9,14 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-
+# mypy: disable-error-code="method-assign"
 
 from openstack.network.v2 import network
 from openstack.network.v2 import subnet
-from openstack.tests.functional import base
+from openstack.tests.functional.network.v2 import common
 
 
-class TestSubnet(base.BaseFunctionalTest):
+class TestSubnet(common.TestTagNeutron):
     IPV4 = 4
     CIDR = "10.100.0.0/24"
     DNS_SERVERS = ["8.8.4.4", "8.8.8.8"]
@@ -45,7 +45,8 @@ class TestSubnet(base.BaseFunctionalTest):
         )
         assert isinstance(sub, subnet.Subnet)
         self.assertEqual(self.SUB_NAME, sub.name)
-        self.SUB_ID = sub.id
+        self.SUB_ID = self.ID = sub.id
+        self.get_command = self.user_cloud.network.get_subnet
 
     def tearDown(self):
         sot = self.user_cloud.network.delete_subnet(self.SUB_ID)
@@ -81,15 +82,3 @@ class TestSubnet(base.BaseFunctionalTest):
             self.SUB_ID, name=self.UPDATE_NAME
         )
         self.assertEqual(self.UPDATE_NAME, sot.name)
-
-    def test_set_tags(self):
-        sot = self.user_cloud.network.get_subnet(self.SUB_ID)
-        self.assertEqual([], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, ["blue"])
-        sot = self.user_cloud.network.get_subnet(self.SUB_ID)
-        self.assertEqual(["blue"], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, [])
-        sot = self.user_cloud.network.get_subnet(self.SUB_ID)
-        self.assertEqual([], sot.tags)

openstack/tests/functional/network/v2/test_subnet_pool.py

@@ -9,13 +9,13 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
-
+# mypy: disable-error-code="method-assign"
 
 from openstack.network.v2 import subnet_pool as _subnet_pool
-from openstack.tests.functional import base
+from openstack.tests.functional.network.v2 import common
 
 
-class TestSubnetPool(base.BaseFunctionalTest):
+class TestSubnetPool(common.TestTagNeutron):
     SUBNET_POOL_ID = None
     MINIMUM_PREFIX_LENGTH = 8
     DEFAULT_PREFIX_LENGTH = 24
@@ -40,7 +40,8 @@ class TestSubnetPool(base.BaseFunctionalTest):
         )
         assert isinstance(subnet_pool, _subnet_pool.SubnetPool)
         self.assertEqual(self.SUBNET_POOL_NAME, subnet_pool.name)
-        self.SUBNET_POOL_ID = subnet_pool.id
+        self.SUBNET_POOL_ID = self.ID = subnet_pool.id
+        self.get_command = self.user_cloud.network.get_subnet_pool
 
     def tearDown(self):
         sot = self.user_cloud.network.delete_subnet_pool(self.SUBNET_POOL_ID)
@@ -71,15 +72,3 @@ class TestSubnetPool(base.BaseFunctionalTest):
             self.SUBNET_POOL_ID, name=self.SUBNET_POOL_NAME_UPDATED
         )
         self.assertEqual(self.SUBNET_POOL_NAME_UPDATED, sot.name)
-
-    def test_set_tags(self):
-        sot = self.user_cloud.network.get_subnet_pool(self.SUBNET_POOL_ID)
-        self.assertEqual([], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, ["blue"])
-        sot = self.user_cloud.network.get_subnet_pool(self.SUBNET_POOL_ID)
-        self.assertEqual(["blue"], sot.tags)
-
-        self.user_cloud.network.set_tags(sot, [])
-        sot = self.user_cloud.network.get_subnet_pool(self.SUBNET_POOL_ID)
-        self.assertEqual([], sot.tags)