Functiona tests for Private nat gateway (#649)

Functiona tests for Private nat gateway

Summary
This PR adds functional tests for private nat gateway.
Details

Added functional tests for private nat gateway.
Added Mixin class to keep common functionality for private dnat rule and snat rule
Refactored and moved cleanup_routers helper into base class to reuse it in both test_zone and test_private_nat_gateway

Reviewed-by: Anton Sidelnikov

Author: shhovhan

otcextensions/tests/functional/base.py

@@ -10,13 +10,15 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import logging
 import os
 import uuid
 
 import openstack.config
 from keystoneauth1 import exceptions as _exceptions
 
 from openstack import connection
+from openstack import exceptions as sdk_exceptions
 from openstack.tests import base
 from otcextensions import sdk
 
@@ -37,6 +39,7 @@ def _get_resource_value(resource_key, default):
 
 IMAGE_NAME = _get_resource_value("image_name", "cirros-0.3.5-x86_64-disk")
 FLAVOR_NAME = _get_resource_value("flavor_name", "m1.small")
+_logger = logging.getLogger(__name__)
 
 
 class BaseFunctionalTest(base.TestCase):
@@ -76,6 +79,31 @@ def setUp(self):
                 )
             )
 
+    def cleanup_stale_routers(self, prefix):
+        """Clean up stale Neutron routers created by functional tests."""
+        for router in self.conn.network.routers():
+            if not router.name or not router.name.startswith(prefix):
+                continue
+            try:
+                ports = list(self.conn.network.ports(device_id=router.id))
+                for port in ports:
+                    fixed_ips = getattr(port, "fixed_ips", []) or []
+                    for fixed_ip in fixed_ips:
+                        subnet_id = fixed_ip.get("subnet_id")
+                        if not subnet_id:
+                            continue
+                        try:
+                            self.conn.network.remove_interface_from_router(
+                                router, subnet_id=subnet_id
+                            )
+                        except sdk_exceptions.SDKException:
+                            pass
+                self.conn.network.delete_router(router, ignore_missing=True)
+            except sdk_exceptions.SDKException as e:
+                _logger.warning(
+                    "Failed to cleanup stale router %s: %s", router.id, str(e)
+                )
+
 
 class NetworkBaseFunctionalTest(BaseFunctionalTest):
     def create_network(self, prefix="sdk-test"):

otcextensions/tests/functional/osclient/privatenat/v3/test_private_nat_gateway.py

@@ -11,16 +11,107 @@
 #    under the License.
 
 import json
+import uuid
 
 from openstackclient.tests.functional import base
 
+from openstack import connection
+from openstack import exceptions as sdk_exceptions
+from openstack import resource
+from otcextensions import sdk
+from otcextensions.tests.functional import base as sdk_base
+from otcextensions.tests.functional.privatenat import PrivateNatEnvironmentMixin
 
-class TestPrivateNatGateway(base.TestCase):
-    """Functional Tests for Private NAT Gateway"""
+
+class TestPrivateNatGateway(PrivateNatEnvironmentMixin, base.TestCase):
+    """Functional tests for private NAT gateway CLI commands."""
 
     def setUp(self):
         super(TestPrivateNatGateway, self).setUp()
+        self.conn = connection.Connection(config=sdk_base.TEST_CLOUD_REGION)
+        sdk.register_otc_extensions(self.conn)
+
+    def _create_gateway(self, prefix, description):
+        stack = self._create_private_nat_network_stack(prefix)
+        self.addCleanup(self._cleanup_private_nat_network_stack, stack)
+        gateway_name = "{prefix}-{suffix}".format(
+            prefix=prefix, suffix=uuid.uuid4().hex[:8]
+        )
+
+        gateway = json.loads(
+            self.openstack(
+                "privatenat gateway create "
+                "--name {name} "
+                "--description {description} "
+                "--downlink-vpc virsubnet_id={virsubnet_id} "
+                "-f json".format(
+                    name=gateway_name,
+                    description=description,
+                    virsubnet_id=stack["subnet"].id,
+                )
+            )
+        )
+        self.addCleanup(self._delete_gateway, gateway["id"])
+        return {"stack": stack, "gateway": gateway}
+
+    def _delete_gateway(self, gateway_id):
+        try:
+            gateway = self.conn.natv3.get_private_nat_gateway(gateway_id)
+        except sdk_exceptions.ResourceNotFound:
+            return
+
+        self.conn.natv3.delete_private_nat_gateway(gateway_id, ignore_missing=True)
+        resource.wait_for_delete(self.conn.natv3, gateway, 2, 120)
 
     def test_private_nat_gateway_list(self):
         json_output = json.loads(self.openstack("privatenat gateway list -f json "))
         self.assertIsNotNone(json_output)
