from unittest to pytest

Author: Laure-di

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

@@ -1,47 +1,119 @@
 import uuid
+import time
+from typing import Generator
 
 import pytest
 
-from scaleway.block.v1 import BlockV1API
-from scaleway.instance.v1 import InstanceV1API, VolumeServerTemplate, VolumeVolumeType, BootType
 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.types import Volume, CreateVolumeRequestFromEmpty
 from vcr_config import scw_vcr
 
 server_name = f"test-sdk-python-{uuid.uuid4().hex[:6]}"
 max_retry = 10
 interval = 0.01
-volume_size = 10000000000
+volume_size = 10_000_000_000
 commercial_type = "DEV1-S"
 zone = "fr-par-1"
-image_id = "c00ae53c-1e29-4087-a384-47f3c5c1cd84"
 
-@pytest.fixture(scope="module")
+
+@pytest.fixture
 @scw_vcr.use_cassette
-def instance_api():
+def apis() -> Generator[tuple[InstanceV1API, BlockV1Alpha1API, str, Server | None, list[Volume]], None, None]:
     client = Client.from_config_file_and_env()
-    api = InstanceV1API(client)
-    return api, client.default_project_id, zone
+    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)
 
-@pytest.fixture(scope="module")
 @scw_vcr.use_cassette
-def block_api():
-    client = Client.from_config_file_and_env()
-    api = BlockV1API(client)
-    return api, client.default_project_id, zone
+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.")
 
 @scw_vcr.use_cassette
-def test_instance_create(instance_api):
-    api, project_id, zone = instance_api
+def create_test_instance_server(instanceAPI: InstanceV1API, zone: str) -> Server:
     volumes = {
         "0": VolumeServerTemplate(
             volume_type=VolumeVolumeType.L_SSD,
             size=volume_size,
             boot=False,
         )
     }
-    instance = api.create_instance(project_id=project_id, zone=zone, name=server_name, dynamic_ip_required=False, volume=volumes, protected=False, boot_type=BootType.LOCAL, image=image_id)
-    assert instance.name == server_name
-    assert instance.project_id == project_id
-    assert instance.zone == zone
-    assert instance.id is not None
+    server = instanceAPI._create_server(
+        commercial_type=commercial_type,
+        zone=zone,
+        name=server_name,
+        dynamic_ip_required=False,
+        volumes=volumes,
+        protected=False,
+        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
+
+@scw_vcr.use_cassette
+def test_attach_additional_volume(apis):
+    instanceAPI, blockAPI, zone, created_server, created_volumes = apis
+
+    server = create_test_instance_server(instanceAPI, zone)
+
+    additional_volumes = create_test_from_empty_volume(blockAPI, 1, zone)
+    created_volumes.extend(additional_volumes)
+
+    additional_volume = additional_volumes[0]
+
+    assert server.id is not None
+    assert server.zone == zone
+
+    assert additional_volume.id is not None
+    assert additional_volume.size == 1_000_000_000
+
+    instanceAPI.attach_server_volume(
+        server_id=server.id,
+        volume_id=additional_volume.id,
+        volume_type=VolumeVolumeType.SBS_VOLUME,
+    )
+
+    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

scaleway/scaleway/vpc/v2/tests/test_vpc_sdk.py

@@ -1,4 +1,4 @@
-from typing import Generator
+from collections.abc import Generator
 
 import pytest
 
@@ -13,17 +13,18 @@
 created_pn_count = 5
 created_vpc_count = 1
 
+# mypy: ignore-errors
 
 @pytest.fixture(scope="module")
-@scw_vcr.use_cassette # type: ignore[misc]
-def vpc_api() -> tuple[VpcV2API, str | None, str]:
+@scw_vcr.use_cassette
+def vpc_api() -> tuple[VpcV2API, str | None, str | None]:
         client = Client.from_config_file_and_env()
         api = VpcV2API(client)
         return api, client.default_project_id, region
 
 
 @pytest.fixture(scope="module")
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def vpc(vpc_api: tuple[VpcV2API, str, str]) -> Generator[VPC, None, None]:
     api, project_id, region = vpc_api
     vpc = api.create_vpc(
@@ -37,7 +38,7 @@ def vpc(vpc_api: tuple[VpcV2API, str, str]) -> Generator[VPC, None, None]:
 
 
 @pytest.fixture
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def private_networks_to_cleanup(vpc_api: tuple[VpcV2API, str, str]) -> Generator[list[PrivateNetwork], None, None]:
     api, _, _ = vpc_api
     items: list[PrivateNetwork] = []
@@ -46,7 +47,7 @@ def private_networks_to_cleanup(vpc_api: tuple[VpcV2API, str, str]) -> Generator
         api.delete_private_network(private_network_id=pn.id)
 
 
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def test_vpc_delete(vpc_api: tuple[VpcV2API, str, str]) -> None:
     api, project_id, region = vpc_api
     vpc: VPC = api.create_vpc(
@@ -65,15 +66,15 @@ def test_vpc_delete(vpc_api: tuple[VpcV2API, str, str]) -> None:
         api.get_vpc(vpc_id=vpc.id)
 
 
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def test_vpc_list(vpc_api: tuple[VpcV2API, str, str]) -> None:
     api, _, region = vpc_api
     vpcs: ListVPCsResponse = api.list_vp_cs(region=region)
     assert isinstance(vpcs.vpcs, list)
     assert vpcs.total_count >= created_vpc_count
 
 
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def test_private_network_create(vpc_api: tuple[VpcV2API, str, str], vpc: VPC , private_networks_to_cleanup: list[PrivateNetwork]) -> None:
     api, project_id, _ = vpc_api
     for i in range(created_pn_count):
@@ -87,31 +88,31 @@ def test_private_network_create(vpc_api: tuple[VpcV2API, str, str], vpc: VPC , p
         assert pn.vpc_id == vpc.id
 
 
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def test_private_network_list(vpc_api: tuple[VpcV2API, str, str]) -> None:
     api, _, region = vpc_api
     networks: ListPrivateNetworksResponse = api.list_private_networks(region=region)
     assert isinstance(networks.private_networks, list)
     assert networks.total_count >= created_pn_count
 
 
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def test_vpc_get(vpc_api: tuple[VpcV2API, str, str], vpc: VPC) -> None:
     api, _, region = vpc_api
     fetched_vpc: VPC = api.get_vpc(vpc_id=vpc.id, region=region)
     assert fetched_vpc is not None
     assert fetched_vpc.id == vpc.id
 
 
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def test_vpc_update(vpc_api: tuple[VpcV2API, str, str], vpc: VPC) -> None:
     api, _, _ = vpc_api
     updated = api.update_vpc(vpc_id=vpc.id, tags=tags)
     assert updated.tags == tags
     assert updated.id == vpc.id
 
 
-@scw_vcr.use_cassette # type: ignore[misc]
+@scw_vcr.use_cassette
 def test_vpc_list_all(vpc_api: tuple[VpcV2API, str, str]) -> None:
     api, _, _ = vpc_api
     vpcs = api.list_vp_cs_all()