Integration test fix for flaky test (#854)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: vshanthe

tests/integration/firewalls/test_firewalls.py

@@ -45,7 +45,7 @@ def test_list_firewall(firewall_id):
         + ["list", "--no-headers", "--text", "--delimiter", ","]
     )
 
-    assert re.search(firewall_id + "," + FIREWALL_LABEL + ",enabled", result)
+    assert re.search(rf"^{firewall_id},", result, re.MULTILINE)
 
 
 @pytest.mark.smoke

tests/integration/linodes/fixtures.py

@@ -10,6 +10,7 @@
     get_random_region_with_caps,
     get_random_text,
     retry_exec_test_command_with_delay,
+    wait_for_condition,
 )
 from tests.integration.linodes.helpers import (
     DEFAULT_LABEL,
@@ -133,8 +134,14 @@ def linode_instance_config_tests(linode_cloud_firewall):
 def linode_disk_config(linode_instance_config_tests):
     linode_id = linode_instance_config_tests
 
+    def disks_ready():
+        disks = get_disk_ids(linode_id=linode_id)
+        return len(disks) > 0
+
+    wait_for_condition(10, 120, disks_ready)
+
+    disk_id = get_disk_ids(linode_id=linode_id)[0]
     label = get_random_text(5) + "_config"
-    disk_id = get_disk_ids(linode_id=linode_id)[1]
 
     config_id = exec_test_command(
         BASE_CMDS["linodes"]

tests/integration/linodes/helpers.py

@@ -1,7 +1,10 @@
 import json
 import time
 
-from tests.integration.helpers import BASE_CMDS, exec_test_command
+from tests.integration.helpers import (
+    BASE_CMDS,
+    exec_test_command,
+)
 
 DEFAULT_RANDOM_PASS = exec_test_command(["openssl", "rand", "-base64", "32"])
 DEFAULT_REGION = "us-ord"

tests/integration/linodes/test_backups.py

@@ -91,9 +91,9 @@ def test_create_backup_with_backup_enabled(linode_backup_enabled):
     os.environ.get("RUN_LONG_TESTS", None) != "True",
     reason="Skipping long-running Test, to run set RUN_LONG_TESTS=True",
 )
-def test_take_snapshot_of_linode():
+def test_take_snapshot_of_linode(firewall_id):
     # get linode id after creation and wait for "running" status
-    linode_id = create_linode_and_wait()
+    linode_id = create_linode_and_wait(firewall_id)
 
     snapshot_label = "test_snapshot1"
 

tests/integration/linodes/test_configs.py

@@ -18,36 +18,31 @@
 )
 
 
-def test_config_create(linode_instance_config_tests):
+def test_config_create(linode_instance_config_tests, linode_disk_config):
     linode_id = linode_instance_config_tests
+    config_id = linode_disk_config
 
