Add transit ip delte endpoint

Author: shhovhan

doc/source/sdk/guides/privatenat.rst

@@ -199,3 +199,12 @@ This interface is used to assign a transit IP address.
 
 .. literalinclude:: ../examples/natv3/create_private_transit_ip.py
    :lines: 16-28
+
+Delete Private Transit IP Address
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This interface is used to delete a transit IP address.
+:class:`~otcextensions.sdk.natv3.v3.transit_ip.PrivateTransitIp`.
+
+.. literalinclude:: ../examples/natv3/delete_private_transit_ip.py
+   :lines: 16-22

doc/source/sdk/proxies/privatenat.rst

@@ -16,4 +16,7 @@ Private NAT Gateway Operations
 
 .. autoclass:: otcextensions.sdk.natv3.v3._proxy.Proxy
   :noindex:
-  :members: private_nat_gateways, get_private_nat_gateway, private_dnat_rules, create_private_dnat_rule, update_private_dnat_rule, delete_private_dnat_rule, private_snat_rules, create_private_snat_rule, update_private_snat_rule, delete_private_snat_rule, private_transit_ips, get_private_transit_ip, create_private_transit_ip
+  :members: private_nat_gateways, get_private_nat_gateway, private_dnat_rules, create_private_dnat_rule,
+    update_private_dnat_rule, delete_private_dnat_rule, private_snat_rules, create_private_snat_rule,
+    update_private_snat_rule, delete_private_snat_rule, private_transit_ips, get_private_transit_ip,
+    create_private_transit_ip, delete_private_transit_ip

examples/natv3/delete_private_transit_ip.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+# 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.
+"""
+Delete a private transit IP address.
+"""
+
+import openstack
+
+openstack.enable_logging(True)
+conn = openstack.connect(cloud="otc")
+
+conn.natv3.delete_private_transit_ip("a2845109-3b2f-4627-b08f-09a726c0a6e7")

otcextensions/osclient/privatenat/v3/transit_ip.py

@@ -14,6 +14,7 @@
 
 import logging
 
+from osc_lib import exceptions
 from osc_lib import utils
 from osc_lib.command import command
 
@@ -262,3 +263,27 @@ def take_action(self, parsed_args):
         data = utils.get_item_properties(obj, columns)
 
         return display_columns, data
