✅ test: Review Comments

Author: YashGaykar0309

tests/integration_utils.py

@@ -1,7 +1,6 @@
 import base64
 import random
 import string
-import time
 
 def get_random_string(length=10):
     alphabet = string.ascii_lowercase + string.digits
@@ -15,24 +14,6 @@ def get_tagged_name(prefix="osc-sdk-python-test", length=10):
 def log_test_step(message):
     print("[tests] {}".format(message), flush=True)
 
-
-def get_resource_until_ready(
-    fetch, ready, timeout=300, interval=10, failure=None, describe=None
-):
-    deadline = time.time() + timeout
-    last_value = None
-    while time.time() < deadline:
-        last_value = fetch()
-        if describe is not None:
-            log_test_step(describe(last_value))
-        if ready(last_value):
-            return last_value
-        if failure is not None and failure(last_value):
-            raise AssertionError("Resource entered an unexpected state: {}".format(last_value))
-        time.sleep(interval)
-    raise AssertionError("Timed out waiting for resource state: {}".format(last_value))
-
-
 def get_first_item(items, message):
     if not items:
         raise AssertionError(message)

tests/test_load_balancer_backend.py

@@ -32,7 +32,7 @@ def test_load_balancer_backend_lifecycle(self):
                 MaxVmsCount=1,
                 Placement={"SubregionName": subregion_name, "Tenancy": "default"},
                 UserData=get_linux_http_user_data(),
-                VmType="tinav4.c1r1p2",
+                VmType="tinav6.c1r1p2",
             )
             vms = vm_response.get("Vms")
             self.assertIsInstance(vms, list)
@@ -49,7 +49,7 @@ def test_load_balancer_backend_lifecycle(self):
                 log_test_step("VM {} state={}".format(vm_id, vm.get("State")))
                 if vm.get("State") == "running":
                     break
-                if vm.get("State") in ("terminated", "shutting-down"):
+                if vm.get("State") in ("stopped", "terminated", "shutting-down"):
                     self.fail("VM {} entered unexpected state {}".format(vm_id, vm.get("State")))
                 time.sleep(10)
 

tests/test_vm.py

@@ -1,17 +1,8 @@
 import unittest
 import sys
-import time
 
 sys.path.append("..")
 from osc_sdk_python import Gateway
-from tests.integration_utils import (
-    build_name_tag_request,
-    get_first_subregion_name,
-    get_latest_public_ubuntu_image_id,
-    log_test_step,
-    read_single_resource,
-)
-
 
 class TestVm(unittest.TestCase):
     def test_listing(self):
@@ -26,59 +17,5 @@ def test_listing_with_context_manager(self):
             self.assertEqual(type(vms), dict)
             self.assertEqual(type(vms.get("Vms")), list)
 
-    def test_vm_lifecycle(self):
-        gw = Gateway()
-        subregion_name = get_first_subregion_name(gw)
-        image_id = get_latest_public_ubuntu_image_id(gw)
-        log_test_step("Using subregion {} and image {}".format(subregion_name, image_id))
-
-        created_vm_ids = []
-        try:
-            log_test_step("Creating VM")
-            response = gw.CreateVms(
-                ImageId=image_id,
-                MinVmsCount=1,
-                MaxVmsCount=1,
-                Placement={"SubregionName": subregion_name, "Tenancy": "default"},
-                VmType="tinav4.c1r1p2",
-            )
-            vms = response.get("Vms")
-            self.assertIsInstance(vms, list)
-            self.assertEqual(len(vms), 1)
-
-            vm_id = vms[0].get("VmId")
-            self.assertTrue(vm_id)
-            created_vm_ids = [vm_id]
-            log_test_step("Created VM {}".format(vm_id))
-
-            gw.CreateTags(**build_name_tag_request(vm_id))
-            log_test_step("Tagged VM {}".format(vm_id))
-
-            vm = None
-            for _ in range(36):
-                vm = read_single_resource(gw, "ReadVms", "Vms", "VmIds", vm_id)
-                log_test_step("VM {} state={}".format(vm_id, vm.get("State")))
-                if vm.get("State") == "running":
-                    break
-                if vm.get("State") in ("terminated", "shutting-down"):
-                    self.fail("VM {} entered unexpected state {}".format(vm_id, vm.get("State")))
-                time.sleep(10)
-
-            self.assertIsNotNone(vm)
-            self.assertEqual(vm.get("State"), "running")
-
-            self.assertEqual(vm.get("ImageId"), image_id)
-            self.assertTrue(
-                any(tag.get("Key") == "Name" for tag in vm.get("Tags", [])),
-                "expected a Name tag on the VM",
-            )
-            log_test_step("VM {} is running".format(vm_id))
-        finally:
-            if created_vm_ids:
-                time.sleep(1)
-                log_test_step("Deleting VM {}".format(created_vm_ids[0]))
-                gw.DeleteVms(VmIds=created_vm_ids)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_volume.py

