Add private dnat rules create endpoint

shhovhan · shhovhan · commit 83e3006f3b34 · 2026-04-09T17:39:27.000+04:00
diff --git a/doc/source/sdk/guides/privatenat.rst b/doc/source/sdk/guides/privatenat.rst
@@ -71,9 +71,18 @@ the IP address mapping and port mapping.
 List Private DNAT Rules
 ^^^^^^^^^^^^^^^^^^^^^^^
 
-This interface is used to query an DNAT rule list and to filter
+This interface is used to query a DNAT rule list and to filter
 the output with query parameters.
 :class:`~otcextensions.sdk.natv3.v3.dnat.PrivateDnat`.
 
 .. literalinclude:: ../examples/natv3/list_private_dnat_rules.py
    :lines: 16-23
+
+Create Private DNAT Rules
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This interface is used to create a DNAT rule.
+:class:`~otcextensions.sdk.natv3.v3.dnat.PrivateDnat`.
+
+.. literalinclude:: ../examples/natv3/create_private_dnat_rules.py
+   :lines: 16-32
diff --git a/examples/natv3/create_private_dnat_rules.py b/examples/natv3/create_private_dnat_rules.py
@@ -0,0 +1,32 @@
+#!/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.
+"""
+Create a Private DNAT Rule
+"""
+
+import openstack
+
+openstack.enable_logging(True)
+conn = openstack.connect(cloud="otc")
+
+dnat_rule_attrs = {
+    "gateway_id": "0adefb29-a6c2-48a5-8637-2be67fa03fec",
+    "transit_ip_id": "3faa719d-6d18-4ccb-a5c7-33e65a09663e",
+    "network_interface_id": "dae9393a-b536-491c-a5a2-72edc1104707",
+    "protocol": "tcp",
+    "internal_service_port": 22,
+    "transit_service_port": 22,
+}
+
+dnat_rule = conn.natv3.create_private_dnat_rule(**dnat_rule_attrs)
+print(dnat_rule)
diff --git a/otcextensions/osclient/privatenat/v3/dnat.py b/otcextensions/osclient/privatenat/v3/dnat.py
@@ -14,14 +14,21 @@
 
 import logging
 
+from osc_lib import exceptions
 from osc_lib import utils
 from osc_lib.command import command
 
+from otcextensions.common import sdk_utils
 from otcextensions.i18n import _
 
 LOG = logging.getLogger(__name__)
 
 
+def _get_columns(item):
+    column_map = {}
+    return sdk_utils.get_osc_show_columns_for_sdk_resource(item, column_map)
+
+
 class ListPrivateDnatRules(command.Lister):
 
     _description = _("List private DNAT rules.")
@@ -240,3 +247,123 @@ def take_action(self, parsed_args):
                 for s in data
             ),
         )
