Add API tests for subnet pool prefix operations

This introduces API tests for the subnetpool_prefix_ops
extension. These tests assert proper addition and removal of
prefixes against subnet pools under the various scenarios that
are supported.

Related-Bug: #1792901
Depends-On: https://review.opendev.org/#/c/648197/
Change-Id: I51564669fc1113556b0927296fa9dd2a8806bce8
diff --git a/.zuul.yaml b/.zuul.yaml
index fbfcad6..dbae6e4 100644
--- a/.zuul.yaml
+++ b/.zuul.yaml
@@ -74,6 +74,7 @@
         - standard-attr-tag
         - standard-attr-timestamp
         - subnet_allocation
+        - subnetpool-prefix-ops
         - trunk
         - trunk-details
         - uplink-status-propagation
diff --git a/neutron_tempest_plugin/api/test_subnetpool_prefix_ops.py b/neutron_tempest_plugin/api/test_subnetpool_prefix_ops.py
new file mode 100644
index 0000000..49cce5b
--- /dev/null
+++ b/neutron_tempest_plugin/api/test_subnetpool_prefix_ops.py
@@ -0,0 +1,97 @@
+# Copyright 2019 SUSE LLC
+# All Rights Reserved.
+#
+#    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.
+
+import netaddr
+from tempest.common import utils
+from tempest.lib import decorators
+
+from neutron_tempest_plugin.api import test_subnetpools
+
+SUBNETPOOL_NAME = 'smoke-subnetpool'
+SUBNET_NAME = 'smoke-subnet'
+
+
+class SubnetPoolPrefixOpsTestMixin(object):
+
+    def _compare_prefix_lists(self, list_expected, list_observed):
+        expected_set = netaddr.IPSet(iterable=list_expected)
+        observed_set = netaddr.IPSet(iterable=list_observed)
+
+        # compact the IPSet's
+        expected_set.compact()
+        observed_set.compact()
+
+        self.assertEqual(expected_set, observed_set)
+
+    @decorators.idempotent_id('b1d56d1f-2818-44ee-b6a3-3c1327c25318')
+    @utils.requires_ext(extension='subnetpool-prefix-ops', service='network')
+    def test_add_remove_prefix(self):
+        created_subnetpool = self._create_subnetpool()
+        req_body = {'prefixes': self.prefixes_to_add}
+
+        # Add a prefix to the subnet pool
+        resp = self.client.add_subnetpool_prefix(created_subnetpool['id'],
+                                                 **req_body)
+        self._compare_prefix_lists(self.prefixes + self.prefixes_to_add,
+                                   resp['prefixes'])
+
+        # Remove the prefix from the subnet pool
+        resp = self.client.remove_subnetpool_prefix(created_subnetpool['id'],
+                                                    **req_body)
+        self._compare_prefix_lists(self.prefixes, resp['prefixes'])
+
+    @decorators.idempotent_id('a36c18fc-10b5-4ebc-ab79-914f826c5bf5')
+    @utils.requires_ext(extension='subnetpool-prefix-ops', service='network')
+    def test_add_overlapping_prefix(self):
+        created_subnetpool = self._create_subnetpool()
+        req_body = {'prefixes': self.overlapping_prefixes}
+
+        # Add an overlapping prefix to the subnet pool
+        resp = self.client.add_subnetpool_prefix(created_subnetpool['id'],
+                                                 **req_body)
+        self._compare_prefix_lists(self.prefixes + self.overlapping_prefixes,
+                                   resp['prefixes'])
+
+
+class SubnetPoolPrefixOpsIpv4Test(test_subnetpools.SubnetPoolsTestBase,
+                                  SubnetPoolPrefixOpsTestMixin):
+
+    prefixes = ['192.168.1.0/24', '10.10.10.0/24']
+    prefixes_to_add = ['192.168.2.0/24']
+    overlapping_prefixes = ['10.10.0.0/16']
+    min_prefixlen = 16
+    ip_version = 4
+
+    @classmethod
+    def resource_setup(cls):
+        super(SubnetPoolPrefixOpsIpv4Test, cls).resource_setup()
+        cls._subnetpool_data = {'prefixes': cls.prefixes,
+                                'min_prefixlen': cls.min_prefixlen}
+
+
+class SubnetPoolPrefixOpsIpv6Test(test_subnetpools.SubnetPoolsTestBase,
+                                  SubnetPoolPrefixOpsTestMixin):
+
+    prefixes = ['2001:db8:1234::/48', '2001:db8:1235::/48']
+    prefixes_to_add = ['2001:db8:4321::/48']
+    overlapping_prefixes = ['2001:db8:1234:1111::/64']
+    min_prefixlen = 48
+    ip_version = 6
+
+    @classmethod
+    def resource_setup(cls):
+        super(SubnetPoolPrefixOpsIpv6Test, cls).resource_setup()
+        cls._subnetpool_data = {'prefixes': cls.prefixes,
+                                'min_prefixlen': cls.min_prefixlen}
diff --git a/neutron_tempest_plugin/services/network/json/network_client.py b/neutron_tempest_plugin/services/network/json/network_client.py
index 422b071..11ba8ef 100644
--- a/neutron_tempest_plugin/services/network/json/network_client.py
+++ b/neutron_tempest_plugin/services/network/json/network_client.py
@@ -227,6 +227,23 @@
         self.expected_success(200, resp.status)
         return service_client.ResponseBody(resp, body)
 
+    def add_subnetpool_prefix(self, id, **kwargs):
+        return self._subnetpool_prefix_operation(id, 'add_prefixes', kwargs)
+
+    def remove_subnetpool_prefix(self, id, **kwargs):
+        return self._subnetpool_prefix_operation(id,
+                                                 'remove_prefixes',
+                                                 kwargs)
+
+    def _subnetpool_prefix_operation(self, id, operation, op_body):
+        uri = self.get_uri("subnetpools")
+        op_prefix_uri = '%s/%s/%s' % (uri, id, operation)
+        body = jsonutils.dumps(op_body)
+        resp, body = self.put(op_prefix_uri, body)
+        body = jsonutils.loads(body)
+        self.expected_success(200, resp.status)
+        return service_client.ResponseBody(resp, body)
+
     # Common methods that are hard to automate
     def create_bulk_network(self, names, shared=False):
         network_list = [{'name': name, 'shared': shared} for name in names]