Add private snat rule create endpoint

Author: shhovhan

doc/source/sdk/guides/privatenat.rst

@@ -129,3 +129,12 @@ the output with query parameters.
 
 .. literalinclude:: ../examples/natv3/list_private_snat_rules.py
    :lines: 16-23
+
+Create Private SNAT Rule
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+This interface is used to create an SNAT rule.
+:class:`~otcextensions.sdk.natv3.v3.snat.PrivateSnat`.
+
+.. literalinclude:: ../examples/natv3/create_private_snat_rule.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, private_snat_rules
+  :members: private_nat_gateways, get_private_nat_gateway, private_dnat_rules, create_private_dnat_rule, private_snat_rules, create_private_snat_rule

examples/natv3/create_private_snat_rule.py

@@ -0,0 +1,30 @@
+#!/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 SNAT Rule
+"""
+
+import openstack
+
+openstack.enable_logging(True)
+conn = openstack.connect(cloud="otc")
+
+snat_rule_attrs = {
+    "gateway_id": "80da6f26-94eb-4537-97f0-5a56f4d04cfb",
+    "virsubnet_id": "5b9ea497-727d-4ad0-a99e-3984b3f5aaed",
+    "transit_ip_ids": ["36a3049a-1682-48b3-b1cf-cb986a3350ef"],
+    "description": "my_snat_rule01",
+}
+
+snat_rule = conn.natv3.create_private_snat_rule(**snat_rule_attrs)
+print(snat_rule)

otcextensions/osclient/privatenat/v3/snat.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)
+
+
 def _get_transit_ip_addresses(rule):
     return ", ".join(
         association.transit_ip_address
@@ -190,3 +197,84 @@ def take_action(self, parsed_args):
                 for s in data
             ),
         )
+
+
+class CreatePrivateSnatRule(command.ShowOne):
+
+    _description = _("Create a private SNAT rule.")
+
+    def get_parser(self, prog_name):
+        parser = super(CreatePrivateSnatRule, 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(
+            "--cidr",
+            metavar="<cidr>",
+            help=_("Specifies the CIDR block that matches the SNAT rule."),
+        )
+        parser.add_argument(
+            "--virsubnet-id",
+            metavar="<virsubnet_id>",
+            help=_("Specifies the ID of the subnet that matches the SNAT rule."),
+        )
+        parser.add_argument(
+            "--description",
+            metavar="<description>",
+            help=_("Provides supplementary information about the SNAT rule."),
+        )
+        parser.add_argument(
+            "--transit-ip-id",
+            metavar="<transit_ip_id>",
+            action="append",
+            dest="transit_ip_ids",
+            required=True,
+            help=_(
+                "Specifies the transit IP address ID. "
+                "Repeat to associate multiple transit IPs."
+            ),
+        )
+        return parser
+
+    def _build_attrs(self, parsed_args):
+        attrs = {
+            "gateway_id": parsed_args.gateway_id,
+            "transit_ip_ids": parsed_args.transit_ip_ids,
+        }
+
+        if len(parsed_args.transit_ip_ids) > 20:
+            raise exceptions.CommandError(
+                _("A maximum number of 20 --transit-ip-id values is allowed.")
+            )
+
+        if parsed_args.cidr and parsed_args.virsubnet_id:
+            raise exceptions.CommandError(
+                _("Specify either --cidr or --virsubnet-id, but not both.")
+            )
+
+        if not parsed_args.cidr and not parsed_args.virsubnet_id:
+            raise exceptions.CommandError(
+                _("One of --cidr or --virsubnet-id must be specified.")
+            )
+
+        if parsed_args.cidr:
+            attrs["cidr"] = parsed_args.cidr
+        if parsed_args.virsubnet_id:
+            attrs["virsubnet_id"] = parsed_args.virsubnet_id
+        if parsed_args.description is not None:
+            attrs["description"] = parsed_args.description
+
+        return attrs
+
+    def take_action(self, parsed_args):
+        client = self.app.client_manager.privatenat
+        attrs = self._build_attrs(parsed_args)
+        obj = client.create_private_snat_rule(**attrs)
+
+        display_columns, columns = _get_columns(obj)
+        data = utils.get_item_properties(obj, columns)
+        return display_columns, data

otcextensions/sdk/natv3/v3/_proxy.py

@@ -179,3 +179,16 @@ def private_snat_rules(self, **query):
             :class:`~otcextensions.sdk.natv3.v3.snat.PrivateSnat`
         """
         return self._list(_snat.PrivateSnat, **query)
+
+    def create_private_snat_rule(self, **attrs):
+        """Create a private SNAT rule from attributes.
+
+        :param dict attrs: Keyword arguments which will be used to create
+            a :class:`~otcextensions.sdk.natv3.v3.snat.PrivateSnat`,
+            comprised of the properties on the PrivateSnat class.
+
+        :returns: The result of private SNAT rule creation.
+
+        :rtype: :class:`~otcextensions.sdk.natv3.v3.snat.PrivateSnat`
+        """
+        return self._create(_snat.PrivateSnat, **attrs)

otcextensions/sdk/natv3/v3/snat.py

@@ -27,6 +27,7 @@ class PrivateSnat(resource.Resource):
     base_path = "/private-nat/snat-rules"
 
     # capabilities