@@ -1,16 +1,8 @@
 import unittest
 import sys
-import time
 
 sys.path.append("..")
 from osc_sdk_python import Gateway
-from tests.integration_utils import (
-    build_name_tag_request,
-    get_first_subregion_name,
-    log_test_step,
-    read_single_resource,
-)
-
 
 class TestVolume(unittest.TestCase):
     def test_listing(self):
@@ -19,64 +11,5 @@ def test_listing(self):
         self.assertEqual(type(volumes), dict)
         self.assertEqual(type(volumes.get("Volumes")), list)
 
-    def test_volume_lifecycle(self):
-        gw = Gateway()
-        subregion_name = get_first_subregion_name(gw)
-        log_test_step("Using subregion {}".format(subregion_name))
-        volume_id = None
-        try:
-            log_test_step("Creating volume")
-            response = gw.CreateVolume(Size=10, SubregionName=subregion_name)
-            volume = response.get("Volume")
-            self.assertIsInstance(volume, dict)
-            volume_id = volume.get("VolumeId")
-            self.assertTrue(volume_id)
-            log_test_step("Created volume {}".format(volume_id))
-
-            gw.CreateTags(**build_name_tag_request(volume_id))
-            log_test_step("Tagged volume {}".format(volume_id))
-
-            volume = None
-            for _ in range(30):
-                volume = read_single_resource(gw, "ReadVolumes", "Volumes", "VolumeIds", volume_id)
-                log_test_step("Volume {} state={}".format(volume_id, volume.get("State")))
-                if volume.get("State") == "available":
-                    break
-                if volume.get("State") == "error":
-                    self.fail(
-                        "Volume {} entered unexpected state {}".format(
-                            volume_id, volume.get("State")
-                        )
-                    )
-                time.sleep(10)
-
-            self.assertIsNotNone(volume)
-
-            # volume = get_resource_until_ready(
-            #     fetch=lambda: read_single_resource(
-            #         gw, "ReadVolumes", "Volumes", "VolumeIds", volume_id
-            #     ),
-            #     ready=lambda item: item.get("State") == "available",
-            #     timeout=300,
-            #     interval=10,
-            #     failure=lambda item: item.get("State") == "error",
-            #     describe=lambda item: "Volume {} state={}".format(
-            #         volume_id, item.get("State")
-            #     ),
-            # )
-
-            self.assertEqual(volume.get("State"), "available")
-            self.assertTrue(
-                any(tag.get("Key") == "Name" for tag in volume.get("Tags", [])),
-                "expected a Name tag on the volume",
-            )
-            log_test_step("Volume {} is available".format(volume_id))
-        finally:
-            if volume_id:
-                time.sleep(1)
-                log_test_step("Deleting volume {}".format(volume_id))
-                gw.DeleteVolume(VolumeId=volume_id)
-
-
 if __name__ == "__main__":
     unittest.main()