Low-level Compute v2 API: security group rules

Dean Troyer · Dean Troyer · commit 1bf6706ad162 · 2017-04-11T02:08:57.000-05:00
api.compute.APIv2 security group rule functions.

novaclient 8.0 is now released without support for the previously
deprecated nova-net functions, so include a new low-level REST
implementation of the removed APIs.

Change-Id: Ieabd61113bc6d3562738686f52bb06aa84fca765
diff --git a/openstackclient/api/compute_v2.py b/openstackclient/api/compute_v2.py
@@ -19,6 +19,12 @@
 from osc_lib.i18n import _
 
 
+# TODO(dtroyer): Mingrate this to osc-lib
+class InvalidValue(Exception):
+    """An argument value is not valid: wrong type, out of range, etc"""
+    message = "Supplied value is not valid"
+
+
 class APIv2(api.BaseAPI):
     """Compute v2 API"""
 
@@ -27,6 +33,29 @@ def __init__(self, **kwargs):
 
     # Overrides
 
+    def _check_integer(self, value, msg=None):
+        """Attempt to convert value to an integer
+
+        Raises InvalidValue on failure
+
+        :param value:
+            Convert this to an integer.  None is converted to 0 (zero).
+        :param msg:
+            An alternate message for the exception, must include exactly
+            one substitution to receive the attempted value.
+        """
+
+        if value is None:
+            return 0
+
+        try:
+            value = int(value)
+        except (TypeError, ValueError):
+            if not msg:
+                msg = "%s is not an integer" % value
+            raise InvalidValue(msg)
+        return value
+
     # TODO(dtroyer): Override find() until these fixes get into an osc-lib
     #                minimum release
     def find(
@@ -209,3 +238,71 @@ def security_group_set(
                 json={'security_group': security_group},
             ).json()['security_group']
         return None
+
+    # Security Group Rules
+
+    def security_group_rule_create(
+        self,
+        security_group_id=None,
+        ip_protocol=None,
+        from_port=None,
+        to_port=None,
+        remote_ip=None,
+        remote_group=None,
+    ):
+        """Create a new security group rule
+
+        https://developer.openstack.org/api-ref/compute/#create-security-group-rule
+
+        :param string security_group_id:
+            Security group ID
+        :param ip_protocol:
+            IP protocol, 'tcp', 'udp' or 'icmp'
+        :param from_port:
+            Source port
+        :param to_port:
+            Destination port
+        :param remote_ip:
+            Source IP address in CIDR notation
+        :param remote_group:
+            Remote security group
+        """
+
+        url = "/os-security-group-rules"
+
+        if ip_protocol.lower() not in ['icmp', 'tcp', 'udp']:
+            raise InvalidValue(
+                "%(s) is not one of 'icmp', 'tcp', or 'udp'" % ip_protocol
+            )
+
+        params = {
+            'parent_group_id': security_group_id,
+            'ip_protocol': ip_protocol,
+            'from_port': self._check_integer(from_port),
+            'to_port': self._check_integer(to_port),
+            'cidr': remote_ip,
+            'group_id': remote_group,
+        }
+
+        return self.create(
+            url,
+            json={'security_group_rule': params},
+        )['security_group_rule']
+
+    def security_group_rule_delete(
+        self,
+        security_group_rule_id=None,
+    ):
+        """Delete a security group rule
+
+        https://developer.openstack.org/api-ref/compute/#delete-security-group-rule
+
+        :param string security_group_rule_id:
+            Security group rule ID
+        """
+
+        url = "/os-security-group-rules"
+        if security_group_rule_id is not None:
+            return self.delete('/%s/%s' % (url, security_group_rule_id))
+
+        return None
diff --git a/openstackclient/network/v2/security_group_rule.py b/openstackclient/network/v2/security_group_rule.py
@@ -376,15 +376,15 @@ def take_action_compute(self, client, parsed_args):
         else:
             remote_ip = '0.0.0.0/0'
 
-        obj = client.security_group_rules.create(
-            group['id'],
-            protocol,
-            from_port,
-            to_port,
-            remote_ip,
-            parsed_args.remote_group,
+        obj = client.api.security_group_rule_create(
+            security_group_id=group['id'],
+            ip_protocol=protocol,
+            from_port=from_port,
+            to_port=to_port,
+            remote_ip=remote_ip,
+            remote_group=parsed_args.remote_group,
         )
-        return _format_security_group_rule_show(obj._info)
+        return _format_security_group_rule_show(obj)
 
 
 class DeleteSecurityGroupRule(common.NetworkAndComputeDelete):
@@ -409,7 +409,7 @@ def take_action_network(self, client, parsed_args):
         client.delete_security_group_rule(obj)
 
     def take_action_compute(self, client, parsed_args):
-        client.security_group_rules.delete(self.r)
+        client.api.security_group_rule_delete(self.r)
 
 
 class ListSecurityGroupRule(common.NetworkAndComputeLister):
diff --git a/openstackclient/tests/unit/api/test_compute_v2.py b/openstackclient/tests/unit/api/test_compute_v2.py
@@ -226,3 +226,84 @@ def test_security_group_set_options_name(self):
             security_group='sg2',
             description='desc2')
         self.assertEqual(self.FAKE_SECURITY_GROUP_RESP_2, ret)
+
+
+class TestSecurityGroupRule(TestComputeAPIv2):
+
+    FAKE_SECURITY_GROUP_RULE_RESP = {
+        'id': '1',
+        'name': 'sgr1',
+        'tenant_id': 'proj-1',
+        'ip_protocol': 'TCP',
+        'from_port': 1,
+        'to_port': 22,
+        'group': {},
+        # 'ip_range': ,
+        # 'cidr': ,
+        # 'parent_group_id': ,
+    }
+
+    def test_security_group_create_no_options(self):
+        self.requests_mock.register_uri(
+            'POST',
+            FAKE_URL + '/os-security-group-rules',
+            json={'security_group_rule': self.FAKE_SECURITY_GROUP_RULE_RESP},
+            status_code=200,
+        )
+        ret = self.api.security_group_rule_create(
+            security_group_id='1',
+            ip_protocol='tcp',
+        )
+        self.assertEqual(self.FAKE_SECURITY_GROUP_RULE_RESP, ret)
+
+    def test_security_group_create_options(self):
+        self.requests_mock.register_uri(
+            'POST',
+            FAKE_URL + '/os-security-group-rules',
+            json={'security_group_rule': self.FAKE_SECURITY_GROUP_RULE_RESP},
+            status_code=200,
+        )
+        ret = self.api.security_group_rule_create(
+            security_group_id='1',
+            ip_protocol='tcp',
+            from_port=22,
+            to_port=22,
+            remote_ip='1.2.3.4/24',
+        )
+        self.assertEqual(self.FAKE_SECURITY_GROUP_RULE_RESP, ret)
+
+    def test_security_group_create_port_errors(self):
+        self.requests_mock.register_uri(
+            'POST',
+            FAKE_URL + '/os-security-group-rules',
+            json={'security_group_rule': self.FAKE_SECURITY_GROUP_RULE_RESP},
+            status_code=200,
+        )
+        self.assertRaises(
+            compute.InvalidValue,
+            self.api.security_group_rule_create,
+            security_group_id='1',
+            ip_protocol='tcp',
+            from_port='',
+            to_port=22,
+            remote_ip='1.2.3.4/24',
+        )
+        self.assertRaises(
+            compute.InvalidValue,
+            self.api.security_group_rule_create,
+            security_group_id='1',
+            ip_protocol='tcp',
+            from_port=0,
+            to_port=[],
+            remote_ip='1.2.3.4/24',
+        )
+
+    def test_security_group_rule_delete(self):
+        self.requests_mock.register_uri(
+            'DELETE',
+            FAKE_URL + '/os-security-group-rules/1',
+            status_code=202,
+        )
+        ret = self.api.security_group_rule_delete('1')
+        self.assertEqual(202, ret.status_code)
+        self.assertEqual("", ret.text)
diff --git a/openstackclient/tests/unit/compute/v2/fakes.py b/openstackclient/tests/unit/compute/v2/fakes.py
@@ -556,10 +556,7 @@ def create_one_security_group_rule(attrs=None):
         # Overwrite default attributes.
         security_group_rule_attrs.update(attrs)
 
-        security_group_rule = fakes.FakeResource(
-            info=copy.deepcopy(security_group_rule_attrs),
-            loaded=True)
-        return security_group_rule
+        return security_group_rule_attrs
 
     @staticmethod
     def create_security_group_rules(attrs=None, count=2):
diff --git a/openstackclient/tests/unit/network/v2/test_security_group_compute.py b/openstackclient/tests/unit/network/v2/test_security_group_compute.py
@@ -356,7 +356,7 @@ class TestShowSecurityGroupCompute(TestSecurityGroupCompute):
     # The security group to be shown.
     _security_group = \
         compute_fakes.FakeSecurityGroup.create_one_security_group(
-            attrs={'rules': [_security_group_rule._info]}
+            attrs={'rules': [_security_group_rule]}
         )
 
     columns = (
@@ -373,7 +373,7 @@ class TestShowSecurityGroupCompute(TestSecurityGroupCompute):
         _security_group['name'],
         _security_group['tenant_id'],
         security_group._format_compute_security_group_rules(
-            [_security_group_rule._info]),
+            [_security_group_rule]),
     )
 
     def setUp(self):
diff --git a/openstackclient/tests/unit/network/v2/test_security_group_rule_compute.py b/openstackclient/tests/unit/network/v2/test_security_group_rule_compute.py