+
+
+class CreatePrivateDnatRule(command.ShowOne):
+
+    _description = _("Create a private DNAT rule.")
+
+    def get_parser(self, prog_name):
+        parser = super(CreatePrivateDnatRule, self).get_parser(prog_name)
+
+        parser.add_argument(
+            "--gateway-id",
+            metavar="<gateway_id>",
+            required=True,
+            help=_("Specifies the private NAT gateway ID."),
+        )
+        parser.add_argument(
+            "--transit-ip-id",
+            metavar="<transit_ip_id>",
+            required=True,
+            help=_("Specifies the transit IP address ID."),
+        )
+        parser.add_argument(
+            "--type",
+            metavar="<type>",
+            choices=["COMPUTE", "VIP", "ELB", "ELBv3", "CUSTOMIZE"],
+            help=_(
+                "Specifies the backend resource type. "
+                "Supported values: COMPUTE, VIP, ELB, ELBv3, CUSTOMIZE."
+            ),
+        )
+        parser.add_argument(
+            "--network-interface-id",
+            metavar="<network_interface_id>",
+            help=_("Specifies the port ID of the backend resource."),
+        )
+        parser.add_argument(
+            "--private-ip-address",
+            metavar="<private_ip_address>",
+            help=_("Specifies the backend private IP address."),
+        )
+        parser.add_argument(
+            "--protocol",
+            metavar="<protocol>",
+            choices=["tcp", "udp", "any", "TCP", "UDP", "ANY"],
+            help=_("Specifies the protocol type. " "Supported values: tcp, udp, any."),
+        )
+        parser.add_argument(
+            "--internal-service-port",
+            metavar="<internal_service_port>",
+            type=int,
+            help=_("Specifies the port number of the backend resource."),
+        )
+        parser.add_argument(
+            "--transit-service-port",
+            metavar="<transit_service_port>",
+            type=int,
+            help=_("Specifies the port number of the transit IP address."),
+        )
+        parser.add_argument(
+            "--description",
+            metavar="<description>",
+            help=_("Provides supplementary information about the DNAT rule."),
+        )
+        parser.add_argument(
+            "--enterprise-project-id",
+            metavar="<enterprise_project_id>",
+            help=_("Specifies the enterprise project ID."),
+        )
+        return parser
+
+    def _build_attrs(self, parsed_args):
+        attrs = {
+            "gateway_id": parsed_args.gateway_id,
+            "transit_ip_id": parsed_args.transit_ip_id,
+        }
+        optional_attrs = (
+            "type",
+            "protocol",
+            "network_interface_id",
+            "private_ip_address",
+            "internal_service_port",
+            "transit_service_port",
+            "description",
+            "enterprise_project_id",
+        )
+
+        if parsed_args.network_interface_id and parsed_args.private_ip_address:
+            raise exceptions.CommandError(
+                _(
+                    "Specify either --network-interface-id or "
+                    "--private-ip-address, but not both."
+                )
+            )
+
+        if not parsed_args.network_interface_id and not parsed_args.private_ip_address:
+            raise exceptions.CommandError(
+                _(
+                    "One of --network-interface-id or --private-ip-address "
+                    "must be specified."
+                )
+            )
+
+        for key in optional_attrs:
+            value = getattr(parsed_args, key, None)
+            if value is not None:
+                if key == "protocol":
+                    attrs["protocol"] = value.lower()
+                else:
+                    attrs[key] = value
+
+        return attrs
+
+    def take_action(self, parsed_args):
+        client = self.app.client_manager.privatenat
+        attrs = self._build_attrs(parsed_args)
+        obj = client.create_private_dnat_rule(**attrs)
+
+        display_columns, columns = _get_columns(obj)
+        data = utils.get_item_properties(obj, columns)
+        return display_columns, data
diff --git a/otcextensions/sdk/natv3/v3/_proxy.py b/otcextensions/sdk/natv3/v3/_proxy.py
@@ -106,3 +106,16 @@ def private_dnat_rules(self, **query):
         :returns: A generator of :class:`~otcextensions.sdk.nat_gateway.v3.dnat.Dnat`
         """
         return self._list(_dnat.PrivateDnat, **query)
+
+    def create_private_dnat_rule(self, **attrs):
+        """Create a private DNAT rule from attributes.
+
+        :param dict attrs: Keyword arguments which will be used to create
+            a :class:`~otcextensions.sdk.natv3.v3.dnat.PrivateDnat`,
+            comprised of the properties on the PrivateDnat class.
+
+        :returns: The result of private DNAT rule creation.
+
+        :rtype: :class:`~otcextensions.sdk.natv3.v3.dnat.PrivateDnat`
+        """
+        return self._create(_dnat.PrivateDnat, **attrs)
diff --git a/otcextensions/sdk/natv3/v3/dnat.py b/otcextensions/sdk/natv3/v3/dnat.py
@@ -18,6 +18,7 @@ class PrivateDnat(resource.Resource):
     base_path = "/private-nat/dnat-rules"
 
     # capabilities
+    allow_create = True
     allow_list = True
 
     _query_mapping = resource.QueryParameters(
diff --git a/otcextensions/tests/unit/osclient/privatenat/v3/fakes.py b/otcextensions/tests/unit/osclient/privatenat/v3/fakes.py
@@ -85,8 +85,8 @@ def generate(cls):
             "description": "test_dnat_rule_description",
             "gateway_id": "private-gw-id-" + uuid.uuid4().hex,
             "transit_ip_id": uuid.uuid4().hex,
-            "enterprise_project_id": "2759da7b-8015-404c-ae0a-a389007b0e2a",
-            "network_interface_id": "dae9393a-b536-491c-a5a2-72edc1104707",
+            "enterprise_project_id": "ep-" + uuid.uuid4().hex,
+            "network_interface_id": "net-" + uuid.uuid4().hex,
             "type": "COMPUTE",
             "protocol": "any",
             "internal_service_port": 0,
diff --git a/otcextensions/tests/unit/osclient/privatenat/v3/test_dnat.py b/otcextensions/tests/unit/osclient/privatenat/v3/test_dnat.py
@@ -12,6 +12,8 @@
 
 from unittest import mock
 
+from osc_lib import exceptions
+
 from otcextensions.osclient.privatenat.v3 import dnat
 from otcextensions.tests.unit.osclient.privatenat.v3 import fakes
 
@@ -158,3 +160,105 @@ def test_list_with_supported_filters(self):
         )
         self.assertEqual(self.columns, columns)
         self.assertEqual(self.data, list(data))
+
+
+class TestCreatePrivateDnatRule(fakes.TestPrivateNat):
+    _data = fakes.FakePrivateDnatRule.create_one()
+
+    def setUp(self):
+        super(TestCreatePrivateDnatRule, self).setUp()
+        self.cmd = dnat.CreatePrivateDnatRule(self.app, None)
+        self.client.create_private_dnat_rule = mock.Mock(return_value=self._data)
+
+    def test_create_with_network_interface(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--transit-ip-id",
+            self._data.transit_ip_id,
+            "--type",
+            "COMPUTE",
+            "--network-interface-id",
+            self._data.network_interface_id,
+            "--protocol",
+            "TCP",
+            "--internal-service-port",
+            "80",
+            "--transit-service-port",
+            "8080",
+            "--description",
+            self._data.description,
+            "--enterprise-project-id",
+            self._data.enterprise_project_id,
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("transit_ip_id", self._data.transit_ip_id),
+            ("type", "COMPUTE"),
+            ("network_interface_id", self._data.network_interface_id),
+            ("protocol", "TCP"),
+            ("internal_service_port", 80),
+            ("transit_service_port", 8080),
+            ("description", self._data.description),
+            ("enterprise_project_id", self._data.enterprise_project_id),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.client.create_private_dnat_rule.assert_called_once_with(
+            gateway_id=self._data.gateway_id,
+            transit_ip_id=self._data.transit_ip_id,
+            type="COMPUTE",
+            network_interface_id=self._data.network_interface_id,
+            protocol="tcp",
+            internal_service_port=80,
+            transit_service_port=8080,
+            description=self._data.description,
+            enterprise_project_id=self._data.enterprise_project_id,
+        )
+        self.assertEqual(len(columns), len(data))
+        self.assertIn("id", columns)
+
+    def test_create_requires_interface_or_private_ip(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--transit-ip-id",
+            self._data.transit_ip_id,
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("transit_ip_id", self._data.transit_ip_id),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        with self.assertRaises(exceptions.CommandError):
+            self.cmd.take_action(parsed_args)
+
+    def test_create_rejects_mixed_args(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--transit-ip-id",
+            self._data.transit_ip_id,
+            "--type",
+            "COMPUTE",
+            "--network-interface-id",
+            self._data.network_interface_id,
+            "--private-ip-address",
+            self._data.private_ip_address,
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("transit_ip_id", self._data.transit_ip_id),
+            ("type", "COMPUTE"),
+            ("network_interface_id", self._data.network_interface_id),
+            ("private_ip_address", self._data.private_ip_address),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        with self.assertRaises(exceptions.CommandError):
+            self.cmd.take_action(parsed_args)
diff --git a/otcextensions/tests/unit/sdk/natv3/v3/test_dnat.py b/otcextensions/tests/unit/sdk/natv3/v3/test_dnat.py
@@ -41,6 +41,7 @@ def test_basic(self):
         self.assertEqual("dnat_rule", sot.resource_key)
         self.assertEqual("dnat_rules", sot.resources_key)
         self.assertEqual("/private-nat/dnat-rules", sot.base_path)
+        self.assertTrue(sot.allow_create)
         self.assertTrue(sot.allow_list)
 
     def test_make_it(self):
diff --git a/otcextensions/tests/unit/sdk/natv3/v3/test_proxy.py b/otcextensions/tests/unit/sdk/natv3/v3/test_proxy.py
@@ -12,6 +12,7 @@
 
 from openstack.tests.unit import test_proxy_base
 from otcextensions.sdk.natv3.v3 import _proxy
+from otcextensions.sdk.natv3.v3 import dnat
 from otcextensions.sdk.natv3.v3 import gateway
 
 
@@ -43,3 +44,12 @@ def test_delete_private_nat_gateway(self):
         self.verify_delete(
             self.proxy.delete_private_nat_gateway, gateway.PrivateNatGateway
         )
+
+
+class TestPrivateDnat(TestNatv3Proxy):
+
+    def test_private_dnat_rules(self):
+        self.verify_list(self.proxy.private_dnat_rules, dnat.PrivateDnat)
+
+    def test_create_private_dnat_rule(self):
+        self.verify_create(self.proxy.create_private_dnat_rule, dnat.PrivateDnat)
diff --git a/setup.cfg b/setup.cfg
@@ -187,6 +187,7 @@ openstack.privatenat.v3 =
   privatenat_gateway_update = otcextensions.osclient.privatenat.v3.private_nat_gateway:UpdatePrivateNatGateway
   privatenat_gateway_delete = otcextensions.osclient.privatenat.v3.private_nat_gateway:DeletePrivateNatGateway
   privatenat_dnat_rule_list = otcextensions.osclient.privatenat.v3.dnat:ListPrivateDnatRules
+  privatenat_dnat_rule_create = otcextensions.osclient.privatenat.v3.dnat:CreatePrivateDnatRule
 
 openstack.auto_scaling.v1 =
   as_group_list = otcextensions.osclient.auto_scaling.v1.group:ListAutoScalingGroup
