Merge pull request #577 from linode/test_vpc

Integration tests for VPC Dual Stack(TPT-3639)

Author: vshanthe

test/integration/conftest.py

@@ -402,9 +402,10 @@ def create_vpc(test_linode_client):
     label = get_test_label(length=10)
 
     vpc = client.vpcs.create(
-        label,
-        get_region(test_linode_client, {"VPCs"}),
+        label=label,
+        region=get_region(test_linode_client, {"VPCs"}),
         description="test description",
+        ipv6=[{"range": "auto"}],
     )
     yield vpc
 
@@ -413,7 +414,11 @@ def create_vpc(test_linode_client):
 
 @pytest.fixture(scope="session")
 def create_vpc_with_subnet(test_linode_client, create_vpc):
-    subnet = create_vpc.subnet_create("test-subnet", ipv4="10.0.0.0/24")
+    subnet = create_vpc.subnet_create(
+        label="test-subnet",
+        ipv4="10.0.0.0/24",
+        ipv6=[{"range": "auto"}],
+    )
 
     yield create_vpc, subnet
 

test/integration/linode_client/test_linode_client.py

@@ -6,11 +6,7 @@
 import pytest
 
 from linode_api4 import ApiError
-from linode_api4.objects import (
-    ConfigInterface,
-    ObjectStorageKeys,
-    Region,
-)
+from linode_api4.objects import ConfigInterface, ObjectStorageKeys, Region
 
 
 @pytest.fixture(scope="session")

test/integration/models/linode/interfaces/test_interfaces.py

@@ -14,6 +14,7 @@
     LinodeInterfacePublicIPv6RangeOptions,
     LinodeInterfacePublicOptions,
     LinodeInterfaceVLANOptions,
+    LinodeInterfaceVPCIPv4AddressOptions,
     LinodeInterfaceVPCIPv4Options,
     LinodeInterfaceVPCIPv4RangeOptions,
     LinodeInterfaceVPCOptions,
@@ -69,6 +70,13 @@ def __assert_vpc(iface: LinodeInterface):
 
         assert len(iface.vpc.ipv4.ranges) == 0
 
+        slaac_entry = iface.vpc.ipv6.slaac[0]
+        assert ipaddress.ip_address(
+            slaac_entry.address
+        ) in ipaddress.ip_network(slaac_entry.range)
+        assert not iface.vpc.ipv6.is_public
+        assert len(iface.vpc.ipv6.ranges) == 0
+
     def __assert_vlan(iface: LinodeInterface):
         __assert_base(iface)
 
@@ -145,19 +153,18 @@ def linode_interface_vpc(
         vpc=LinodeInterfaceVPCOptions(
             subnet_id=subnet.id,
             ipv4=LinodeInterfaceVPCIPv4Options(
-                # TODO (Enhanced Interfaces): Not currently working as expected
-                # addresses=[
-                #     LinodeInterfaceVPCIPv4AddressOptions(
-                #         address="auto",
-                #         primary=True,
-                #         nat_1_1_address="any",
-                #     )
-                # ],
+                addresses=[
+                    LinodeInterfaceVPCIPv4AddressOptions(
+                        address="auto",
+                        primary=True,
+                        nat_1_1_address=None,
+                    )
+                ],
                 ranges=[
                     LinodeInterfaceVPCIPv4RangeOptions(
                         range="/29",
                     )
-                ]
+                ],
             ),
         ),
     ), instance, vpc, subnet
@@ -256,7 +263,7 @@ def test_linode_interface_create_vpc(linode_interface_vpc):
     assert iface.version
 
     assert iface.default_route.ipv4
-    assert not iface.default_route.ipv6
+    assert iface.default_route.ipv6
 
     assert iface.vpc.vpc_id == vpc.id
     assert iface.vpc.subnet_id == subnet.id
@@ -267,6 +274,16 @@ def test_linode_interface_create_vpc(linode_interface_vpc):
 
     assert iface.vpc.ipv4.ranges[0].range.split("/")[1] == "29"
 
