TPS-4285: Implement integration tests for ReservedIP for IPv4

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+testpaths = test
+markers =
+    smoke: mark a test as a smoke test
+    flaky: mark a test as a flaky test for rerun
+python_files = *_test.py test_*.py

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,42 @@ 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=["test1"]
+    )
+
+    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()

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

@@ -8,6 +8,8 @@
     Instance,
     LinodeInterface,
     LinodeInterfaceDefaultRouteOptions,
+    InterfaceGeneration,
+    LinodeInterfaceOptions,
     LinodeInterfacePublicIPv4AddressOptions,
     LinodeInterfacePublicIPv4Options,
     LinodeInterfacePublicIPv6Options,
@@ -18,7 +20,27 @@
     LinodeInterfaceVPCIPv4Options,
     LinodeInterfaceVPCIPv4RangeOptions,
     LinodeInterfaceVPCOptions,
+    ReservedIPAddress,
 )
+from test.integration.helpers import get_test_label
+
+
+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(
@@ -359,3 +381,51 @@ 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

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,36 @@ 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()

test/integration/models/networking/test_networking.py

@@ -1,3 +1,4 @@
+import ipaddress
 import time
 from test.integration.conftest import (
     get_api_ca_file,
@@ -13,8 +14,12 @@
 
 import pytest
 
-from linode_api4 import Instance, LinodeClient
-from linode_api4.objects import Config, ConfigInterfaceIPv4, Firewall, IPAddress
+from linode_api4 import (
+    Instance,
+    LinodeClient,
+    ApiError,
+)
+from linode_api4.objects import Config, ConfigInterfaceIPv4, Firewall, IPAddress, ReservedIPAddress
 from linode_api4.objects.networking import (
     FirewallCreateDevicesOptions,
     NetworkTransferPrice,
@@ -351,3 +356,173 @@
     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  # NOTE: Skipped as tags not available in the API yet
+
+
+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
+
+
+@pytest.mark.skip   # NOTE: Skipped as tags not available in the API yet
+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 reserved_ip.tags == tags  # NOTE: Skipped as tags not available in the API yet
+
+    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):
+    # TODO: Currently it uses client (token), should not it be publicly accessible (no token required) ???
+    client = test_linode_client
+    types = client.networking.reserved_ip_types()
+    assert types.only
+
+    pricing = types.first()
+    assert pricing.id == "reserved-ipv4"
+    assert pricing.label == "Reserved IPv4"
+    assert pricing.price.hourly
+    # assert pricing.price.monthly is None
+    # assert pricing.region_prices == []
+
+
+@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 == tags  # TODO: Skipped as tags not available in the API yet
+
+
+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.rdns = "test.example.org"  # TODO: Should be enabled ?
+    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
+
+
+# def test_create_unassigned_reserved_ip_with_rdns():
+#     pass