+    allow_create = True
     allow_list = True
 
     _query_mapping = resource.QueryParameters(
@@ -62,6 +63,10 @@ class PrivateSnat(resource.Resource):
     #: Provides supplementary information about the SNAT rule.
     description = resource.Body("description")
 
+    #: Specifies the IDs of associated transit IP addresses.
+    #: Accepted on create and update requests.
+    transit_ip_ids = resource.Body("transit_ip_ids", type=list)
+
     #: Specifies the list of details of associated transit IP addresses.
     transit_ip_associations = resource.Body(
         "transit_ip_associations", type=list, list_type=AssociatedTransitIp

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

@@ -119,6 +119,7 @@ def generate(cls):
                     "transit_ip_address": "172.20.1.10",
                 }
             ],
+            "transit_ip_ids": [],
             "enterprise_project_id": "ep-" + uuid.uuid4().hex,
             "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),
             "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),

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

@@ -12,6 +12,8 @@
 
 from unittest import mock
 
+from osc_lib import exceptions
+
 from otcextensions.osclient.privatenat.v3 import snat
 from otcextensions.tests.unit.osclient.privatenat.v3 import fakes
 
@@ -128,3 +130,130 @@ def test_list_with_supported_filters(self):
         )
         self.assertEqual(self.columns, columns)
         self.assertEqual(self.data, list(data))
+
+
+class TestCreatePrivateSnatRule(fakes.TestPrivateNat):
+    _data = fakes.FakePrivateSnatRule.create_one()
+
+    def setUp(self):
+        super(TestCreatePrivateSnatRule, self).setUp()
+        self.cmd = snat.CreatePrivateSnatRule(self.app, None)
+        self.client.create_private_snat_rule = mock.Mock(return_value=self._data)
+
+    def test_create_with_virsubnet(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--virsubnet-id",
+            self._data.virsubnet_id,
+            "--transit-ip-id",
+            "tip-1",
+            "--transit-ip-id",
+            "tip-2",
+            "--description",
+            self._data.description,
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("virsubnet_id", self._data.virsubnet_id),
+            ("transit_ip_ids", ["tip-1", "tip-2"]),
+            ("description", self._data.description),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.client.create_private_snat_rule.assert_called_once_with(
+            gateway_id=self._data.gateway_id,
+            virsubnet_id=self._data.virsubnet_id,
+            transit_ip_ids=["tip-1", "tip-2"],
+            description=self._data.description,
+        )
+        self.assertEqual(len(columns), len(data))
+        self.assertIn("id", columns)
+
+    def test_create_with_cidr(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--cidr",
+            "10.1.1.64/30",
+            "--transit-ip-id",
+            "tip-1",
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("cidr", "10.1.1.64/30"),
+            ("transit_ip_ids", ["tip-1"]),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        self.cmd.take_action(parsed_args)
+
+        self.client.create_private_snat_rule.assert_called_once_with(
+            gateway_id=self._data.gateway_id,
+            cidr="10.1.1.64/30",
+            transit_ip_ids=["tip-1"],
+        )
+
+    def test_create_requires_cidr_or_virsubnet(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--transit-ip-id",
+            "tip-1",
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("transit_ip_ids", ["tip-1"]),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        with self.assertRaises(exceptions.CommandError):
+            self.cmd.take_action(parsed_args)
+
+    def test_create_both_cidr_and_virsubnet(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--cidr",
+            "10.1.1.64/30",
+            "--virsubnet-id",
+            self._data.virsubnet_id,
+            "--transit-ip-id",
+            "tip-1",
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("cidr", "10.1.1.64/30"),
+            ("virsubnet_id", self._data.virsubnet_id),
+            ("transit_ip_ids", ["tip-1"]),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        with self.assertRaises(exceptions.CommandError):
+            self.cmd.take_action(parsed_args)
+
+    def test_create_more_than_20_transit_ip_ids(self):
+        arglist = [
+            "--gateway-id",
+            self._data.gateway_id,
+            "--virsubnet-id",
+            self._data.virsubnet_id,
+        ]
+        verifylist = [
+            ("gateway_id", self._data.gateway_id),
+            ("virsubnet_id", self._data.virsubnet_id),
+        ]
+
+        for i in range(21):
+            arglist.extend(["--transit-ip-id", "tip-%s" % i])
+
+        verifylist.append(("transit_ip_ids", ["tip-%s" % i for i in range(21)]))
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        with self.assertRaises(exceptions.CommandError):
+            self.cmd.take_action(parsed_args)

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

@@ -69,3 +69,6 @@ class TestPrivateSnat(TestNatv3Proxy):
 
     def test_private_snat_rules(self):
         self.verify_list(self.proxy.private_snat_rules, snat.PrivateSnat)
+
+    def test_create_private_snat_rule(self):
+        self.verify_create(self.proxy.create_private_snat_rule, snat.PrivateSnat)

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

@@ -42,6 +42,7 @@ def test_basic(self):
         self.assertEqual("snat_rule", sot.resource_key)
         self.assertEqual("snat_rules", sot.resources_key)
         self.assertEqual("/private-nat/snat-rules", sot.base_path)
+        self.assertTrue(sot.allow_create)
         self.assertTrue(sot.allow_list)
 
     def test_make_it(self):
@@ -52,6 +53,7 @@ def test_make_it(self):
         self.assertEqual(EXAMPLE["gateway_id"], sot.gateway_id)
         self.assertEqual(EXAMPLE["cidr"], sot.cidr)
         self.assertEqual(EXAMPLE["virsubnet_id"], sot.virsubnet_id)
+        self.assertIsNone(sot.transit_ip_ids)
         self.assertEqual(1, len(sot.transit_ip_associations))
         self.assertIsInstance(sot.transit_ip_associations[0], snat.AssociatedTransitIp)
         self.assertEqual(