Add create endpoint of transit ip addresses

Author: shhovhan

doc/source/sdk/guides/privatenat.rst

@@ -181,3 +181,12 @@ the output with query parameters.
 
 .. literalinclude:: ../examples/natv3/list_private_transit_ips.py
    :lines: 16-23
+
+Create Private Transit IP Address
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This interface is used to assign a transit IP address.
+:class:`~otcextensions.sdk.natv3.v3.transit_ip.PrivateTransitIp`.
+
+.. literalinclude:: ../examples/natv3/create_private_transit_ip.py
+   :lines: 16-28

doc/source/sdk/proxies/privatenat.rst

@@ -16,4 +16,4 @@ 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
+  :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, create_private_transit_ip

examples/natv3/create_private_transit_ip.py

@@ -0,0 +1,28 @@
+#!/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.
+"""
+Assign a private transit IP address.
+"""
+
+import openstack
+
+openstack.enable_logging(True)
+conn = openstack.connect(cloud="otc")
+
+transit_ip = conn.natv3.create_private_transit_ip(
+    virsubnet_id="2759da7b-8015-404c-ae0a-a389007b0e2a",
+    ip_address="192.168.1.68",
+    enterprise_project_id="2759da7b-8015-404c-ae0a-a389007b0e2a",
+    tags=[{"key": "key1", "value": "value1"}],
+)
+print(transit_ip)

otcextensions/osclient/privatenat/v3/transit_ip.py

@@ -17,11 +17,18 @@
 from osc_lib import utils
 from osc_lib.command import command
 
+from otcextensions.common import sdk_utils
+from otcextensions.common.utils import normalize_tags
 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 ListPrivateTransitIps(command.Lister):
 
     _description = _("List private transit IP addresses.")
@@ -176,3 +183,60 @@ def take_action(self, parsed_args):
                 for t in data
             ),
         )
+
+
+class CreatePrivateTransitIp(command.ShowOne):
+    _description = _("Assign a private transit IP address.")
+
+    def get_parser(self, prog_name):
+        parser = super(CreatePrivateTransitIp, self).get_parser(prog_name)
+        parser.add_argument(
+            "--virsubnet-id",
+            metavar="<virsubnet_id>",
+            required=True,
+            help=_("Specifies the subnet ID of the current project."),
+        )
+        parser.add_argument(
+            "--ip-address",
+            metavar="<ip_address>",
+            help=_("Specifies the transit IP address."),
+        )
+        parser.add_argument(
+            "--enterprise-project-id",
+            metavar="<enterprise_project_id>",
+            help=_(
+                "Specifies the enterprise project ID associated with the "
+                "transit IP address."
+            ),
+        )
+        parser.add_argument(
+            "--tags",
+            metavar="<tags>",
+            action="append",
+            help=_(
+                "Specifies the tag list in KEY=VALUE format. "
+                "Repeat for multiple values."
+            ),
+        )
+        return parser
+
+    def take_action(self, parsed_args):
+        client = self.app.client_manager.privatenat
+
+        attrs = {
+            "virsubnet_id": parsed_args.virsubnet_id,
+        }
+
+        if parsed_args.ip_address:
+            attrs["ip_address"] = parsed_args.ip_address
+        if parsed_args.enterprise_project_id:
+            attrs["enterprise_project_id"] = parsed_args.enterprise_project_id
+        if parsed_args.tags:
+            attrs["tags"] = normalize_tags(parsed_args.tags)
+
+        obj = client.create_private_transit_ip(**attrs)
+
+        display_columns, columns = _get_columns(obj)
+        data = utils.get_item_properties(obj, columns)
+
+        return display_columns, data

otcextensions/sdk/natv3/v3/_proxy.py