+    assert iface.default_route.ipv6
+    ipv6 = iface.vpc.ipv6
+    assert ipv6 and ipv6.is_public is False
+
+    if ipv6.slaac:
+        assert ipv6.ranges == [] and len(ipv6.slaac) == 1
+        assert ipv6.slaac[0].range and ipv6.slaac[0].address
+    elif ipv6.ranges:
+        assert ipv6.slaac == [] and len(ipv6.ranges) > 0
+
 
 def test_linode_interface_update_vpc(linode_interface_vpc):
     iface, instance, vpc, subnet = linode_interface_vpc

test/integration/models/linode/test_linode.py

@@ -1,3 +1,4 @@
+import ipaddress
 import time
 from test.integration.conftest import get_region
 from test.integration.helpers import (
@@ -651,7 +652,7 @@ def __assert_public(iface: LinodeInterface):
         __assert_base(iface)
 
         assert not iface.default_route.ipv4
-        assert iface.default_route.ipv6
+        assert not iface.default_route.ipv6
 
         assert len(iface.public.ipv4.addresses) == 0
         assert len(iface.public.ipv4.shared) == 0
@@ -666,7 +667,7 @@ def __assert_vpc(iface: LinodeInterface):
         __assert_base(iface)
 
         assert iface.default_route.ipv4
-        assert not iface.default_route.ipv6
+        assert iface.default_route.ipv6
 
         assert iface.vpc.vpc_id == vpc.id
         assert iface.vpc.subnet_id == subnet.id
@@ -679,6 +680,14 @@ def __assert_vpc(iface: LinodeInterface):
         assert len(iface.vpc.ipv4.ranges) == 1
         assert iface.vpc.ipv4.ranges[0].range == "10.0.0.5/32"
 
+        assert len(iface.vpc.ipv6.slaac) == 1
+
+        ipaddress.IPv6Network(iface.vpc.ipv6.slaac[0].range)
+        ipaddress.IPv6Address(iface.vpc.ipv6.slaac[0].address)
+
+        assert len(iface.vpc.ipv6.ranges) == 0
+        assert iface.vpc.ipv6.is_public is False
+
     def __assert_vlan(iface: LinodeInterface):
         __assert_base(iface)
 
@@ -888,9 +897,12 @@ def test_create_vpc(
         test_linode_client,
         linode_and_vpc_for_legacy_interface_tests_offline,
     ):
-        vpc, subnet, linode, _ = (
-            linode_and_vpc_for_legacy_interface_tests_offline
-        )
+        (
+            vpc,
+            subnet,
+            linode,
+            _,
+        ) = linode_and_vpc_for_legacy_interface_tests_offline
 
         config: Config = linode.configs[0]
 
@@ -927,11 +939,30 @@ def test_create_vpc(
         assert vpc_range_ip.address_range == "10.0.0.5/32"
         assert not vpc_range_ip.active
 
+        assert isinstance(vpc.ipv6, list)
+        assert len(vpc.ipv6) > 0
+        assert isinstance(vpc.ipv6[0].range, str)
+        assert ":" in vpc.ipv6[0].range
+
         # TODO:: Add `VPCIPAddress.filters.linode_id == linode.id` filter back
 
         # Attempt to resolve the IP from /vpcs/ips
         all_vpc_ips = test_linode_client.vpcs.ips()
-        assert all_vpc_ips[0].dict == vpc_ip.dict
+        matched_ip = next(
+            (
+                ip
+                for ip in all_vpc_ips
+                if ip.address == vpc_ip.address
+                and ip.vpc_id == vpc_ip.vpc_id
+                and ip.linode_id == vpc_ip.linode_id
+            ),
+            None,
+        )
+
+        assert (
+            matched_ip is not None
+        ), f"Expected VPC IP {vpc_ip.address} not found in /vpcs/ips"
+        assert matched_ip.dict == vpc_ip.dict
 
         # Test getting the ips under this specific VPC
         vpc_ips = vpc.ips
@@ -941,13 +972,50 @@ def test_create_vpc(
         assert vpc_ips[0].linode_id == linode.id
         assert vpc_ips[0].nat_1_1 == linode.ips.ipv4.public[0].address
 
+        # Validate VPC IPv6 IPs from /vpcs/ips
+        all_vpc_ipv6 = test_linode_client.get("/vpcs/ipv6s")["data"]
+
+        # Find matching VPC IPv6 entry
+        matched_ipv6 = next(
+            (
+                ip
+                for ip in all_vpc_ipv6
+                if ip["vpc_id"] == vpc.id
+                and ip["linode_id"] == linode.id
+                and ip["interface_id"] == interface.id
+                and ip["subnet_id"] == subnet.id
+            ),
+            None,
+        )
+
+        assert (
+            matched_ipv6
+        ), f"No VPC IPv6 found for Linode {linode.id} in VPC {vpc.id}"
+
+        assert matched_ipv6["ipv6_range"].count(":") >= 2
+        assert not matched_ipv6["ipv6_is_public"]
+
+        ipv6_addresses = matched_ipv6.get("ipv6_addresses", [])
+        assert (
+            isinstance(ipv6_addresses, list) and ipv6_addresses
+        ), "No IPv6 addresses found"
+
+        slaac = ipv6_addresses[0]
+        assert (
+            isinstance(slaac.get("slaac_address"), str)
+            and ":" in slaac["slaac_address"]
+        )
+
     def test_update_vpc(
         self,
         linode_and_vpc_for_legacy_interface_tests_offline,
     ):
-        vpc, subnet, linode, _ = (
-            linode_and_vpc_for_legacy_interface_tests_offline
-        )
+        (
+            vpc,
+            subnet,
+            linode,
+            _,
+        ) = linode_and_vpc_for_legacy_interface_tests_offline
 
         config: Config = linode.configs[0]
 

test/integration/models/vpc/test_vpc.py

@@ -10,6 +10,7 @@ def test_get_vpc(test_linode_client, create_vpc):
     vpc = test_linode_client.load(VPC, create_vpc.id)
     test_linode_client.vpcs()
     assert vpc.id == create_vpc.id
+    assert isinstance(vpc.ipv6[0].range, str)
 
 
 @pytest.mark.smoke
@@ -31,7 +32,11 @@ def test_update_vpc(test_linode_client, create_vpc):
 def test_get_subnet(test_linode_client, create_vpc_with_subnet):
     vpc, subnet = create_vpc_with_subnet
     loaded_subnet = test_linode_client.load(VPCSubnet, subnet.id, vpc.id)
-
+    assert loaded_subnet.ipv4 == subnet.ipv4
+    assert loaded_subnet.ipv6 is not None
+    assert loaded_subnet.ipv6[0].range.startswith(
+        vpc.ipv6[0].range.split("::")[0]
+    )
     assert loaded_subnet.id == subnet.id
 
 
@@ -88,7 +93,9 @@ def test_fails_create_subnet_invalid_data(create_vpc):
         create_vpc.subnet_create("test-subnet", ipv4=invalid_ipv4)
 
     assert excinfo.value.status == 400
-    assert "ipv4 must be an IPv4 network" in str(excinfo.value.json)
+    error_msg = str(excinfo.value.json)
+
+    assert "Must be an IPv4 network" in error_msg
 
 
 def test_fails_update_subnet_invalid_data(create_vpc_with_subnet):
@@ -101,3 +108,35 @@ def test_fails_update_subnet_invalid_data(create_vpc_with_subnet):
 
     assert excinfo.value.status == 400
     assert "Label must include only ASCII" in str(excinfo.value.json)
+
+
+def test_fails_create_subnet_with_invalid_ipv6_range(create_vpc):
+    valid_ipv4 = "10.0.0.0/24"
+    invalid_ipv6 = [{"range": "2600:3c11:e5b9::/5a"}]
+
+    with pytest.raises(ApiError) as excinfo:
+        create_vpc.subnet_create(
+            label="bad-ipv6-subnet",
+            ipv4=valid_ipv4,
+            ipv6=invalid_ipv6,
+        )
+
+    assert excinfo.value.status == 400
+    error = excinfo.value.json["errors"]
+
+    assert any(
+        e["field"] == "ipv6[0].range"
+        and "Must be an IPv6 network" in e["reason"]
+        for e in error
+    )
+
+
+def test_get_vpc_ipv6s(test_linode_client):
+    ipv6s = test_linode_client.get("/vpcs/ipv6s")["data"]
+
+    assert isinstance(ipv6s, list)
+
+    for ipv6 in ipv6s:
+        assert "vpc_id" in ipv6
+        assert isinstance(ipv6["ipv6_range"], str)
+        assert isinstance(ipv6["ipv6_addresses"], list)