-    label = get_random_text(5) + "_config"
-    disk_id = get_disk_ids(linode_id=linode_id)[1]
-
-    result = exec_test_command(
+    res = exec_test_command(
         BASE_CMDS["linodes"]
         + [
-            "config-create",
+            "config-view",
             linode_id,
-            "--label",
-            label,
-            "--devices.sda.disk_id",
-            disk_id,
+            config_id,
             "--text",
         ]
     )
 
     headers = ["id", "label", "kernel"]
-
-    assert_headers_in_lines(headers, result.splitlines())
-    assert label in result
+    assert_headers_in_lines(headers, res.splitlines())
+    assert config_id in res
 
 
 def test_config_delete(linode_instance_config_tests):
     linode_id = linode_instance_config_tests
 
+    disk_id = get_disk_ids(linode_id=linode_id)[0]
+
     label = get_random_text(5) + "_config"
-    disk_id = get_disk_ids(linode_id=linode_id)[1]
 
     config_id = exec_test_command(
         BASE_CMDS["linodes"]
@@ -65,12 +60,7 @@ def test_config_delete(linode_instance_config_tests):
     )
 
     retry_exec_test_command_with_delay(
-        BASE_CMDS["linodes"]
-        + [
-            "config-delete",
-            linode_id,
-            config_id,
-        ]
+        BASE_CMDS["linodes"] + ["config-delete", linode_id, config_id]
     )
 
 

tests/integration/linodes/test_disk.py

@@ -17,7 +17,7 @@
 def test_disk_resize_clone_and_create(linode_instance_disk_tests):
     linode_id = linode_instance_disk_tests
 
-    disk_id = get_disk_ids(linode_id=linode_id)[1]
+    disk_id = get_disk_ids(linode_id=linode_id)[0]
 
     # resize disk
     retry_exec_test_command_with_delay(
@@ -99,7 +99,7 @@ def disk_poll_func():
 
 def test_disk_reset_password(linode_instance_disk_tests):
     linode_id = linode_instance_disk_tests
-    disk_id = get_disk_ids(linode_id)[1]
+    disk_id = get_disk_ids(linode_id)[0]
 
     retry_exec_test_command_with_delay(
         BASE_CMDS["linodes"]
@@ -118,7 +118,7 @@ def test_disk_reset_password(linode_instance_disk_tests):
 
 def test_disk_update(linode_instance_disk_tests):
     linode_id = linode_instance_disk_tests
-    disk_id = get_disk_ids(linode_id)[1]
+    disk_id = get_disk_ids(linode_id)[0]
 
     update_label = get_random_text(5) + "newdisk"
 

tests/integration/linodes/test_interfaces.py

@@ -4,6 +4,7 @@
 from tests.integration.helpers import (
     BASE_CMDS,
     exec_test_command,
+    wait_for_condition,
 )
 from tests.integration.linodes.fixtures import (  # noqa: F401
     linode_with_vpc_interface_as_args,
@@ -14,21 +15,42 @@
 def assert_interface_configuration(
     linode_json: Dict[str, Any], vpc_json: Dict[str, Any]
 ):
-    config_json = json.loads(
-        exec_test_command(
-            BASE_CMDS["linodes"]
-            + [
-                "configs-list",
-                str(linode_json["id"]),
-                "--json",
-                "--suppress-warnings",
-            ]
+    linode_id = str(linode_json["id"])
+    configs = []
+
+    def fetch_configs():
+        nonlocal configs
+        configs = json.loads(
+            exec_test_command(
+                BASE_CMDS["linodes"]
+                + [
+                    "configs-list",
+                    linode_id,
+                    "--json",
+                    "--suppress-warnings",
+                ]
+            )
         )
-    )[0]
+        return len(configs) > 0
+
+    wait_for_condition(5, 180, fetch_configs)
+
+    assert configs, f"No configs found for Linode {linode_id}"
+    config_json = configs[0]
+
+    interfaces = config_json["interfaces"]
 
-    vpc_interface = config_json["interfaces"][0]
-    public_interface = config_json["interfaces"][1]
+    vpc_interface = next((i for i in interfaces if i["purpose"] == "vpc"), None)
+    public_interface = next(
+        (i for i in interfaces if i["purpose"] == "public"), None
+    )
 
+    assert (
+        vpc_interface
+    ), "Expected interface with purpose 'vpc' in configuration"
+    assert (
+        public_interface
+    ), "Expected interface with purpose 'public' in configuration"
     assert vpc_interface["primary"]
     assert vpc_interface["purpose"] == "vpc"
     assert vpc_interface["subnet_id"] == vpc_json["subnets"][0]["id"]
@@ -38,7 +60,6 @@ def assert_interface_configuration(
     assert vpc_interface["ip_ranges"][0] == "10.0.0.6/32"
 
     assert not public_interface["primary"]
-    assert public_interface["purpose"] == "public"
 
 
 def test_with_vpc_interface_as_args(linode_with_vpc_interface_as_args):