from unittest to pytest

Author: Laure-di

scaleway/scaleway/instance/v1/custom_api.py

@@ -1,3 +1,4 @@
+import time
 from typing import Optional, Dict, cast
 
 from requests import Response
@@ -8,6 +9,8 @@
 from .custom_marshalling import marshal_GetServerUserDataRequest
 from .custom_types import GetServerUserDataRequest, GetAllServerUserDataResponse
 
+max_retry = 10
+interval = 0.01
 
 class InstanceUtilsV1API(InstanceV1API):
     """
@@ -125,3 +128,16 @@ def set_all_server_user_data(
             )
 
         return None
+
+    def wait_test_instance_server(self, server_id, zone):
+        int = interval
+        for i in range(1, max_retry):
+            int *= i
+            s = InstanceUtilsV1API.get_server(self, zone=zone, server_id=server_id)
+
+            if s.server.state == "running" or s.server.state == "stopped":
+                return s
+
+            time.sleep(int)
+
+        return self._throw_on_error()

scaleway/scaleway/instance/v1/tests/test_instance_sdk.py

@@ -1,13 +1,13 @@
 import uuid
-import time
 from typing import Generator
 
 import pytest
 
+from scaleway.instance.v1.custom_api import InstanceUtilsV1API
+from scaleway_async.instance.v1 import AttachServerVolumeRequestVolumeType
 from scaleway_core.client import Client
-from scaleway.instance.v1.api import InstanceV1API
 from scaleway.instance.v1.types import VolumeVolumeType, BootType, Server, VolumeServerTemplate
-from scaleway.block.v1alpha1 import BlockV1Alpha1API, Volume
+from scaleway.block.v1alpha1 import BlockV1Alpha1API
 from scaleway.block.v1alpha1.types import Volume, CreateVolumeRequestFromEmpty
 from vcr_config import scw_vcr
 
@@ -18,52 +18,28 @@
 commercial_type = "DEV1-S"
 zone = "fr-par-1"
 
-
-@pytest.fixture
+@pytest.fixture(scope="module")
 @scw_vcr.use_cassette
-def apis() -> Generator[tuple[InstanceV1API, BlockV1Alpha1API, str, Server | None, list[Volume]], None, None]:
+def instance_block_api() -> tuple[InstanceUtilsV1API, BlockV1Alpha1API]:
     client = Client.from_config_file_and_env()
-    instanceAPI = InstanceV1API(client, bypass_validation=True)
-    blockAPI = BlockV1Alpha1API(client, bypass_validation=True)
-
-    created_server: Server | None = None
-    created_volumes: list[Volume] = []
-
-    yield instanceAPI, blockAPI, zone, created_server, created_volumes
-
-    for volume in created_volumes:
-        if created_server:
-            instanceAPI.detach_server_volume(
-                server_id=created_server.id, volume_id=volume.id
-            )
-        blockAPI.wait_for_volume(volume_id=volume.id, zone=zone)
-        wait_test_instance_server(instanceAPI, created_server.id, zone)
-        blockAPI.delete_volume(volume_id=volume.id)
-
-    if created_server:
-        instanceAPI.delete_server(zone=zone, server_id=created_server.id)
-
-@scw_vcr.use_cassette
-def wait_test_instance_server(instanceAPI: InstanceV1API, server_id: str, zone: str) -> Server:
-    delay = interval
-    for i in range(1, max_retry):
-        delay *= i
-        s = instanceAPI.get_server(zone=zone, server_id=server_id)
-        if s.server.state in ("running", "stopped"):
-            return s.server
-        time.sleep(delay)
-    pytest.fail("Server did not reach 'running' or 'stopped' state in time.")
+    instance_api = InstanceUtilsV1API(client)
+    block_api = BlockV1Alpha1API(client)
+    return instance_api, block_api
 
+@pytest.fixture(scope="module")
 @scw_vcr.use_cassette
-def create_test_instance_server(instanceAPI: InstanceV1API, zone: str) -> Server:
+def instance_volume(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API]) -> Generator[
+    tuple[Server | None, list[Volume]], None, None]:
+    instance_api, block_api = instance_block_api
+    volumes_list: list[Volume] = []
     volumes = {
         "0": VolumeServerTemplate(
             volume_type=VolumeVolumeType.L_SSD,
             size=volume_size,
             boot=False,
         )
     }
-    server = instanceAPI._create_server(
+    instance = instance_api._create_server(
         commercial_type=commercial_type,
         zone=zone,
         name=server_name,
@@ -73,47 +49,60 @@ def create_test_instance_server(instanceAPI: InstanceV1API, zone: str) -> Server
         boot_type=BootType.LOCAL,
         image="c00ae53c-1e29-4087-a384-47f3c5c1cd84",
     )
-    wait_test_instance_server(instanceAPI, server.server.id, zone)
-    return server.server
-
-@scw_vcr.use_cassette
-def create_test_from_empty_volume(blockAPI: BlockV1Alpha1API, number: int, zone: str) -> list[Volume]:
-    volumes: list[Volume] = []
-    for _ in range(number):
-        volume = blockAPI.create_volume(
-            from_empty=CreateVolumeRequestFromEmpty(size=1_000_000_000),
-        )
-        blockAPI.wait_for_volume(volume_id=volume.id, zone=zone)
-        volumes.append(volume)
-    return volumes
+    instance_api.wait_test_instance_server(instance.server.id, zone=zone)
+    yield instance.server, volumes_list
+    for volume in volumes_list:
+        instance_api.detach_server_volume(server_id=instance.server.id, volume_id=volume.id, zone=zone)
+        block_api.wait_for_volume(volume_id=volume.id, zone=zone)
+        block_api.delete_volume(volume_id=volume.id, zone=zone)
+    instance_api.delete_server(server_id=instance.server.id, zone=zone)
 
 @scw_vcr.use_cassette
-def test_attach_additional_volume(apis):
-    instanceAPI, blockAPI, zone, created_server, created_volumes = apis
+def test_attach_additionnal_volume(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API], instance_volume: tuple[Server, list[Volume]]) -> None:
+    instance_api, block_api = instance_block_api
+    server, volumes_list = instance_volume
+    additional_volume = block_api.create_volume(
+        from_empty=CreateVolumeRequestFromEmpty(size=volume_size),
+    )
+    updated_volume = block_api.wait_for_volume(volume_id=additional_volume.id, zone=zone)
+    volumes_list.append(updated_volume)
+    assert updated_volume.size == volume_size
+    assert updated_volume.id is not None
 
-    server = create_test_instance_server(instanceAPI, zone)
+    instance_api.attach_server_volume(server_id=server.id, volume_id=additional_volume.id, zone=zone, volume_type=AttachServerVolumeRequestVolumeType.SBS_VOLUME)
+    instance_api.wait_test_instance_server(server_id=server.id, zone=zone)
+    block_api.wait_for_volume(volume_id=additional_volume.id, zone=zone)
 
-    additional_volumes = create_test_from_empty_volume(blockAPI, 1, zone)
-    created_volumes.extend(additional_volumes)
+    server_details = instance_api.get_server(server_id=server.id, zone=zone).server
+    attached_volume_ids = [v.id for v in server_details.volumes.values()]
+    assert additional_volume.id in attached_volume_ids
 
-    additional_volume = additional_volumes[0]
+@scw_vcr.use_cassette
+def test_list_server(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API]) -> None:
+    instance_api, block_api = instance_block_api
+    servers = instance_api.list_servers(zone=zone)
+    assert len(servers.servers) == 1
+    assert servers.servers[0].name == server_name
 
-    assert server.id is not None
-    assert server.zone == zone
+@scw_vcr.use_cassette
+def test_create_new_server(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API]) -> None:
+    instance_api, block_api = instance_block_api
+    server = instance_api._create_server(
+        commercial_type=commercial_type,
+        zone=zone,
+        name=server_name,
+        dynamic_ip_required=False,
+        volumes={},
+        protected=False,
+        boot_type=BootType.LOCAL,
+    )
+    assert server.server.id is not None
+    assert server.server.name == server_name
 
-    assert additional_volume.id is not None
-    assert additional_volume.size == 1_000_000_000
+    servers = instance_api.list_servers(zone=zone)
+    assert len(servers.servers) == 2
 
-    instanceAPI.attach_server_volume(
-        server_id=server.id,
-        volume_id=additional_volume.id,
-        volume_type=VolumeVolumeType.SBS_VOLUME,
-    )
+    instance_api.delete_server(server_id=server.server.id, zone=zone)
 
-    blockAPI.wait_for_volume(volume_id=additional_volume.id, zone=zone)
 
-    updated_server = instanceAPI.get_server(zone=zone, server_id=server.id)
-    attached_volumes = updated_server.server.volumes or {}
-    attached_volume_ids = [v.id for v in attached_volumes.values()]
 
-    assert additional_volume.id in attached_volume_ids