add tests to close issues

Author: Laure-di

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

@@ -3,6 +3,7 @@
 
 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
@@ -18,6 +19,8 @@
 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]:
@@ -81,7 +84,7 @@ def test_attach_additionnal_volume(instance_block_api: tuple[InstanceUtilsV1API,
 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
@@ -99,10 +102,15 @@ 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,50 @@
+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