add tests to close issues

Author: Laure-di

scaleway/scaleway/instance/v1/custom_api.py

@@ -12,6 +12,7 @@
 max_retry = 10
 interval = 0.01
 
+
 class InstanceUtilsV1API(InstanceV1API):
     """
     This API extends InstanceV1API by adding utility methods for managing Instance resources,

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

@@ -3,10 +3,16 @@
 
 import pytest
 
+from scaleway.instance.v1 import IpType
 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.types import VolumeVolumeType, BootType, Server, VolumeServerTemplate
+from scaleway.instance.v1.types import (
+    VolumeVolumeType,
+    BootType,
+    Server,
+    VolumeServerTemplate,
+)
 from scaleway.block.v1alpha1 import BlockV1Alpha1API
 from scaleway.block.v1alpha1.types import Volume, CreateVolumeRequestFromEmpty
 from vcr_config import scw_vcr
@@ -18,6 +24,9 @@
 commercial_type = "DEV1-S"
 zone = "fr-par-1"
 
+# mypy: ignore-errors
+
+
 @pytest.fixture(scope="module")
 @scw_vcr.use_cassette
 def instance_block_api() -> tuple[InstanceUtilsV1API, BlockV1Alpha1API]:
@@ -26,10 +35,12 @@ def instance_block_api() -> tuple[InstanceUtilsV1API, BlockV1Alpha1API]:
     block_api = BlockV1Alpha1API(client)
     return instance_api, block_api
 
+
 @pytest.fixture(scope="module")
 @scw_vcr.use_cassette
-def instance_volume(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API]) -> Generator[
-    tuple[Server | None, list[Volume]], None, None]:
+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 = {
@@ -52,40 +63,59 @@ def instance_volume(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1A
     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)
+        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_additionnal_volume(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API], instance_volume: tuple[Server, list[Volume]]) -> None:
+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)
+    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
 
-    instance_api.attach_server_volume(server_id=server.id, volume_id=additional_volume.id, zone=zone, volume_type=AttachServerVolumeRequestVolumeType.SBS_VOLUME)
+    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)
 
     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
 
+
 @scw_vcr.use_cassette
-def test_list_server(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API]) -> None:
+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 len(servers.servers) >= 1
     assert servers.servers[0].name == server_name
 
+
 @scw_vcr.use_cassette
-def test_create_new_server(instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API]) -> None:
+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,
@@ -99,10 +129,16 @@ def test_create_new_server(instance_block_api: tuple[InstanceUtilsV1API, BlockV1
     assert server.server.id is not None
     assert server.server.name == server_name
 
-    servers = instance_api.list_servers(zone=zone)
-    assert len(servers.servers) == 2
+    servers = instance_api.list_servers_all(zone=zone)
+    assert len(servers) >= 2
 
     instance_api.delete_server(server_id=server.server.id, zone=zone)
 
 
-
+@scw_vcr.use_cassette
+def test_create_ip(
+    instance_block_api: tuple[InstanceUtilsV1API, BlockV1Alpha1API],
+) -> None:
+    instance_api, block_api = instance_block_api
+    ip = instance_api.create_ip(type_=IpType.ROUTED_IPV6).ip
+    assert ip.id is not None

scaleway/scaleway/k8s/v1/tests/test_k8s_sk.py

@@ -0,0 +1,61 @@
+from typing import Any, Generator
+
+import pytest
+
+from scaleway.k8s.v1 import K8SV1API, CNI
+from scaleway.vpc.v2 import VpcV2API, PrivateNetwork
+from scaleway_core.api import ScalewayException
+from scaleway_core.client import Client
+from vcr_config import scw_vcr
+
+# mypy: ignore-errors
+
+
+@pytest.fixture(scope="module")
+@scw_vcr.use_cassette
+def k8s_api() -> K8SV1API:
+    client = Client.from_config_file_and_env()
+    k8s_api = K8SV1API(client)
+    return k8s_api
+
+
+@pytest.fixture(scope="module")
+@scw_vcr.use_cassette
+def private_network() -> Generator[PrivateNetwork, Any, None]:
+    client = Client.from_config_file_and_env()
+    vpc_api = VpcV2API(client)
+    vpc = vpc_api.create_vpc(enable_routing=False)
+    pn = vpc_api.create_private_network(
+        vpc_id=vpc.id, default_route_propagation_enabled=True
+    )
+    yield pn
+    vpc_api.delete_vpc(vpc_id=vpc.id)
+
+
+def safe_wait_for_cluster(k8s_api: K8SV1API, cluster_id: str):
+    try:
+        return k8s_api.wait_for_cluster(cluster_id=cluster_id)
+    except ScalewayException as e:
+        err_type = getattr(e, "type", None) or getattr(e, "error", None)
+        if err_type == "not_found" or "not found" in str(e).lower():
+            return None
+        raise
+
+
+@scw_vcr.use_cassette
+def test_k8s_cluster_list(k8s_api: K8SV1API, private_network: PrivateNetwork) -> None:
+    cluster = k8s_api.create_cluster(
+        type_="kapsule",
+        version="1.32.7",
+        cni=CNI.CILIUM,
+        description="",
+        private_network_id=private_network.id,
+    )
+    assert cluster.id is not None
+
+    clusters = k8s_api.list_clusters()
+    assert clusters.total_count >= 1
+
+    k8s_api.delete_cluster(cluster_id=cluster.id, with_additional_resources=True)
+    cluster = safe_wait_for_cluster(k8s_api=k8s_api, cluster_id=cluster.id)
+    assert cluster is None

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

@@ -2,7 +2,13 @@
 
 import pytest
 
-from scaleway.vpc.v2 import VpcV2API, VPC, PrivateNetwork, ListVPCsResponse, ListPrivateNetworksResponse
+from scaleway.vpc.v2 import (
+    VpcV2API,
+    VPC,
+    PrivateNetwork,
+    ListVPCsResponse,
+    ListPrivateNetworksResponse,
+)
 from scaleway_core.api import ScalewayException
 from scaleway_core.client import Client
 from scaleway_core.utils import random_name
@@ -15,12 +21,13 @@
 
 # mypy: ignore-errors
 
+
 @pytest.fixture(scope="module")
 @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
+    client = Client.from_config_file_and_env()
+    api = VpcV2API(client)
+    return api, client.default_project_id, region
 
 
 @pytest.fixture(scope="module")
@@ -39,7 +46,9 @@ def vpc(vpc_api: tuple[VpcV2API, str, str]) -> Generator[VPC, None, None]:
 
 @pytest.fixture
 @scw_vcr.use_cassette
-def private_networks_to_cleanup(vpc_api: tuple[VpcV2API, str, str]) -> Generator[list[PrivateNetwork], None, None]:
+def private_networks_to_cleanup(
+    vpc_api: tuple[VpcV2API, str, str],
+) -> Generator[list[PrivateNetwork], None, None]:
     api, _, _ = vpc_api
     items: list[PrivateNetwork] = []
     yield items
@@ -75,7 +84,11 @@ def test_vpc_list(vpc_api: tuple[VpcV2API, str, str]) -> None:
 
 
 @scw_vcr.use_cassette
-def test_private_network_create(vpc_api: tuple[VpcV2API, str, str], vpc: VPC , private_networks_to_cleanup: list[PrivateNetwork]) -> None:
+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):
         pn: PrivateNetwork = api.create_private_network(

scaleway/vcr_config.py

@@ -4,16 +4,22 @@
 import vcr
 import inspect
 
-PYTHON_UPDATE_CASSETTE = os.getenv("PYTHON_UPDATE_CASSETTE", "false").lower() in ("1", "true", "yes")
+PYTHON_UPDATE_CASSETTE = os.getenv("PYTHON_UPDATE_CASSETTE", "false").lower() in (
+    "1",
+    "true",
+    "yes",
+)
+
 
 def func_path(function):
     path = Path(Path(inspect.getfile(function)).parent, "cassettes")
     Path.mkdir(path, exist_ok=True)
     filename = function.__name__ + ".cassette.yaml"
     return Path(path, filename)
 
+
 scw_vcr = vcr.VCR(
     record_mode="all" if PYTHON_UPDATE_CASSETTE else "none",
     filter_headers=["x-auth-token"],
-    func_path_generator=func_path
-)
+    func_path_generator=func_path,
+)