@@ -251,3 +251,16 @@ def private_transit_ips(self, **query):
             :class:`~otcextensions.sdk.natv3.v3.transit_ip.PrivateTransitIp`
         """
         return self._list(_transit_ip.PrivateTransitIp, **query)
+
+    def create_private_transit_ip(self, **attrs):
+        """Assign a private NAT transit IP address.
+
+        :param dict attrs: Keyword arguments which will be used to assign
+            a :class:`~otcextensions.sdk.natv3.v3.transit_ip.PrivateTransitIp`,
+            comprised of the properties on the PrivateTransitIp class.
+
+        :returns: The assigned private transit IP address.
+
+        :rtype: :class:`~otcextensions.sdk.natv3.v3.transit_ip.PrivateTransitIp`
+        """
+        return self._create(_transit_ip.PrivateTransitIp, **attrs)

otcextensions/sdk/natv3/v3/transit_ip.py

@@ -24,6 +24,7 @@ class PrivateTransitIp(resource.Resource):
 
     # capabilities
     allow_list = True
+    allow_create = True
 
     _query_mapping = resource.QueryParameters(
         "limit",

otcextensions/tests/unit/osclient/privatenat/v3/test_transit_ip.py

@@ -128,3 +128,58 @@ def test_list_with_supported_filters(self):
         )
         self.assertEqual(self.columns, columns)
         self.assertEqual(self.data, list(data))
+
+
+class TestCreatePrivateTransitIp(fakes.TestPrivateNat):
+    _data = fakes.FakePrivateTransitIp.create_one()
+
+    def setUp(self):
+        super(TestCreatePrivateTransitIp, self).setUp()
+        self.cmd = transit_ip.CreatePrivateTransitIp(self.app, None)
+        self.client.create_private_transit_ip = mock.Mock(return_value=self._data)
+
+    def test_create(self):
+        arglist = [
+            "--virsubnet-id",
+            self._data.virsubnet_id,
+            "--ip-address",
+            self._data.ip_address,
+            "--enterprise-project-id",
+            self._data.enterprise_project_id,
+            "--tags",
+            "key1=value1",
+        ]
+        verifylist = [
+            ("virsubnet_id", self._data.virsubnet_id),
+            ("ip_address", self._data.ip_address),
+            ("enterprise_project_id", self._data.enterprise_project_id),
+            ("tags", ["key1=value1"]),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.client.create_private_transit_ip.assert_called_once_with(
+            virsubnet_id=self._data.virsubnet_id,
+            ip_address=self._data.ip_address,
+            enterprise_project_id=self._data.enterprise_project_id,
+            tags=[{"key": "key1", "value": "value1"}],
+        )
+        self.assertEqual(len(columns), len(data))
+        self.assertIn("id", columns)
+
+    def test_create_minimal(self):
+        arglist = [
+            "--virsubnet-id",
+            self._data.virsubnet_id,
+        ]
+        verifylist = [
+            ("virsubnet_id", self._data.virsubnet_id),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        self.cmd.take_action(parsed_args)
+
+        self.client.create_private_transit_ip.assert_called_once_with(
+            virsubnet_id=self._data.virsubnet_id,
+        )

otcextensions/tests/unit/sdk/natv3/v3/test_proxy.py

@@ -88,3 +88,8 @@ class TestPrivateTransitIp(TestNatv3Proxy):
 
     def test_private_transit_ips(self):
         self.verify_list(self.proxy.private_transit_ips, transit_ip.PrivateTransitIp)
+
+    def test_create_private_transit_ip(self):
+        self.verify_create(
+            self.proxy.create_private_transit_ip, transit_ip.PrivateTransitIp
+        )

otcextensions/tests/unit/sdk/natv3/v3/test_transit_ip.py

@@ -42,6 +42,7 @@ def test_basic(self):
         self.assertEqual("transit_ip", sot.resource_key)
         self.assertEqual("transit_ips", sot.resources_key)
         self.assertEqual("/private-nat/transit-ips", sot.base_path)
+        self.assertTrue(sot.allow_create)
         self.assertTrue(sot.allow_list)
 
     def test_make_it(self):

setup.cfg

@@ -197,6 +197,7 @@ openstack.privatenat.v3 =
   privatenat_dnat_rule_update = otcextensions.osclient.privatenat.v3.dnat:UpdatePrivateDnatRule
   privatenat_dnat_rule_delete = otcextensions.osclient.privatenat.v3.dnat:DeletePrivateDnatRule
   privatenat_transit_ip_list = otcextensions.osclient.privatenat.v3.transit_ip:ListPrivateTransitIps
+  privatenat_transit_ip_create = otcextensions.osclient.privatenat.v3.transit_ip:CreatePrivateTransitIp
 
 openstack.auto_scaling.v1 =
   as_group_list = otcextensions.osclient.auto_scaling.v1.group:ListAutoScalingGroup