TPT-4280: Add integration tests for Reserved IP for IPv4

tests/integration/linodes/helpers.py

@@ -1,4 +1,5 @@
 import json
+import os
 import time
 
 from tests.integration.helpers import (
@@ -7,7 +8,11 @@
 )
 
 DEFAULT_RANDOM_PASS = exec_test_command(["openssl", "rand", "-base64", "32"])
-DEFAULT_REGION = "us-ord"
+DEFAULT_REGION = (
+    "pl-labkrk-2"
+    if "devcloud" in os.getenv("LINODE_CLI_API_HOST")
+    else "us-ord"
+)
 
 DEFAULT_TEST_IMAGE = exec_test_command(
     [

tests/integration/networking/fixtures.py

@@ -0,0 +1,69 @@
+import pytest
+
+from tests.integration.helpers import (
+    BASE_CMDS,
+    delete_target_id,
+    exec_test_command,
+)
+from tests.integration.linodes.helpers import (
+    DEFAULT_REGION,
+    create_linode,
+    create_linode_and_wait,
+)
+
+
+@pytest.fixture
+def create_reserved_ip(request):
+    tags = getattr(request, "param", None)
+    command = BASE_CMDS["networking"] + [
+        "reserved-ip-add",
+        "--region",
+        DEFAULT_REGION,
+        "--text",
+        "--delimiter",
+        ",",
+    ]
+
+    if tags:
+        command += ["--tags", tags]
+
+    headers, values = get_command_heads_and_vals(command)
+    yield headers, values
+
+    delete_target_id("networking", values[0], "reserved-ip-delete")
+
+
+@pytest.fixture(scope="package")
+def test_linode_id(linode_cloud_firewall):
+    linode_id = create_linode_and_wait(firewall_id=linode_cloud_firewall)
+
+    yield linode_id
+
+    delete_target_id(target="linodes", id=linode_id)
+
+
+@pytest.fixture(scope="package")
+def test_linode_id_shared_ipv4(linode_cloud_firewall):
+    target_region = "us-mia"
+
+    linode_ids = (
+        create_linode(
+            test_region=target_region, firewall_id=linode_cloud_firewall
+        ),
+        create_linode(
+            test_region=target_region, firewall_id=linode_cloud_firewall
+        ),
+    )
+
+    yield linode_ids
+
+    for id_num in linode_ids:
+        delete_target_id(target="linodes", id=id_num)
+
+
+def get_command_heads_and_vals(command):
+    result = exec_test_command(command).splitlines()
+    headers = [item for item in result[0].split(",")]
+    values = [item for item in result[1].split(",")]
+
+    return headers, values

tests/integration/networking/test_networking.py

@@ -1,3 +1,4 @@
+import ipaddress
 import json
 import re
 
@@ -6,41 +7,26 @@
 
 from tests.integration.helpers import (
     BASE_CMDS,
-    delete_target_id,
+    assert_headers_in_lines,
     exec_test_command,
 )
-from tests.integration.linodes.helpers import (
-    create_linode,
-    create_linode_and_wait,
+from tests.integration.linodes.helpers import DEFAULT_REGION
+from tests.integration.networking.fixtures import (  # noqa: F401
+    create_reserved_ip,
+    get_command_heads_and_vals,
+    test_linode_id,
+    test_linode_id_shared_ipv4,
 )
 
-
-@pytest.fixture(scope="package")
-def test_linode_id(linode_cloud_firewall):
-    linode_id = create_linode_and_wait(firewall_id=linode_cloud_firewall)
-
-    yield linode_id
-
-    delete_target_id(target="linodes", id=linode_id)
-
-
-@pytest.fixture(scope="package")
-def test_linode_id_shared_ipv4(linode_cloud_firewall):
-    target_region = "us-mia"
-
-    linode_ids = (
-        create_linode(
-            test_region=target_region, firewall_id=linode_cloud_firewall
-        ),
-        create_linode(
-            test_region=target_region, firewall_id=linode_cloud_firewall
-        ),
-    )
-
-    yield linode_ids
-
-    for id in linode_ids:
-        delete_target_id(target="linodes", id=id)
+RESERVED_IP_HEADERS = [
+    "address",
+    "type",
+    "public",
+    "rdns",
+    "linode_id",
+    "reserved",
+    "tags",
+]
 
 
 def has_shared_ip(linode_id: int, ip: str) -> bool:
@@ -61,6 +47,19 @@ def has_shared_ip(linode_id: int, ip: str) -> bool:
     return False
 
 
+def verify_reserved_ip(reserved_ip):
+    assert isinstance(
+        ipaddress.ip_address(reserved_ip[0]), ipaddress.IPv4Address
+    )
+    assert reserved_ip[1] == "ipv4"
+    assert reserved_ip[2] == "True"
+    assert reserved_ip[4] == DEFAULT_REGION
+    assert not reserved_ip[5]
+    assert reserved_ip[7] == "True"
+    # TODO: To be clarified if it should be returned in CLI
+    # assert reserved_ip["assigned_entity"] is None
+
+
 def test_display_ips_for_available_linodes(test_linode_id):
     result = exec_test_command(
         BASE_CMDS["networking"]
@@ -69,7 +68,7 @@ def test_display_ips_for_available_linodes(test_linode_id):
 
     assert re.search(r"^[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}", result)
     assert re.search(
-        r"ipv4,True,[0-9]{1,3}\-[0-9]{1,3}\-[0-9]{1,3}\-[0-9]{1,3}\.ip.linodeusercontent.com,.*,[0-9][0-9][0-9][0-9][0-9][0-9][0-9]*",
+        r"ipv4,True,(False|True),[0-9]{1,3}\-[0-9]{1,3}\-[0-9]{1,3}\-[0-9]{1,3}\.ip\.linodeusercontent\.com,[0-9]*",
         result,
     )
     assert re.search("ipv6,True,,.*,[0-9][0-9][0-9][0-9][0-9][0-9]*", result)
@@ -145,6 +144,162 @@ def test_allocate_additional_private_ipv4_address(test_linode_id):
     )
 
 
+@pytest.mark.smoke
+@pytest.mark.parametrize("create_reserved_ip", ["test", None], indirect=True)
+def test_create_reserved_ip(create_reserved_ip):
+    headers, reserved_ip = create_reserved_ip
+    assert_headers_in_lines(RESERVED_IP_HEADERS, [headers])
+    verify_reserved_ip(reserved_ip)
+
+    tags = reserved_ip[-1]
+    assert tags == "test" if tags else tags == ""
+
+
+@pytest.mark.parametrize("create_reserved_ip", ["test"], indirect=True)
+def test_update_reserved_ip_tags(create_reserved_ip):
+    _, reserved_ip = create_reserved_ip
+    assert reserved_ip[-1] == "test"
+
+    result = exec_test_command(
+        BASE_CMDS["networking"]
+        + [
+            "reserved-ip-update",
+            "--tags",
+            "updated",
+            "--tags",
+            "updated2",
+            reserved_ip[0],
+            "--text",
+            "--no-headers",
+            "--delimiter",
+            ",",
+        ]
+    ).split(",")
+    verify_reserved_ip(result)
+    assert result[-1] == "updated updated2"
+
+
+def test_create_reserved_ip_assigned(create_reserved_ip, test_linode_id):
+    _, reserved_ip = create_reserved_ip
+    linode_id = test_linode_id
+
+    exec_test_command(
+        BASE_CMDS["networking"]
+        + [
+            "ip-assign",
+            "--assignments.linode_id",
+            linode_id,
+            "--assignments.address",
+            reserved_ip[0],
+            "--region",
+            DEFAULT_REGION,
+        ]
+    )
+
+    command = BASE_CMDS["linodes"] + [
+        "ip-view",
+        linode_id,
+        reserved_ip[0],
+        "--text",
+        "--delimiter",
+        ",",
+    ]
+    headers, values = get_command_heads_and_vals(command)
+
+    assert_headers_in_lines(RESERVED_IP_HEADERS[:-1], [headers])
+    # TODO: To be clarified if tags should be returned in CLI (currently it is)
+    # assert "tags" not in headers
+    assert values[0] == reserved_ip[0]
+    assert str(values[5]) == linode_id
+    assert values[7] == "True"
+
+
+def test_get_reserved_ip_types():
+    headers_exp = ["id", "label", "price.hourly", "price.monthly"]
+    command = BASE_CMDS["networking"] + [
+        "reserved-ip-types-list",
+        "--text",
+        "--delimiter",
+        ",",
+    ]
+    headers, values = get_command_heads_and_vals(command)
+
+    assert_headers_in_lines(headers_exp, [headers])
+    assert values[0] == "reserved-ipv4"
+    assert values[1] == "Reserved IPv4"
+    assert any(price != 0 for price in values[2:4])
+
+
+def test_get_reserved_ip_view(create_reserved_ip):
+    _, reserved_ip = create_reserved_ip
+    command = BASE_CMDS["networking"] + [
+        "reserved-ip-view",
+        reserved_ip[0],
+        "--text",
+        "--delimiter",
+        ",",
+    ]
+    headers, values = get_command_heads_and_vals(command)
+
+    assert_headers_in_lines(RESERVED_IP_HEADERS, [headers])
+    verify_reserved_ip(values)
+
+
+def test_get_reserved_ips_list(create_reserved_ip):
+    result = exec_test_command(
+        BASE_CMDS["networking"]
+        + [
+            "reserved-ips-list",
+            "--text",
+            "--no-headers",
+            "--format",
+            "reserved",
+        ]
+    ).splitlines()
+
+    assert all(item == "True" for item in result)
+
+
+def test_update_ephemeral_to_reserved(test_linode_id):
+    linode_id = test_linode_id
+
+    ephemeral_ip = exec_test_command(
+        BASE_CMDS["linodes"]
+        + [
+            "view",
+            linode_id,
+            "--text",
+            "--no-headers",
+            "--format",
+            "ipv4",
+        ]
+    ).split(" ")[0]
+
+    exec_test_command(
+        BASE_CMDS["networking"]
+        + [
+            "ip-update",
+            ephemeral_ip,
+            "--reserved",
+            "true",
+        ]
+    )
+
+    is_reserved = exec_test_command(
+        BASE_CMDS["networking"]
+        + [
+            "reserved-ip-view",
+            ephemeral_ip,
+            "--text",
+            "--no-headers",
+            "--format",
+            "reserved",
+        ]
+    )
+
+    assert is_reserved == "True"
+
+
 def test_share_ipv4_address(
     test_linode_id_shared_ipv4, monkeypatch: MonkeyPatch
 ):

tests/integration/tags/test_tags.py

@@ -8,6 +8,9 @@
     exec_test_command,
     get_random_text,
 )
+from tests.integration.networking.fixtures import (  # noqa: F401
+    create_reserved_ip,
+)
 
 
 @pytest.fixture(scope="session")
@@ -42,3 +45,36 @@ def test_fail_to_create_tag_shorter_than_three_char():
     )
     assert "Request failed: 400" in result
     assert "Length must be 3-50 characters" in result
+
+
+def test_create_tag_for_reserved_ip(create_reserved_ip):
+    _, reserved_ip = create_reserved_ip
+    tag_name = get_random_text(5) + "-tag"
+
+    exec_test_command(
+        BASE_CMDS["tags"]
+        + [
+            "create",
+            "--label",
+            tag_name,
+            "--reserved_ipv4_addresses",
+            reserved_ip[0],
+        ]
+    )
+
+    result = exec_test_command(
+        BASE_CMDS["networking"]
+        + [
+            "reserved-ip-view",
+            reserved_ip[0],
+            "--text",
+            "--no-headers",
+            "--format",
+            "tags",
+        ]
+    )
+
+    assert result == tag_name
+
+    # TODO: GET tags/{label} and DELETE tags/{label} missed or not implemented at all
+    # delete_target_id("tags", tag_name)