test_vpc.py

from test.integration.conftest import get_region
from test.integration.helpers import get_test_label

import pytest

from linode_api4 import VPC, ApiError, VPCIPv4DefaultRange, VPCSubnet


@pytest.mark.smoke
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
def test_update_vpc(test_linode_client, create_vpc):
    vpc = create_vpc
    new_label = create_vpc.label + "-updated"
    new_desc = "updated description"

    vpc.label = new_label
    vpc.description = new_desc
    vpc.save()

    vpc = test_linode_client.load(VPC, create_vpc.id)

    assert vpc.label == new_label
    assert vpc.description == new_desc


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


@pytest.mark.smoke
def test_update_subnet(test_linode_client, create_vpc_with_subnet):
    vpc, subnet = create_vpc_with_subnet
    new_label = subnet.label + "-updated"

    subnet.label = new_label
    subnet.save()

    subnet = test_linode_client.load(VPCSubnet, subnet.id, vpc.id)

    assert subnet.label == new_label


def test_fails_create_vpc_invalid_data(test_linode_client):
    with pytest.raises(ApiError) as excinfo:
        test_linode_client.vpcs.create(
            label="invalid_label!!",
            region=get_region(test_linode_client, {"VPCs"}),
            description="test description",
        )
    assert excinfo.value.status == 400


def test_get_all_vpcs(test_linode_client, create_multiple_vpcs):
    vpc_1, vpc_2 = create_multiple_vpcs

    all_vpcs = test_linode_client.vpcs()

    assert str(vpc_1) in str(all_vpcs.lists)
    assert str(vpc_2) in str(all_vpcs.lists)


def test_fails_update_vpc_invalid_data(create_vpc):
    vpc = create_vpc

    invalid_label = "invalid!!"
    vpc.label = invalid_label

    with pytest.raises(ApiError) as excinfo:
        vpc.save()

    assert excinfo.value.status == 400


def test_fails_create_subnet_invalid_data(create_vpc):
    invalid_ipv4 = "10.0.0.0"

    with pytest.raises(ApiError) as excinfo:
        create_vpc.subnet_create("test-subnet", ipv4=invalid_ipv4)

    assert excinfo.value.status == 400
    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):
    invalid_label = "invalid_subnet_label!!"
    vpc, subnet = create_vpc_with_subnet
    subnet.label = invalid_label

    with pytest.raises(ApiError) as excinfo:
        subnet.save()

    assert excinfo.value.status == 400
    assert "Must only use 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)


def test_get_vpc_default_ranges(test_linode_client):
    """
    Tests that VPC default IPv4 ranges can be retrieved.
    """
    result = test_linode_client.vpcs.default_ranges()

    assert isinstance(result, VPCIPv4DefaultRange)
    assert isinstance(result.ipv4_ranges, list)
    assert len(result.ipv4_ranges) > 0
    assert isinstance(result.forbidden_ipv4_ranges, list)
    assert len(result.forbidden_ipv4_ranges) > 0


def test_vpc_with_ipv4(test_linode_client):
    """
    Tests creating a VPC with ipv4 ranges, getting/listing it,
    updating the ipv4, and deleting it.
    """
    client = test_linode_client
    label = get_test_label(length=10)
    region = get_region(client, {"VPCs", "Custom VPC IPv4 Ranges"})

    # Create
    vpc = client.vpcs.create(
        label=label,
        region=region,
        description="integration test vpc with ipv4",
        ipv4=[{"range": "10.0.0.0/8"}],
    )

    try:
        assert vpc.id is not None
        assert vpc.label == label
        assert len(vpc.ipv4) > 0
        assert vpc.ipv4[0].range == "10.0.0.0/8"

        # Get by ID
        loaded_vpc = client.load(VPC, vpc.id)
        assert loaded_vpc.id == vpc.id
        assert loaded_vpc.ipv4[0].range == "10.0.0.0/8"

        # List and verify present
        all_vpcs = client.vpcs()
        vpc_ids = [v.id for v in all_vpcs]
        assert vpc.id in vpc_ids

        # Update ipv4
        vpc.ipv4 = [{"range": "192.168.0.0/17"}]
        vpc.save()

        updated_vpc = client.load(VPC, vpc.id)
        assert updated_vpc.ipv4[0].range == "192.168.0.0/17"
    finally:
        # Delete
        vpc.delete()