+
+    def test_private_nat_gateway_create(self):
+        env = self._create_gateway("cli-private-nat-create", "cli-private-nat-create")
+        created = env["gateway"]
+
+        self.assertIsNotNone(created)
+        self.assertEqual("cli-private-nat-create", created["description"])
+
+    def test_private_nat_gateway_show(self):
+        env = self._create_gateway("cli-private-nat-show", "cli-private-nat-show")
+        gateway_id = env["gateway"]["id"]
+
+        shown = json.loads(
+            self.openstack("privatenat gateway show -f json " + gateway_id)
+        )
+        self.assertEqual(gateway_id, shown["id"])
+        self.assertEqual("cli-private-nat-show", shown["description"])
+
+    def test_private_nat_gateway_update(self):
+        env = self._create_gateway(
+            "cli-private-nat-update", "cli-private-nat-before-update"
+        )
+        gateway_id = env["gateway"]["id"]
+
+        updated = json.loads(
+            self.openstack(
+                "privatenat gateway update "
+                "--description cli-private-nat-after-update "
+                "{gateway_id} -f json".format(gateway_id=gateway_id)
+            )
+        )
+        self.assertEqual(gateway_id, updated["id"])
+        self.assertEqual("cli-private-nat-after-update", updated["description"])
+
+        shown = json.loads(
+            self.openstack("privatenat gateway show -f json " + gateway_id)
+        )
+        self.assertEqual("cli-private-nat-after-update", shown["description"])
+
+    def test_private_nat_gateway_delete(self):
+        env = self._create_gateway("cli-private-nat-delete", "cli-private-nat-delete")
+        gateway_id = env["gateway"]["id"]
+        gateway = self.conn.natv3.get_private_nat_gateway(gateway_id)
+
+        self.openstack("privatenat gateway delete " + gateway_id)
+        resource.wait_for_delete(self.conn.natv3, gateway, 2, 120)
+
+        listed = json.loads(self.openstack("privatenat gateway list -f json "))
+        self.assertFalse(any(item["Id"] == gateway_id for item in listed))

otcextensions/tests/functional/privatenat.py

