diff --git a/test/integration/conftest.py b/test/integration/conftest.py index a5c832f4f..8583e5b9c 100644 --- a/test/integration/conftest.py +++ b/test/integration/conftest.py @@ -26,6 +26,7 @@ PlacementGroupPolicy, PlacementGroupType, PostgreSQLDatabase, + ReservedIPAddress, ) from linode_api4.errors import ApiError from linode_api4.linode_client import LinodeClient, MonitorClient @@ -728,3 +729,47 @@ def test_monitor_client(get_monitor_token_for_db_entities): ) return client, entity_ids + + +@pytest.fixture +def create_reserved_ip(test_linode_client): + client = test_linode_client + region = get_region(client, {"Linodes", "Cloud Firewall"}, site_type="core") + reserved_ip = client.networking.reserved_ip_create( + region=region, tags=["test"] + ) + + yield reserved_ip + + # Delete only if IP exists (some tests delete it earlier) + if client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + ): + reserved_ip.delete() + + +@pytest.fixture +def create_reserved_ip_assigned(test_linode_client, create_linode): + client = test_linode_client + linode = create_linode + reserved_ip = client.networking.reserved_ip_create( + region=linode.region, + tags=["test", "assigned"], + ) + + client.networking.ip_addresses_assign( + assignments=[{"address": reserved_ip.address, "linode_id": linode.id}], + region=linode.region, + ) + + reserved_ip = test_linode_client.load( + ReservedIPAddress, reserved_ip.address + ) + + yield linode, reserved_ip + + # Delete only if IP exists (some tests delete it earlier) + if client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + ): + reserved_ip.delete() diff --git a/test/integration/models/linode/interfaces/test_interfaces.py b/test/integration/models/linode/interfaces/test_interfaces.py index 650a9cb6c..cc55dfb54 100644 --- a/test/integration/models/linode/interfaces/test_interfaces.py +++ b/test/integration/models/linode/interfaces/test_interfaces.py @@ -1,13 +1,16 @@ import copy import ipaddress +from test.integration.helpers import get_test_label import pytest from linode_api4 import ( ApiError, Instance, + InterfaceGeneration, LinodeInterface, LinodeInterfaceDefaultRouteOptions, + LinodeInterfaceOptions, LinodeInterfacePublicIPv4AddressOptions, LinodeInterfacePublicIPv4Options, LinodeInterfacePublicIPv6Options, @@ -18,9 +21,28 @@ LinodeInterfaceVPCIPv4Options, LinodeInterfaceVPCIPv4RangeOptions, LinodeInterfaceVPCOptions, + ReservedIPAddress, ) +def build_interface_public_ipv4(firewall, ip_address): + return LinodeInterfaceOptions( + firewall_id=firewall, + default_route=LinodeInterfaceDefaultRouteOptions( + ipv4=True, + ), + public=LinodeInterfacePublicOptions( + ipv4=LinodeInterfacePublicIPv4Options( + addresses=[ + LinodeInterfacePublicIPv4AddressOptions( + address=ip_address, primary=True + ) + ], + ), + ), + ) + + def test_linode_create_with_linode_interfaces( create_vpc_with_subnet, linode_with_linode_interfaces, @@ -359,3 +381,59 @@ def test_linode_interface_firewalls(e2e_test_firewall, linode_interface_public): firewall = firewalls[0] assert firewall.id == e2e_test_firewall.id assert firewall.label == e2e_test_firewall.label + + +@pytest.mark.parametrize( + "iface_type", + [InterfaceGeneration.LEGACY_CONFIG, InterfaceGeneration.LINODE], +) +def test_linode_interfaces_with_reserved_ips( + test_linode_client, e2e_test_firewall, create_reserved_ip, iface_type +): + client = test_linode_client + reserved_ip = create_reserved_ip + label = get_test_label(length=8) + + if iface_type == InterfaceGeneration.LEGACY_CONFIG: + linode, _ = client.linode.instance_create( + "g6-nanode-1", + reserved_ip.region, + image="linode/debian12", + label=label, + firewall=e2e_test_firewall, + interface_generation=iface_type, + ipv4=[reserved_ip.address], + ) + else: + interface = build_interface_public_ipv4( + e2e_test_firewall.id, reserved_ip.address + ) + linode, _ = client.linode.instance_create( + "g6-nanode-1", + reserved_ip.region, + image="linode/debian12", + label=label, + interface_generation=iface_type, + interfaces=[interface], + ) + + linode_ips = linode.ips.ipv4.public + assert len(linode_ips) == 1 + assert linode_ips[0].address == reserved_ip.address + assert linode_ips[0].reserved == True + assert linode_ips[0].linode_id == linode.id + assert linode_ips[0].assigned_entity.id == linode.id + assert linode_ips[0].assigned_entity.type == "linode" + assert linode_ips[0].assigned_entity.label == linode.label + assert ( + linode_ips[0].assigned_entity.url == f"/v4/linode/instances/{linode.id}" + ) + + linode.delete() + reserved_ips_list = client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + ) + assert len(reserved_ips_list) == 1 + assert reserved_ips_list[0].reserved == True + assert reserved_ips_list[0].linode_id is None + assert reserved_ips_list[0].assigned_entity is None diff --git a/test/integration/models/linode/test_linode.py b/test/integration/models/linode/test_linode.py index 9f6194fa9..fadedeced 100644 --- a/test/integration/models/linode/test_linode.py +++ b/test/integration/models/linode/test_linode.py @@ -19,6 +19,7 @@ Instance, InterfaceGeneration, LinodeInterface, + ReservedIPAddress, Type, ) from linode_api4.objects.linode import InstanceDiskEncryptionType, MigrationType @@ -1147,3 +1148,42 @@ def test_update_linode_maintenance_policy(create_linode, test_linode_client): linode.invalidate() assert result assert linode.maintenance_policy_id == non_default_policy.slug + + +def test_update_linode_with_reserved_ip_in_address( + test_linode_client, e2e_test_firewall, create_reserved_ip +): + label = get_test_label(length=8) + client = test_linode_client + reserved_ip = create_reserved_ip + + linode, _ = client.linode.instance_create( + "g6-nanode-1", + reserved_ip.region, + image="linode/debian12", + label=label, + firewall=e2e_test_firewall, + ) + + linode_ips = linode.ips.ipv4.public + assert len(linode_ips) == 1 + assert linode_ips[0].address != reserved_ip.address + + linode.ip_allocate(True, reserved_ip.address) + delattr(linode, "_ips") + linode_ips = linode.ips.ipv4.public + assert len(linode_ips) == 2 + assert reserved_ip.address in [ip.address for ip in linode_ips] + + reserved_ip = client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + )[0] + assert reserved_ip.linode_id == linode.id + assert reserved_ip.assigned_entity.id == linode.id + assert reserved_ip.assigned_entity.type == "linode" + assert reserved_ip.assigned_entity.label == linode.label + assert ( + reserved_ip.assigned_entity.url == f"/v4/linode/instances/{linode.id}" + ) + + linode.delete() diff --git a/test/integration/models/networking/test_networking.py b/test/integration/models/networking/test_networking.py index 27ffbb444..d088d7dea 100644 --- a/test/integration/models/networking/test_networking.py +++ b/test/integration/models/networking/test_networking.py @@ -1,3 +1,4 @@ +import ipaddress import time from test.integration.conftest import ( get_api_ca_file, @@ -12,9 +13,20 @@ ) import pytest +import requests -from linode_api4 import Instance, LinodeClient -from linode_api4.objects import Config, ConfigInterfaceIPv4, Firewall, IPAddress +from linode_api4 import ( + ApiError, + Instance, + LinodeClient, +) +from linode_api4.objects import ( + Config, + ConfigInterfaceIPv4, + Firewall, + IPAddress, + ReservedIPAddress, +) from linode_api4.objects.networking import ( FirewallCreateDevicesOptions, NetworkTransferPrice, @@ -351,3 +363,190 @@ def test_ip_info(test_linode_client, create_linode): assert ip_info.subnet_mask is not None assert ip_info.type == "ipv4" assert ip_info.vpc_nat_1_1 is None + + +def verify_reserved_ip(reserved_ip): + assert isinstance( + ipaddress.ip_address(reserved_ip.address), ipaddress.IPv4Address + ) + assert reserved_ip.type == "ipv4" + assert reserved_ip.public == True + assert reserved_ip.reserved == True + assert reserved_ip.linode_id is None + assert reserved_ip.assigned_entity is None + + +def verify_reserved_ip_assigned(reserved_ip, resource): + assert isinstance( + ipaddress.ip_address(reserved_ip.address), ipaddress.IPv4Address + ) + assert reserved_ip.type == "ipv4" + assert reserved_ip.public == True + assert reserved_ip.reserved == True + assert reserved_ip.linode_id == resource.id + assert reserved_ip.region.id == resource.region.id + assert reserved_ip.assigned_entity.id == resource.id + assert reserved_ip.assigned_entity.type == "linode" + assert reserved_ip.assigned_entity.label == resource.label + assert ( + reserved_ip.assigned_entity.url == f"/v4/linode/instances/{resource.id}" + ) + + +@pytest.mark.smoke +@pytest.mark.parametrize( + "region, tags", + [ + (TEST_REGION, ["test"]), + (TEST_REGION, None), + ], +) +def test_create_reserved_ip(request, test_linode_client, region, tags): + client = test_linode_client + reserved_ip = client.networking.reserved_ip_create(region=region, tags=tags) + request.addfinalizer(reserved_ip.delete) + + verify_reserved_ip(reserved_ip) + assert reserved_ip.tags == tags if tags else reserved_ip.tags == [] + + +def test_create_reserved_ip_wo_region_fail(test_linode_client): + client = test_linode_client + + with pytest.raises(ApiError) as exc_info: + client.networking.reserved_ip_create(region=None, tags=["test"]) + + error_msg = str(exc_info.value.json) + assert exc_info.value.status == 400 + assert "region is required" in error_msg + + +def test_update_reserved_ip_tags(test_linode_client, create_reserved_ip): + client = test_linode_client + reserved_ip = create_reserved_ip + verify_reserved_ip(reserved_ip) + assert reserved_ip.tags == ["test"] + + reserved_ip.tags = ["updated"] + reserved_ip.save() + reserved_ip = client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + )[0] + verify_reserved_ip(reserved_ip) + assert reserved_ip.tags == ["updated"] + + +def test_create_reserved_ip_assigned( + test_linode_client, create_reserved_ip_assigned +): + client = test_linode_client + linode, reserved_ip = create_reserved_ip_assigned + verify_reserved_ip_assigned(reserved_ip, linode) + assert sorted(reserved_ip.tags) == ["assigned", "test"] + + # ips_list = client.networking.ips() + # assert reserved_ip.address in [ip.address for ip in ips_list] + + reserved_ips_list = client.networking.reserved_ips() + assert reserved_ip.address in [ip.address for ip in reserved_ips_list] + + linode_ips = linode.ips.ipv4.public + assert len(linode_ips) == 2 + assert any([ip.reserved for ip in linode_ips]) + + reserved_ip.delete() + reserved_ips_list = client.networking.reserved_ips() + assert reserved_ip.address not in [ip.address for ip in reserved_ips_list] + + reserved_ips_list = client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + ) + assert len(reserved_ips_list) == 0 + + delattr(linode, "_ips") + linode_ips = linode.ips.ipv4.public + assert len(linode_ips) == 2 + assert not any([ip.reserved for ip in linode_ips]) + assert not any([ip.tags for ip in linode_ips]) # Tags should be removed + + +def test_get_reserved_ip_types(test_linode_client, create_reserved_ip): + client = test_linode_client + endpoint = client.base_url + "/networking/reserved/ips/types" + types = requests.get(endpoint).json()[ + "data" + ] # Pricing should be publicly available + + assert isinstance(types, list) + assert types[0]["id"] == "reserved-ipv4" + assert types[0]["label"] == "Reserved IPv4" + assert "hourly" in types[0]["price"] + assert "monthly" in types[0]["price"] + assert any(price != 0 for price in list(types[0]["price"].values())) + assert isinstance(types[0]["region_prices"], list) + + +@pytest.mark.smoke +@pytest.mark.parametrize( + "reserved, region", + [ + (True, TEST_REGION), + (True, None), + ], +) +def test_create_reserved_ip_with_allocate( + test_linode_client, create_linode, reserved, region +): + client = test_linode_client + linode = create_linode + + if region: + reserved_ip = client.networking.ip_allocate( + reserved=reserved, region=TEST_REGION + ) + verify_reserved_ip(reserved_ip) + else: + reserved_ip = client.networking.ip_allocate( + reserved=reserved, linode=linode.id + ) + verify_reserved_ip_assigned(reserved_ip, linode) + + assert reserved_ip.tags == [] + + +def test_reserve_ephemeral_ip(test_linode_client, create_linode): + client = test_linode_client + linode = create_linode + + ip_address = client.load(IPAddress, linode.ipv4[0]) + assert ip_address.linode_id == linode.id + assert ip_address.reserved == False + + ip_address.reserved = True + ip_address.save() + ip_address = client.load(IPAddress, linode.ipv4[0]) + assert ip_address.linode_id == linode.id + assert ip_address.reserved == True + + ip_address.reserved = False + ip_address.save() + ip_address = client.load(IPAddress, linode.ipv4[0]) + assert ip_address.linode_id == linode.id + assert ip_address.reserved == False + + +def test_convert_unassigned_reserved_ip_to_ephemeral( + test_linode_client, create_reserved_ip +): + client = test_linode_client + reserved_ip = create_reserved_ip + verify_reserved_ip(reserved_ip) + + ip_address = client.load(IPAddress, reserved_ip.address) + ip_address.reserved = False + ip_address.save() + + reserved_ips_list = client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + ) + assert len(reserved_ips_list) == 0 diff --git a/test/integration/models/nodebalancer/test_nodebalancer.py b/test/integration/models/nodebalancer/test_nodebalancer.py index 692efb027..522dad3d2 100644 --- a/test/integration/models/nodebalancer/test_nodebalancer.py +++ b/test/integration/models/nodebalancer/test_nodebalancer.py @@ -15,6 +15,7 @@ NodeBalancerNode, NodeBalancerType, RegionPrice, + ReservedIPAddress, ) TEST_REGION = get_region( @@ -113,6 +114,34 @@ def test_create_nb(test_linode_client, e2e_test_firewall): nb.delete() +def test_create_nb_with_reserved_ip( + test_linode_client, e2e_test_firewall, create_reserved_ip +): + client = test_linode_client + reserved_ip = create_reserved_ip + label = get_test_label(8) + + nb = client.nodebalancer_create( + region=TEST_REGION, + label=label, + firewall=e2e_test_firewall.id, + client_udp_sess_throttle=5, + ipv4=reserved_ip.address, + ) + + assert TEST_REGION, nb.region + assert label == nb.label + assert nb.ipv4.address == reserved_ip.address + assert nb.ipv4.public == True + assert nb.ipv4.reserved == True + + nb.delete() + reserved_ip = client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + )[0] + assert reserved_ip.assigned_entity is None + + def test_get_nodebalancer_config(test_linode_client, create_nb_config): config = test_linode_client.load( NodeBalancerConfig, diff --git a/test/integration/models/tag/test_tag.py b/test/integration/models/tag/test_tag.py index d2edf84c5..51ba8a1b7 100644 --- a/test/integration/models/tag/test_tag.py +++ b/test/integration/models/tag/test_tag.py @@ -2,7 +2,7 @@ import pytest -from linode_api4.objects import Tag +from linode_api4.objects import ReservedIPAddress, Tag @pytest.fixture @@ -15,8 +15,43 @@ def test_tag(test_linode_client): tag.delete() +@pytest.fixture +def create_tag_with_reserved_ip(test_linode_client, create_reserved_ip): + unique_tag = get_test_label() + "_tag" + reserved_ip = create_reserved_ip + + tag = test_linode_client.tags.create( + unique_tag, reserved_ipv4_addresses=[reserved_ip.address] + ) + reserved_ip = test_linode_client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + )[0] + + yield tag, reserved_ip + + tag.delete() + + @pytest.mark.smoke def test_get_tag(test_linode_client, test_tag): tag = test_linode_client.load(Tag, test_tag.id) assert tag.id == test_tag.id + + +def test_get_tag_with_reserved_ip( + test_linode_client, create_tag_with_reserved_ip +): + tag, reserved_ip = create_tag_with_reserved_ip + tag = test_linode_client.load(Tag, tag.id).objects[0] + + assert vars(tag).keys() == vars(reserved_ip).keys() + assert tag.address == reserved_ip.address + assert tag.reserved == reserved_ip.reserved + assert tag.tags == reserved_ip.tags + + tag.delete() + reserved_ip = test_linode_client.networking.reserved_ips( + ReservedIPAddress.address == reserved_ip.address + ) + assert len(reserved_ip) == 0