Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions test/integration/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -479,6 +479,7 @@ def create_vpc(test_linode_client):
"VPC IPv6 Stack",
"Linode Interfaces",
"Custom VPC IPv4 Ranges",
"Linodes",
},
),
description="test description",
Expand Down
91 changes: 54 additions & 37 deletions test/integration/models/linode/test_linode.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ def linode_and_vpc_for_legacy_interface_tests_offline(
label = get_test_label(length=8)

instance = test_linode_client.linode.instance_create(
"g6-standard-1",
"g5-standard-1",
Comment thread
zliang-akamai marked this conversation as resolved.
vpc.region,
booted=False,
image="linode/debian11",
Expand Down Expand Up @@ -134,15 +134,21 @@ def linode_for_vpu_tests(test_linode_client, e2e_test_firewall):
pytest.skip("No VPU capacity is currently available")

label = get_test_label(length=8)

linode_instance = client.linode.instance_create(
vpu_type,
region,
image="linode/debian12",
label=label,
firewall=e2e_test_firewall,
root_pass="aComplex@Password123",
)
try:
linode_instance = client.linode.instance_create(
vpu_type,
region,
image="linode/debian12",
label=label,
firewall=e2e_test_firewall,
root_pass="aComplex@Password123",
)
except ApiError as e:
reasons = e.errors or [str(e)]
unavailable_msg = "not currently available in the selected region"
if e.status == 400 and any(unavailable_msg in r for r in reasons):
pytest.skip("No VPU capacity is currently available")
raise

yield linode_instance

Expand Down Expand Up @@ -1019,18 +1025,24 @@ def test_create_vpc(

# 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()
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,
)
# Attempt to resolve the IP from /vpcs/ips. The account-wide listing
# may lag behind instance creation, so poll until the IP appears.
def resolve_vpc_ip():
return next(
(
ip
for ip in test_linode_client.vpcs.ips()
if ip.address == vpc_ip.address
and ip.vpc_id == vpc_ip.vpc_id
and ip.linode_id == vpc_ip.linode_id
),
None,
)

try:
matched_ip = wait_for_condition(5, 120, resolve_vpc_ip)
except TimeoutError:
matched_ip = None

assert (
matched_ip is not None
Expand All @@ -1045,21 +1057,26 @@ 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,
)
# Validate VPC IPv6 IPs from /vpcs/ipv6s. The account-wide listing may
# lag behind instance creation, so poll until the entry appears.
def resolve_vpc_ipv6():
all_vpc_ipv6 = test_linode_client.get("/vpcs/ipv6s")["data"]
return 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,
)

try:
matched_ipv6 = wait_for_condition(5, 120, resolve_vpc_ipv6)
except TimeoutError:
matched_ipv6 = None

assert (
matched_ipv6
Expand Down