Adding basic functional test for DMS instance

otcextensions/tests/functional/sdk/dms/v1/test_instance.py

@@ -0,0 +1,160 @@
+# 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 openstack
+from openstack import exceptions
+from openstack import resource
+
+
+from otcextensions.tests.functional import base
+import uuid
+import random
+
+_logger = openstack._log.setup_logging('openstack')
+
+
+class TestInstance(base.BaseFunctionalTest):
+    UUID = uuid.uuid4().hex[:8]
+    INSTANCE_NAME = 'sdk-test-dms-' + UUID
+    ROUTER_NAME = 'sdk-test-router-' + UUID
+    NET_NAME = 'sdk-test-net-' + UUID
+    SUBNET_NAME = 'sdk-test-subnet-' + UUID
+    SG_NAME = 'sdk-test-sg-' + UUID
+    ROUTER_ID = None
+    NET_ID = None
+    SUBNET_ID = None
+    SG_ID = None
+    instances = []
+
+    def setUp(self):
+        super(TestInstance, self).setUp()
+        self._initialize_network()
+        products = list(self.conn.dms.products())
+        products_ids = [product.product_id for product in products]
+        product_id = random.choice(products_ids)
+        availability_zone_names = [
+            product.availability_zones for product in products
+            if product_id == product.product_id
+        ]
+        availability_zone_name = random.choice(
+            sum(availability_zone_names, [])
+        )
+        availability_zone_id = [
+            availability_zone.id
+            for availability_zone in self.conn.dms.availability_zones()
+            if availability_zone_name == availability_zone.code
+        ]
+        try:
+            self.instance = self.conn.dms.create_instance(
+                name=self.INSTANCE_NAME,
+                engine="kafka", engine_version="2.3.0", storage=4800,
+                router_id=self.ROUTER_ID, network_id=self.NET_ID,
+                security_group_id=self.SG_ID,
+                availability_zones=availability_zone_id,
+                product_id=product_id,
+                storage_spec_code="dms.physical.storage.ultra"
+            )
+        except exceptions.DuplicateResource:
+            self.instance = self.conn.dms.find_instance(
+                alias=self.INSTANCE_NAME
+            )
+        resource.wait_for_status(
+            session=self.conn.dms,
+            resource=self.instance,
+            status='RUNNING',
+            failures=['CREATEFAILED'],
+            interval=5,
+            wait=900)
+        self.assertIn('instance_id', self.instance)
+        self.instances.append(self.instance)
+
+    def tearDown(self):
+        super(TestInstance, self).tearDown()
+        try:
+            for instance in self.instances:
+                if instance.id:
+                    self.conn.dms.delete_instance(instance)
+                    resource.wait_for_delete(
+                        session=self.conn.dms,
+                        resource=instance,
+                        interval=2,
+                        wait=60)
+        except openstack.exceptions.SDKException as e:
+            _logger.warning('Got exception during clearing resources %s'
+                            % e.message)
+        self._deinitialize_network()
+
+    def test_list(self):
+        self.all_instances = list(self.conn.dms.instances())
+        self.assertGreaterEqual(len(self.all_instances), 0)
+        if len(self.all_instances) > 0:
+            instance = self.all_instances[0]
+            instance = self.conn.dms.get_instance(instance=instance.id)
+            self.assertIsNotNone(instance)
+
+    def _initialize_network(self):
+        self.cidr = '192.168.0.0/16'
+        self.ipv4 = 4
+
+        sg = self.conn.network.create_security_group(name=self.SG_NAME)
+        self.assertEqual(self.SG_NAME, sg.name)
+        TestInstance.SG_ID = sg.id
+
+        network = self.conn.network.create_network(name=self.NET_NAME)
+        self.assertEqual(self.NET_NAME, network.name)
+        TestInstance.NET_ID = network.id
+
+        subnet = self.conn.network.create_subnet(
+            name=self.SUBNET_NAME,
+            ip_version=self.ipv4,
+            network_id=TestInstance.NET_ID,
+            cidr=self.cidr
+        )
+        self.assertEqual(self.SUBNET_NAME, subnet.name)
+        TestInstance.SUBNET_ID = subnet.id
+
+        router = self.conn.network.create_router(name=self.ROUTER_NAME)
+        self.assertEqual(self.ROUTER_NAME, router.name)
+        TestInstance.ROUTER_ID = router.id
+        interface = router.add_interface(self.conn.network,
+                                         subnet_id=TestInstance.SUBNET_ID)
+        self.assertEqual(interface['subnet_id'], TestInstance.SUBNET_ID)
+        self.assertIn('port_id', interface)
+        # add CIDR to Router for compatibility
+        service = self.conn.identity.find_service('vpc')
+        endpoints = list(self.conn.identity.endpoints(service_id=service.id))
+        endpoint_url = endpoints[0].url
+        endpoint = endpoint_url[:endpoint_url.rfind("/") + 1]
+        cidr = self.conn.compute.put(
+            endpoint + self.conn.session.get_project_id() + '/vpcs/'
+            + self.ROUTER_ID, data='{"vpc": {"cidr": "192.168.0.0/16"}}',
+            headers={'content-type': 'application/json'})
+        self.assertIn('cidr', cidr.json()['vpc'])
+        self.assertIn('192.168.0.0/16', cidr.json()['vpc']['cidr'])
+
+    def _deinitialize_network(self):
+        router = self.conn.network.get_router(TestInstance.ROUTER_ID)
+        interface = router.remove_interface(self.conn.network,
+                                            subnet_id=TestInstance.SUBNET_ID)
+        self.assertEqual(interface['subnet_id'], TestInstance.SUBNET_ID)
+        self.assertIn('port_id', interface)
+        sot = self.conn.network.delete_router(
+            TestInstance.ROUTER_ID, ignore_missing=False)
+        self.assertIsNone(sot)
+        sot = self.conn.network.delete_subnet(
+            TestInstance.SUBNET_ID, ignore_missing=False)
+        self.assertIsNone(sot)
+        sot = self.conn.network.delete_network(
+            TestInstance.NET_ID, ignore_missing=False)
+        self.assertIsNone(sot)
+        sot = self.conn.network.delete_security_group(
+            TestInstance.SG_ID, ignore_missing=False)
+        self.assertIsNone(sot)