TPT-4564: Refactor and fixes of integration tests

tests/integration/conftest.py

@@ -18,6 +18,7 @@
 
 from linodecli import ENV_TOKEN_NAME
 from tests.integration.helpers import (
+    DEFAULT_REGION,
     check_attribute_value,
     delete_target_id,
     exec_test_command,
@@ -228,8 +229,8 @@ def create_vpc_w_subnet():
                 vpc_label,
                 "--region",
                 region,
-                # "--ipv6.range",    TODO: Uncomment after VPC Dual Stack is ready to ship
-                # "auto",
+                "--ipv6.range",
+                "auto",
                 "--subnets.ipv4",
                 "10.0.0.0/24",
                 "--subnets.label",
@@ -248,3 +249,25 @@ def pytest_configure(config):
     config.addinivalue_line(
         "markers", "smoke: mark test as part of smoke test suite"
     )
+
+
+@pytest.fixture
+def create_reserved_ip(request):
+    tags = getattr(request, "param", None)
+    command = [
+        "linode-cli",
+        "networking",
+        "reserved-ip-add",
+        "--region",
+        DEFAULT_REGION,
+        "--json",
+    ]
+
+    if tags:
+        command += ["--tags", tags]
+
+    result = json.loads(exec_test_command(command))[0]
+
+    yield result
+
+    delete_target_id("networking", result["address"], "reserved-ip-delete")

tests/integration/database/fixtures.py

@@ -58,7 +58,7 @@ def mysql_cluster():
             "--label",
             mysql_database_label,
             "--engine",
-            "mysql/8",
+            "mysql/8.4",
             "--text",
             "--delimiter",
             ",",

tests/integration/database/test_database_engine_config.py

@@ -404,7 +404,7 @@ def mysql_db_engine_config(linode_cloud_firewall):
             + [
                 "mysql-create",
                 "--engine",
-                "mysql/8",
+                "mysql/8.4",
                 "--label",
                 label,
                 "--region",

tests/integration/firewalls/conftest.py

@@ -4,18 +4,32 @@
 
 from tests.integration.helpers import (
     BASE_CMDS,
+    check_attribute_value,
     delete_target_id,
     exec_test_command,
     get_random_text,
+    wait_for_condition,
 )
 
 
+def get_firewall_defaults():
+    result = json.loads(
+        exec_test_command(
+            BASE_CMDS["firewalls"]
+            + [
+                "firewall-settings-list",
+                "--json",
+            ],
+        )
+    )[0]["default_firewall_ids"]
+
+    return result
+
+
 @pytest.fixture(scope="function")
 def _firewall_id_and_label():
-    # generate a unique label
-    label = "fw-" + get_random_text(5)
-    # create it and capture the ID
-    result = exec_test_command(
+    label = "test-fw-" + get_random_text(5)
+    firewall_id = exec_test_command(
         BASE_CMDS["firewalls"]
         + [
             "create",
@@ -31,32 +45,41 @@ def _firewall_id_and_label():
             "id",
         ]
     )
-    fw_id = result
-    yield fw_id, label
-    # cleanup
-    delete_target_id(target="firewalls", id=fw_id)
+
+    # Verify firewall status is reachable before proceeding with tests
+    wait_for_condition(
+        5,
+        60,
+        check_attribute_value,
+        "firewalls",
+        "view",
+        firewall_id,
+        "status",
+        "enabled",
+    )
+
+    yield firewall_id, label
+
+    # cleanup (possible for non-default firewalls only)
+    delete_target_id(target="firewalls", id=firewall_id)
 
 
 @pytest.fixture(scope="function")
-def test_firewall_id(_firewall_id_and_label):
+def get_firewall_id(_firewall_id_and_label):
     """Only the ID, so old tests keep working."""
     return _firewall_id_and_label[0]
 
 
 @pytest.fixture(scope="function")
-def test_firewall_label(_firewall_id_and_label):
+def get_firewall_label(_firewall_id_and_label):
     """Only the label, for tests that need it explicitly."""
     return _firewall_id_and_label[1]
 
 
 @pytest.fixture
 def restore_firewall_defaults():
     # Fetch and store current default firewall settings
-    result = exec_test_command(
-        BASE_CMDS["firewalls"] + ["firewall-settings-list", "--json"]
-    )
-    settings = json.loads(result)
-    original_defaults = settings[0]["default_firewall_ids"]
+    original_defaults = get_firewall_defaults()
 
     yield original_defaults
 

tests/integration/firewalls/test_firewall_settings.py

@@ -8,7 +8,7 @@
 )
 
 
-def test_firewall_settings_defaults(test_firewall_id, test_firewall_label):
+def test_firewall_settings_defaults(get_firewall_id, get_firewall_label):
     # list all firewalls and extract the IDs
     list_result = exec_test_command(
         BASE_CMDS["firewalls"]
@@ -21,8 +21,8 @@ def test_firewall_settings_defaults(test_firewall_id, test_firewall_label):
     ]
 
     assert (
-        test_firewall_id in firewall_ids
-    ), f"{test_firewall_id} not found in firewall list"
+        get_firewall_id in firewall_ids
+    ), f"{get_firewall_id} not found in firewall list"
 
     # get the default firewall settings
     settings_result = exec_test_command(
@@ -57,7 +57,7 @@ def test_firewall_settings_defaults(test_firewall_id, test_firewall_label):
             ), f"{key} ID ({val}) not found in firewall list"
 
 
-def test_update_firewall_defaults(test_firewall_id, restore_firewall_defaults):
+def test_update_firewall_defaults(get_firewall_id, restore_firewall_defaults):
     # Fetch current default firewall settings
     settings = json.loads(
         exec_test_command(
@@ -89,8 +89,8 @@ def test_update_firewall_defaults(test_firewall_id, restore_firewall_defaults):
     ]
 
     assert (
-        test_firewall_id in firewall_ids
-    ), f"{test_firewall_id} not found in firewall list"
+        get_firewall_id in firewall_ids
+    ), f"{get_firewall_id} not found in firewall list"
 
     new_id = next(
         fid

tests/integration/firewalls/test_firewalls.py

@@ -195,20 +195,22 @@ def test_update_firewall(firewall_id):
     assert re.search(firewall_id + "," + updated_label + ",enabled", result)
 
 
-@pytest.mark.skip("skip until there is a way to delete default firewall")
-def test_firewall_settings_update_and_list(test_firewall_id):
+@pytest.mark.skip(
+    reason="Test skipped until there is a way to delete default firewall"
+)
+def test_firewall_settings_update_and_list(get_firewall_id):
     for cmd in [
         BASE_CMDS["firewalls"]
         + [
             "firewall-settings-update",
-            "--default_firewall_ids.vpc_interfac",
-            test_firewall_id,
+            "--default_firewall_ids.vpc_interface",
+            get_firewall_id,
             "--default_firewall_ids.public_interface",
-            test_firewall_id,
+            get_firewall_id,
             "--default_firewall_ids.nodebalancer",
-            test_firewall_id,
+            get_firewall_id,
             "--default_firewall_ids.linode",
-            test_firewall_id,
+            get_firewall_id,
             "--json",
         ],
         BASE_CMDS["firewalls"]
@@ -217,15 +219,11 @@ def test_firewall_settings_update_and_list(test_firewall_id):
             "--json",
         ],
     ]:
-        data = json.loads(exec_test_command(cmd).stdout.decode().rstrip())
-        firewall_ids = data[0]["default_firewall_ids"]
-        for key in [
-            "linode",
-            "nodebalancer",
-            "public_interface",
-            "vpc_interface",
-        ]:
-            assert firewall_ids[key] == int(test_firewall_id)
+        data = json.loads(exec_test_command(cmd))
+        def_firewall_ids = data[0]["default_firewall_ids"]
+
+        for key in def_firewall_ids:
+            assert def_firewall_ids[key] == int(get_firewall_id)
 
 
 def test_firewall_templates_list(monkeypatch: MonkeyPatch):

tests/integration/helpers.py

@@ -1,4 +1,5 @@
 import json
+import os
 import random
 import re
 import subprocess
@@ -14,6 +15,12 @@
 FAILED_STATUS_CODE = 256
 COMMAND_JSON_OUTPUT = ["--suppress-warnings", "--no-defaults", "--json"]
 
+DEFAULT_REGION = (
+    "pl-labkrk-2"
+    if "devcloud" in os.getenv("LINODE_CLI_API_HOST", "")
+    else "us-ord"
+)
+
 # TypeVars for generic type hints below
 T = TypeVar("T")
 

tests/integration/linodes/fixtures.py

@@ -5,6 +5,7 @@
 
 from tests.integration.helpers import (
     BASE_CMDS,
+    DEFAULT_REGION,
     delete_target_id,
     exec_test_command,
     get_random_region_with_caps,
@@ -16,7 +17,6 @@
     DEFAULT_LABEL,
     DEFAULT_LINODE_TYPE,
     DEFAULT_RANDOM_PASS,
-    DEFAULT_REGION,
     DEFAULT_TEST_IMAGE,
     create_linode,
     create_linode_and_wait,
@@ -487,8 +487,6 @@ def test_linode_instance(linode_cloud_firewall):
             "--delimiter",
             ",",
             "--no-headers",
-            "--format",
-            "id",
             "--no-defaults",
             "--format",
             "id",
@@ -700,3 +698,33 @@ def linode_with_authorization_key(linode_cloud_firewall):
 
     yield result
     delete_target_id(target="linodes", id=result[0])
+
+
+@pytest.fixture
+def linode_with_reserved_ip(linode_cloud_firewall, create_reserved_ip):
+    res_ip = create_reserved_ip["address"]
+
+    linode_id = exec_test_command(
+        BASE_CMDS["linodes"]
+        + [
+            "create",
+            "--type",
+            "g6-nanode-1",
+            "--region",
+            DEFAULT_REGION,
+            "--firewall_id",
+            linode_cloud_firewall,
+            "--ipv4",
+            res_ip,
+            "--text",
+            "--no-headers",
+            "--format",
+            "id",
+        ]
+    )
+
+    wait_until(linode_id=linode_id, timeout=180, status="offline")
+
+    yield res_ip, linode_id
+
+    delete_target_id(target="linodes", id=linode_id)

tests/integration/linodes/helpers.py

@@ -1,18 +1,13 @@
 import json
-import os
 import time
 
 from tests.integration.helpers import (
     BASE_CMDS,
+    DEFAULT_REGION,
     exec_test_command,
 )
 
 DEFAULT_RANDOM_PASS = exec_test_command(["openssl", "rand", "-base64", "32"])
-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/linodes/test_linode_interfaces.py

@@ -146,10 +146,6 @@ def test_interface_settings_update(
                 interface_id,
                 "--default_route.ipv6_interface_id",
                 interface_id,
-                "--default_route.ipv4_eligible_interface_ids",
-                interface_id,
-                "--default_route.ipv6_eligible_interface_ids",
-                interface_id,
                 "--json",
             ]
         )
@@ -161,8 +157,6 @@ def test_interface_settings_update(
     default_route = settings["default_route"]
     assert default_route["ipv4_interface_id"] == int(interface_id)
     assert default_route["ipv6_interface_id"] == int(interface_id)
-    assert default_route["ipv4_eligible_interface_ids"] == [int(interface_id)]
-    assert default_route["ipv6_eligible_interface_ids"] == [int(interface_id)]
 
 
 def test_interface_update(linode_interface_public, monkeypatch: MonkeyPatch):