+
+
+class DeletePrivateTransitIp(command.Command):
+    _description = _("Delete a private transit IP address.")
+
+    def get_parser(self, prog_name):
+        parser = super(DeletePrivateTransitIp, self).get_parser(prog_name)
+        parser.add_argument(
+            "transit_ip",
+            metavar="<transit_ip>",
+            help=_("Specifies the transit IP address ID."),
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        client = self.app.client_manager.privatenat
+
+        try:
+            client.delete_private_transit_ip(parsed_args.transit_ip)
+        except Exception as e:
+            msg = _(
+                "Failed to delete private transit IP address with ID '%(id)s': %(e)s"
+            ) % {"id": parsed_args.transit_ip, "e": e}
+            raise exceptions.CommandError(msg)

otcextensions/sdk/natv3/v3/_proxy.py

@@ -277,3 +277,21 @@ def create_private_transit_ip(self, **attrs):
         :rtype: :class:`~otcextensions.sdk.natv3.v3.transit_ip.PrivateTransitIp`
         """
         return self._create(_transit_ip.PrivateTransitIp, **attrs)
+
+    def delete_private_transit_ip(self, transit_ip, ignore_missing=True):
+        """Delete a private NAT transit IP address.
+
+        :param transit_ip: The value can be the ID of a transit IP address or a
+            :class:`~otcextensions.sdk.natv3.v3.transit_ip.PrivateTransitIp`
+            instance.
+        :param bool ignore_missing: When set to ``False``,
+            :class:`~openstack.exceptions.ResourceNotFound` will be raised when
+            the transit IP address does not exist. When set to ``True``, no
+            exception will be raised when attempting to delete a nonexistent
+            transit IP address.
+
+        :returns: ``None``
+        """
+        return self._delete(
+            _transit_ip.PrivateTransitIp, transit_ip, ignore_missing=ignore_missing
+        )

otcextensions/sdk/natv3/v3/transit_ip.py

@@ -23,9 +23,10 @@ class PrivateTransitIp(resource.Resource):
     base_path = "/private-nat/transit-ips"
 
     # capabilities
-    allow_fetch = True
     allow_list = True
+    allow_fetch = True
     allow_create = True
+    allow_delete = True
 
     _query_mapping = resource.QueryParameters(
         "limit",

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

@@ -14,13 +14,76 @@
 
 from openstackclient.tests.functional import base
 
+from openstack import connection
+from openstack import exceptions as sdk_exceptions
+from otcextensions import sdk
+from otcextensions.tests.functional import base as sdk_base
+from otcextensions.tests.functional.privatenat import PrivateNatEnvironmentMixin
 
-class TestPrivateTransitIp(base.TestCase):
+
+class TestPrivateTransitIp(PrivateNatEnvironmentMixin, base.TestCase):
     """Functional Tests for private NAT transit IP addresses"""
 
     def setUp(self):
         super(TestPrivateTransitIp, self).setUp()
+        self.conn = connection.Connection(config=sdk_base.TEST_CLOUD_REGION)
+        sdk.register_otc_extensions(self.conn)
 
     def test_privatenat_transit_ip_list(self):
         json_output = json.loads(self.openstack("privatenat transit ip list -f json "))
         self.assertIsNotNone(json_output)
+
+    def test_privatenat_transit_ip_create(self):
+        env = self._prepare_private_nat_subnet_environment(
+            "cli-private-transit-ip-create"
+        )
+
+        created = json.loads(
+            self.openstack(
+                "privatenat transit ip create "
+                "--virsubnet-id {virsubnet_id} "
+                "--tags test=cli "
+                "-f json".format(virsubnet_id=env["subnet"].id)
+            )
+        )
+        self.addCleanup(self._delete_private_transit_ip, created["id"])
+
+        self.assertIsNotNone(created)
+        self.assertEqual("ACTIVE", created["status"])
+
+    def test_privatenat_transit_ip_show(self):
+        env = self._prepare_private_nat_subnet_environment(
+            "cli-private-transit-ip-show"
+        )
+
+        created = json.loads(
+            self.openstack(
+                "privatenat transit ip create "
+                "--virsubnet-id {virsubnet_id} "
+                "-f json".format(virsubnet_id=env["subnet"].id)
+            )
+        )
+        self.addCleanup(self._delete_private_transit_ip, created["id"])
+
+        shown = json.loads(
+            self.openstack("privatenat transit ip show -f json " + created["id"])
+        )
+        self.assertEqual(created["id"], shown["id"])
+
+    def test_privatenat_transit_ip_delete(self):
+        env = self._prepare_private_nat_subnet_environment(
+            "cli-private-transit-ip-delete"
+        )
+
+        created = json.loads(
+            self.openstack(
+                "privatenat transit ip create "
+                "--virsubnet-id {virsubnet_id} "
+                "-f json".format(virsubnet_id=env["subnet"].id)
+            )
+        )
+
+        self.openstack("privatenat transit ip delete " + created["id"])
+
+        with self.assertRaises(sdk_exceptions.ResourceNotFound):
+            self.conn.natv3.get_private_transit_ip(created["id"])

otcextensions/tests/functional/privatenat.py

@@ -14,24 +14,28 @@
 
 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 _delete_private_transit_ip(self, transit_ip_id):
+        self.conn.natv3.delete_private_transit_ip(transit_ip_id, ignore_missing=True)
 
-    def _create_private_nat_network_stack(self, prefix):
-        cidr = "192.168.0.0/16"
-        suffix = uuid.uuid4().hex[:8]
+    def _create_private_nat_vpc(self, prefix, suffix):
         vpc_name = "{prefix}-vpc-{suffix}".format(prefix=prefix, suffix=suffix)
-        subnet_name = "{prefix}-subnet-{suffix}".format(prefix=prefix, suffix=suffix)
+        return self.conn.vpc.create_vpc(name=vpc_name, cidr="192.168.0.0/16")
+
+    def _delete_private_nat_vpc(self, vpc):
+        try:
+            vpc = self.conn.vpc.get_vpc(vpc.id)
+        except sdk_exceptions.ResourceNotFound:
+            return
+
+        self.conn.vpc.delete_vpc(vpc, ignore_missing=True)
+        resource.wait_for_delete(self.conn.vpc, vpc, 2, 120)
 
-        vpc = self.conn.vpc.create_vpc(name=vpc_name, cidr=cidr)
-        gw, _ = cidr.split("/")
+    def _create_private_nat_subnet(self, vpc, prefix, suffix):
+        subnet_name = "{prefix}-subnet-{suffix}".format(prefix=prefix, suffix=suffix)
+        gw, _ = vpc.cidr.split("/")
         subnet = self.conn.vpc.create_subnet(
             name=subnet_name,
             vpc_id=vpc.id,
@@ -40,6 +44,22 @@ def _create_private_nat_network_stack(self, prefix):
             dns_list=["100.125.4.25", "100.125.129.199"],
         )
         resource.wait_for_status(self.conn.vpc, subnet, "ACTIVE", None, 2, 60)
+        return subnet
+
+    def _delete_private_nat_subnet(self, subnet):
+        try:
+            subnet = self.conn.vpc.get_subnet(subnet.id)
+        except sdk_exceptions.ResourceNotFound:
+            return
+
+        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)
+
+    def _create_private_nat_network_stack(self, prefix):
+        suffix = uuid.uuid4().hex[:8]
+        vpc = self._create_private_nat_vpc(prefix, suffix)
+        subnet = self._create_private_nat_subnet(vpc, prefix, suffix)
 
         return {
             "vpc": vpc,
@@ -52,23 +72,19 @@ def _cleanup_private_nat_network_stack(self, stack):
         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)
+            self._delete_private_nat_subnet(subnet)
 
         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)
+            self._delete_private_nat_vpc(vpc)
+
+    def _prepare_private_nat_subnet_environment(self, prefix):
+        suffix = uuid.uuid4().hex[:8]
+        vpc = self._create_private_nat_vpc(prefix, suffix)
+        self.addCleanup(self._delete_private_nat_vpc, vpc)
+
+        subnet = self._create_private_nat_subnet(vpc, prefix, suffix)
+        self.addCleanup(self._delete_private_nat_subnet, subnet)
+        return {"vpc": vpc, "subnet": subnet}
 
     def _create_private_nat_gateway(self, subnet_id, prefix):
         gateway = self.conn.natv3.create_private_nat_gateway(
@@ -90,39 +106,6 @@ def _delete_private_nat_gateway(self, gateway):
         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)

otcextensions/tests/functional/sdk/natv3/v3/test_private_transit_ip.py

@@ -11,21 +11,53 @@
 # under the License.
 
 import openstack
+from openstack import exceptions as sdk_exceptions
 from otcextensions.tests.functional import base
+from otcextensions.tests.functional.privatenat import PrivateNatEnvironmentMixin
 
 _logger = openstack._log.setup_logging("openstack")
 
 
-class TestPrivateTransitIp(base.BaseFunctionalTest):
-
+class TestPrivateTransitIp(PrivateNatEnvironmentMixin, base.BaseFunctionalTest):
     def test_list_transit_ips(self):
         transit_ips = list(self.conn.natv3.private_transit_ips())
         self.assertGreaterEqual(len(transit_ips), 0)
 
     def test_get_private_transit_ip(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")
+        env = self._prepare_private_nat_subnet_environment("sdk-private-transit-ip-get")
+
+        created = self.conn.natv3.create_private_transit_ip(
+            virsubnet_id=env["subnet"].id
+        )
+        self.addCleanup(self._delete_private_transit_ip, created.id)
+
+        transit_ip = self.conn.natv3.get_private_transit_ip(created.id)
+        self.assertEqual(created.id, transit_ip.id)
+
+    def test_create_private_transit_ip(self):
+        env = self._prepare_private_nat_subnet_environment(
+            "sdk-private-transit-ip-create"
+        )
+
+        transit_ip = self.conn.natv3.create_private_transit_ip(
+            virsubnet_id=env["subnet"].id,
+            tags=[{"key": "test", "value": "sdk"}],
+        )
+        self.addCleanup(self._delete_private_transit_ip, transit_ip.id)
+
+        self.assertIsNotNone(transit_ip.id)
+        self.assertEqual("ACTIVE", transit_ip.status)
+
+    def test_delete_private_transit_ip(self):
+        env = self._prepare_private_nat_subnet_environment(
+            "sdk-private-transit-ip-delete"
+        )
+
+        transit_ip = self.conn.natv3.create_private_transit_ip(
+            virsubnet_id=env["subnet"].id
+        )
+
+        self.conn.natv3.delete_private_transit_ip(transit_ip, ignore_missing=False)
 
-        transit_ip = self.conn.natv3.get_private_transit_ip(transit_ip_id)
-        self.assertEqual(transit_ip_id, transit_ip.id)
+        with self.assertRaises(sdk_exceptions.ResourceNotFound):
+            self.conn.natv3.get_private_transit_ip(transit_ip.id)