@@ -0,0 +1,132 @@
+# 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 uuid
+
+from openstack import exceptions as sdk_exceptions
+from openstack import resource
+from otcextensions.tests.functional import base
+
+
+class PrivateNatEnvironmentMixin(object):
+    def _private_transit_ip_id(self):
+        transit_ip_id = base._get_resource_value("private_transit_ip_id", None)
+        if not transit_ip_id:
+            self.skipTest("functional.private_transit_ip_id is required")
+        return transit_ip_id
+
+    def _create_private_nat_network_stack(self, prefix):
+        cidr = "192.168.0.0/16"
+        suffix = uuid.uuid4().hex[:8]
+        vpc_name = "{prefix}-vpc-{suffix}".format(prefix=prefix, suffix=suffix)
+        subnet_name = "{prefix}-subnet-{suffix}".format(prefix=prefix, suffix=suffix)
+
+        vpc = self.conn.vpc.create_vpc(name=vpc_name, cidr=cidr)
+        gw, _ = cidr.split("/")
+        subnet = self.conn.vpc.create_subnet(
+            name=subnet_name,
+            vpc_id=vpc.id,
+            cidr=vpc.cidr,
+            gateway_ip=gw[:-2] + ".1",
+            dns_list=["100.125.4.25", "100.125.129.199"],
+        )
+        resource.wait_for_status(self.conn.vpc, subnet, "ACTIVE", None, 2, 60)
+
+        return {
+            "vpc": vpc,
+            "subnet": subnet,
+            "network_id": subnet.neutron_network_id,
+        }
+
+    def _cleanup_private_nat_network_stack(self, stack):
+        subnet = stack.get("subnet")
+        vpc = stack.get("vpc")
+
+        if subnet:
+            try:
+                subnet = self.conn.vpc.get_subnet(subnet.id)
+            except sdk_exceptions.ResourceNotFound:
+                subnet = None
+            if subnet:
+                resource.wait_for_status(self.conn.vpc, subnet, "ACTIVE", None, 2, 60)
+                self.conn.vpc.delete_subnet(subnet, ignore_missing=True)
+                resource.wait_for_delete(self.conn.vpc, subnet, 2, 120)
+
+        if vpc:
+            try:
+                vpc = self.conn.vpc.get_vpc(vpc.id)
+            except sdk_exceptions.ResourceNotFound:
+                vpc = None
+            if vpc:
+                self.conn.vpc.delete_vpc(vpc, ignore_missing=True)
+                resource.wait_for_delete(self.conn.vpc, vpc, 2, 120)
+
+    def _create_private_nat_gateway(self, subnet_id, prefix):
+        gateway = self.conn.natv3.create_private_nat_gateway(
+            name="{prefix}-gateway-{suffix}".format(
+                prefix=prefix, suffix=uuid.uuid4().hex[:8]
+            ),
+            spec="Small",
+            downlink_vpcs=[{"virsubnet_id": subnet_id}],
+        )
+        resource.wait_for_status(self.conn.natv3, gateway, "ACTIVE", None, 2, 120)
+        return gateway
+
+    def _delete_private_nat_gateway(self, gateway):
+        try:
+            gateway = self.conn.natv3.get_private_nat_gateway(gateway.id)
+        except sdk_exceptions.ResourceNotFound:
+            return
+
+        self.conn.natv3.delete_private_nat_gateway(gateway, ignore_missing=True)
+        resource.wait_for_delete(self.conn.natv3, gateway, 2, 120)
+
+    def _create_private_nat_port(self, network_id, prefix):
+        if hasattr(self, "create_port"):
+            return self.create_port(network_id, prefix=prefix)
+        return self.conn.network.create_port(
+            name="{prefix}-port-{suffix}".format(
+                prefix=prefix, suffix=uuid.uuid4().hex[:8]
+            ),
+            network_id=network_id,
+        )
+
+    def _delete_private_nat_port(self, port):
+        if hasattr(self, "destroy_port"):
+            try:
+                self.destroy_port(port)
+            except sdk_exceptions.ResourceNotFound:
+                return
+            return
+        self.conn.network.delete_port(port, ignore_missing=True)
+
+    def _prepare_private_nat_environment(self, prefix):
+        env = {
+            "transit_ip_id": self._private_transit_ip_id(),
+        }
+
+        stack = self._prepare_private_nat_gateway_environment(prefix)
+        env["stack"] = stack["stack"]
+        env["gateway"] = stack["gateway"]
+
+        port = self._create_private_nat_port(stack["stack"]["network_id"], prefix)
+        self.addCleanup(self._delete_private_nat_port, port)
+        env["port"] = port
+        return env
+
+    def _prepare_private_nat_gateway_environment(self, prefix):
+        stack = self._create_private_nat_network_stack(prefix)
+        self.addCleanup(self._cleanup_private_nat_network_stack, stack)
+
+        gateway = self._create_private_nat_gateway(stack["subnet"].id, prefix)
+        self.addCleanup(self._delete_private_nat_gateway, gateway)
+        return {"stack": stack, "gateway": gateway}

otcextensions/tests/functional/sdk/dns/v2/test_zone.py

@@ -20,7 +20,7 @@
 class TestZone(TestDns):
     def setUp(self):
         super(TestZone, self).setUp()
-        self._cleanup_routers()
+        self.cleanup_stale_routers("sdk-dns-test-add-router-")
         self.zone = None
         self.networks = []
 
@@ -52,32 +52,6 @@ def tearDown(self):
 
         super(TestZone, self).tearDown()
 
-    def _cleanup_routers(self):
-        """Clean up router resources if they are not deleted for any reason."""
-        for router in self.conn.network.routers():
-            if router.name and router.name.startswith("sdk-dns-test-add-router-"):
-                try:
-                    ports = list(self.conn.network.ports(device_id=router.id))
-                    for port in ports:
-                        fixed_ips = getattr(port, "fixed_ips", []) or []
-                        for fixed_ip in fixed_ips:
-                            subnet_id = fixed_ip.get("subnet_id")
-                            if subnet_id:
-                                try:
-                                    self.conn.network.remove_interface_from_router(
-                                        router, subnet_id=subnet_id
-                                    )
-                                except openstack.exceptions.SDKException:
-                                    pass
-
-                    self.conn.network.delete_router(router, ignore_missing=True)
-                except openstack.exceptions.SDKException as e:
-                    _logger.warning(
-                        "Failed to cleanup stale router %s: %s",
-                        router.id,
-                        str(e),
-                    )
-
     def _create_zone(self, zone_name, router_id=None, zone_type="public"):
         attrs = {"name": zone_name}