diff --git a/tests/integration/vpc/conftest.py b/tests/integration/vpc/conftest.py index d4cbb033..67342240 100644 --- a/tests/integration/vpc/conftest.py +++ b/tests/integration/vpc/conftest.py @@ -11,7 +11,7 @@ @pytest.fixture -def test_vpc_w_subnet(): +def get_test_vpc_w_subnet(): vpc_json = create_vpc_w_subnet() vpc_id = str(vpc_json["id"]) @@ -21,9 +21,8 @@ def test_vpc_w_subnet(): @pytest.fixture -def test_vpc_wo_subnet(): +def get_test_vpc_wo_subnet(): region = get_random_region_with_caps(required_capabilities=["VPCs"]) - label = get_random_text(5) + "-label" vpc_id = exec_test_command( @@ -48,15 +47,43 @@ def test_vpc_wo_subnet(): @pytest.fixture -def test_subnet(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet - subnet_label = get_random_text(5) + "-label" - res = exec_test_command( +def get_test_vpc_w_rdma_type(): + # GPUDirect RDMA capability not available for now + # region = get_random_region_with_caps(required_capabilities=["VPCs", "GPUDirect RDMA"]) + region = get_random_region_with_caps(required_capabilities=["VPCs"]) + label = get_random_text(5) + "-test-rdma-vpc" + + vpc_id = exec_test_command( + BASE_CMDS["vpcs"] + + [ + "create", + "--label", + label, + "--region", + region, + "--vpc_type", + "rdma", + "--no-headers", + "--text", + "--format=id", + ] + ) + + yield vpc_id + + delete_target_id(target="vpcs", id=vpc_id) + + +@pytest.fixture +def get_test_subnet(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet + label = get_random_text(5) + "-label" + subnet_id = exec_test_command( BASE_CMDS["vpcs"] + [ "subnet-create", "--label", - subnet_label, + label, "--ipv4", "10.0.0.0/24", vpc_id, @@ -64,9 +91,31 @@ def test_subnet(test_vpc_wo_subnet): "--no-headers", "--delimiter=,", ] - ) + ).split(",")[0] + + yield vpc_id, subnet_id + + +@pytest.fixture +def get_test_subnet_w_rdma_type(get_test_vpc_w_rdma_type): + vpc_id = get_test_vpc_w_rdma_type + label = get_random_text(5) + "-test-rdma-subnet" + subnet_id = exec_test_command( + BASE_CMDS["vpcs"] + + [ + "subnet-create", + "--label", + label, + "--ipv4", + "10.0.0.0/24", + vpc_id, + "--text", + "--no-headers", + "--delimiter=,", + ] + ).split(",")[0] - yield res, subnet_label + yield vpc_id, subnet_id @pytest.fixture diff --git a/tests/integration/vpc/test_vpc.py b/tests/integration/vpc/test_vpc.py index 6616aa9f..9f594898 100644 --- a/tests/integration/vpc/test_vpc.py +++ b/tests/integration/vpc/test_vpc.py @@ -12,36 +12,60 @@ get_random_text, ) -BASE_CMD = ["linode-cli", "vpcs"] +HEADERS_VPC = ["id", "label", "description", "region", "vpc_type"] +HEADERS_SUBNET = ["id", "label", "ipv4", "vpc_type"] # TODO: Remove this variable and @pytest.mark.skipif once VPC Dual Stack is ready to ship disable_vpc_dual_stack_tests = True -def test_list_vpcs(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet - res = exec_test_command(BASE_CMDS["vpcs"] + ["ls", "--text"]) - headers = ["id", "label", "description", "region"] +def get_vpcs_list(params: str = None): + params = params.split() if params else [] + command = BASE_CMDS["vpcs"] + ["ls", "--text"] + params + return exec_test_command(command) - for header in headers: - assert header in res - assert vpc_id in res +def get_vpc_view(vpc_id: int = None): + return json.loads( + exec_test_command(BASE_CMDS["vpcs"] + ["view", vpc_id, "--json"]) + )[0] -def test_view_vpc(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet - res = exec_test_command( - BASE_CMDS["vpcs"] + ["view", vpc_id, "--text", "--no-headers"] - ) +def get_subnets_list(vpc_id: int, params: str = None): + params = params.split() if params else [] + command = BASE_CMDS["vpcs"] + ["subnets-list", vpc_id] + params + return exec_test_command(command) + + +def get_subnet_view(vpc_id: int, subnet_id: int): + return json.loads( + exec_test_command( + BASE_CMDS["vpcs"] + ["subnet-view", vpc_id, subnet_id, "--json"] + ) + )[0] + + +def test_list_vpcs(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet + output = get_vpcs_list("--page-size 100") + + assert all(header in output for header in HEADERS_VPC) + assert vpc_id in output + + +def test_view_vpc(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet + output = get_vpc_view(vpc_id) - assert vpc_id in res + assert all([header in output.keys() for header in HEADERS_VPC]) + assert str(output["id"]) == vpc_id + assert output["vpc_type"] == "regular" @pytest.mark.smoke -def test_update_vpc(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet +def test_update_vpc(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet new_label = get_random_text(5) + "label" @@ -61,7 +85,7 @@ def test_update_vpc(test_vpc_wo_subnet): ) description = exec_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + ["view", vpc_id, "--text", "--no-headers", "--format=description"] ) @@ -69,49 +93,46 @@ def test_update_vpc(test_vpc_wo_subnet): assert "new description" in description -def test_list_subnets(test_vpc_w_subnet): - vpc_id = test_vpc_w_subnet +def test_vpc_with_rdma_type(get_test_vpc_w_rdma_type): + vpc_id = get_test_vpc_w_rdma_type - res = exec_test_command( - BASE_CMD + ["subnets-list", vpc_id, "--text", "--delimiter=,"] - ) + output = get_vpcs_list("--page-size 100") + assert all(header in output for header in HEADERS_VPC) + assert vpc_id in output - lines = res.splitlines() + output = get_vpcs_list("--page-size 100 --format=vpc_type --no-headers") + assert any(["rdma" in output.split()]) - headers = ["id", "label", "ipv4"] + output = get_vpc_view(vpc_id) + assert str(output["id"]) == vpc_id + assert output["vpc_type"] == "rdma" - for header in headers: - assert header in lines[0] - for line in lines[1:]: - assert re.match( - r"^(\d+),(\w+),(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}/\d+)$", line - ), "String format does not match" +def test_list_subnets(get_test_vpc_w_subnet): + vpc_id = get_test_vpc_w_subnet + output = get_subnets_list(vpc_id, "--text --delimiter=,").splitlines() + assert all(header in output[0] for header in HEADERS_SUBNET) -def test_view_subnet(test_vpc_wo_subnet, test_subnet): - # note calling test_subnet fixture will add subnet to test_vpc_wo_subnet - res, label = test_subnet + for line in output[1:]: + assert re.match( + r"^(\d+),(\w+),(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}/\d+),(\w+)$", + line, + ), "String format does not match" - res = res.split(",") - vpc_subnet_id = res[0] +def test_view_subnet(get_test_subnet): + vpc_id, subnet_id = get_test_subnet + output = get_subnet_view(vpc_id, subnet_id) - vpc_id = test_vpc_wo_subnet - - output = exec_test_command( - BASE_CMDS["vpcs"] + ["subnet-view", vpc_id, vpc_subnet_id, "--text"] - ) - - headers = ["id", "label", "ipv4"] - for header in headers: - assert header in output - assert vpc_subnet_id in output + assert all(header in output.keys() for header in HEADERS_SUBNET) + assert str(output["id"]) == subnet_id + assert output["vpc_type"] == "regular" @pytest.mark.smoke -def test_update_subnet(test_vpc_w_subnet): - vpc_id = test_vpc_w_subnet +def test_update_subnet(get_test_vpc_w_subnet): + vpc_id = get_test_vpc_w_subnet new_label = get_random_text(5) + "label" @@ -121,7 +142,7 @@ def test_update_subnet(test_vpc_w_subnet): ) updated_label = exec_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + [ "subnet-update", vpc_id, @@ -137,12 +158,29 @@ def test_update_subnet(test_vpc_w_subnet): assert new_label == updated_label +def test_subnet_with_rdma_type_vpc(get_test_subnet_w_rdma_type): + vpc_id, subnet_id = get_test_subnet_w_rdma_type + + output = get_subnets_list(vpc_id) + assert all(header in output for header in HEADERS_SUBNET) + assert subnet_id in output + + output = get_subnets_list(vpc_id, "--format=vpc_type --no-headers") + assert any(["rdma" in output.split()]) + + output = get_subnet_view(vpc_id, subnet_id) + assert str(output["id"]) == subnet_id + assert output["vpc_type"] == "rdma" + + +@pytest.mark.skip(reason="Defect: ARB-8019") def test_fails_to_create_vpc_invalid_label(): invalid_label = "invalid_label" region = get_random_region_with_caps(required_capabilities=["VPCs"]) res = exec_failing_test_command( - BASE_CMD + ["create", "--label", invalid_label, "--region", region], + BASE_CMDS["vpcs"] + + ["create", "--label", invalid_label, "--region", region], ExitCodes.REQUEST_FAILED, ) @@ -150,27 +188,29 @@ def test_fails_to_create_vpc_invalid_label(): assert "Must only use ASCII letters, numbers, and dashes" in res -def test_fails_to_create_vpc_duplicate_label(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet +def test_fails_to_create_vpc_duplicate_label(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet label = exec_test_command( - BASE_CMD + ["view", vpc_id, "--text", "--no-headers", "--format=label"] + BASE_CMDS["vpcs"] + + ["view", vpc_id, "--text", "--no-headers", "--format=label"] ) region = get_random_region_with_caps(required_capabilities=["VPCs"]) res = exec_failing_test_command( - BASE_CMD + ["create", "--label", label, "--region", region], + BASE_CMDS["vpcs"] + ["create", "--label", label, "--region", region], ExitCodes.REQUEST_FAILED, ) assert "Label must be unique among your VPCs" in res -def test_fails_to_update_vpc_invalid_label(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet +@pytest.mark.skip(reason="Defect: ARB-8019") +def test_fails_to_update_vpc_invalid_label(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet invalid_label = "invalid_label" res = exec_failing_test_command( - BASE_CMD + ["update", vpc_id, "--label", invalid_label], + BASE_CMDS["vpcs"] + ["update", vpc_id, "--label", invalid_label], ExitCodes.REQUEST_FAILED, ) @@ -178,12 +218,13 @@ def test_fails_to_update_vpc_invalid_label(test_vpc_wo_subnet): assert "Must only use ASCII letters, numbers, and dashes" in res -def test_fails_to_create_vpc_subnet_w_invalid_label(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet +@pytest.mark.skip(reason="Defect: ARB-8019") +def test_fails_to_create_vpc_subnet_w_invalid_label(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet invalid_label = "invalid_label" res = exec_failing_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + [ "subnet-create", "--label", @@ -199,18 +240,19 @@ def test_fails_to_create_vpc_subnet_w_invalid_label(test_vpc_wo_subnet): assert "Must only use ASCII letters, numbers, and dashes" in res -def test_fails_to_update_vpc_subnet_w_invalid_label(test_vpc_w_subnet): - vpc_id = test_vpc_w_subnet +@pytest.mark.skip(reason="Defect: ARB-8019") +def test_fails_to_update_vpc_subnet_w_invalid_label(get_test_vpc_w_subnet): + vpc_id = get_test_vpc_w_subnet invalid_label = "invalid_label" subnet_id = exec_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + ["subnets-list", vpc_id, "--text", "--format=id", "--no-headers"] ) res = exec_failing_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + [ "subnet-update", vpc_id, @@ -236,7 +278,7 @@ def test_create_vpc_with_ipv6_auto(): label = get_random_text(5) + "-vpc" res = exec_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + [ "create", "--label", @@ -269,7 +311,7 @@ def test_create_vpc_with_custom_ipv6_prefix_length(prefix_len): label = get_random_text(5) + f"-vpc{prefix_len}" res = exec_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + [ "create", "--label", @@ -294,12 +336,12 @@ def test_create_vpc_with_custom_ipv6_prefix_length(prefix_len): @pytest.mark.skipif( disable_vpc_dual_stack_tests, reason="Dual-stack tests disabled" ) -def test_create_subnet_with_ipv6_auto(test_vpc_wo_subnet): - vpc_id = test_vpc_wo_subnet +def test_create_subnet_with_ipv6_auto(get_test_vpc_wo_subnet): + vpc_id = get_test_vpc_wo_subnet subnet_label = get_random_text(5) + "-ipv6subnet" res = exec_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + [ "subnet-create", "--label", @@ -338,7 +380,7 @@ def test_fails_to_create_vpc_with_invalid_ipv6_range(): label = get_random_text(5) + "-invalidvpc" res = exec_failing_test_command( - BASE_CMD + BASE_CMDS["vpcs"] + [ "create", "--label", @@ -357,7 +399,7 @@ def test_fails_to_create_vpc_with_invalid_ipv6_range(): def test_list_vpc_ip_address(): res = exec_test_command( - BASE_CMD + ["ips-all-list", "--text", "--delimiter=,"] + BASE_CMDS["vpcs"] + ["ips-all-list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -374,7 +416,7 @@ def test_list_vpc_ip_address(): def test_list_vpc_ipv6s_address(): res = exec_test_command( - BASE_CMD + ["ipv6s-all-list", "--text", "--delimiter=,"] + BASE_CMDS["vpcs"] + ["ipv6s-all-list", "--text", "--delimiter=,"] ) lines = res.splitlines()