Merge branch 'dev' into standardize_file_structure

Author: ykim-akamai

tests/integration/firewalls/conftest.py

@@ -0,0 +1,69 @@
+import json
+
+import pytest
+
+from tests.integration.helpers import (
+    delete_target_id,
+    exec_test_command,
+    get_random_text,
+)
+
+BASE_CMD = ["linode-cli", "firewalls"]
+
+
+@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(
+        BASE_CMD
+        + [
+            "create",
+            "--label",
+            label,
+            "--rules.outbound_policy",
+            "ACCEPT",
+            "--rules.inbound_policy",
+            "DROP",
+            "--text",
+            "--no-headers",
+            "--format",
+            "id",
+        ]
+    )
+    fw_id = result.stdout.decode().strip()
+    yield fw_id, label
+    # cleanup
+    delete_target_id(target="firewalls", id=fw_id)
+
+
+@pytest.fixture(scope="function")
+def test_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):
+    """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_CMD + ["firewall-settings-list", "--json"])
+    settings = json.loads(result.stdout.decode())
+    original_defaults = settings[0]["default_firewall_ids"]
+
+    yield original_defaults
+
+    # Restore the original defaults after test
+    args = []
+    for key, val in original_defaults.items():
+        if val is not None:
+            args.extend([f"--default_firewall_ids.{key}", str(val)])
+
+    if args:
+        exec_test_command(BASE_CMD + ["firewall-settings-update"] + args)

tests/integration/firewalls/test_firewall_settings.py

@@ -0,0 +1,135 @@
+import json
+
+import pytest
+
+from tests.integration.helpers import (
+    exec_test_command,
+)
+
+BASE_CMD = ["linode-cli", "firewalls"]
+
+
+def test_firewall_settings_defaults(test_firewall_id, test_firewall_label):
+    # list all firewalls and extract the IDs
+    list_result = (
+        exec_test_command(
+            BASE_CMD + ["list", "--no-headers", "--text", "--delimiter", ","]
+        )
+        .stdout.decode()
+        .strip()
+    )
+
+    firewall_lines = list_result.splitlines()
+    firewall_ids = [
+        line.split(",")[0].strip() for line in firewall_lines if line
+    ]
+
+    assert (
+        test_firewall_id in firewall_ids
+    ), f"{test_firewall_id} not found in firewall list"
+
+    # get the default firewall settings
+    settings_result = exec_test_command(
+        BASE_CMD + ["firewall-settings-list", "--json"]
+    ).stdout.decode()
+
+    settings = json.loads(settings_result)
+    assert (
+        isinstance(settings, list) and len(settings) > 0
+    ), "Unexpected settings format"
+
+    default_ids = settings[0]["default_firewall_ids"]
+
+    # Validate all expected keys exist and map to a valid firewall ID
+    expected_keys = [
+        "linode",
+        "nodebalancer",
+        "public_interface",
+        "vpc_interface",
+    ]
+
+    for key in expected_keys:
+        assert key in default_ids, f"Missing default_firewall_ids key: {key}"
+        val = default_ids[key]
+        assert val is None or isinstance(
+            val, int
+        ), f"{key} value is not None or int: {val}"
+        if isinstance(val, int):
+            assert val > 0, f"{key} should be a non-zero firewall ID"
+            assert (
+                str(val) in firewall_ids
+            ), f"{key} ID ({val}) not found in firewall list"
+
+
+def test_update_firewall_defaults(test_firewall_id, restore_firewall_defaults):
+    # Fetch current default firewall settings
+    settings = json.loads(
+        exec_test_command(
+            BASE_CMD + ["firewall-settings-list", "--json"]
+        ).stdout.decode()
+    )
+    default_ids_before = settings[0]["default_firewall_ids"]
+
+    # Skip if no default firewall configured at all
+    if all(v is None for v in default_ids_before.values()):
+        pytest.skip("Skipping: no default firewall configured for the account.")
+
+    old_default_id = (
+        str(default_ids_before["linode"])
+        if default_ids_before["linode"]
+        else None
+    )
+
+    # List all firewalls
+    firewall_list = (
+        exec_test_command(
+            BASE_CMD + ["list", "--no-headers", "--text", "--delimiter", ","]
+        )
+        .stdout.decode()
+        .strip()
+    )
+
+    firewall_ids = [
+        line.split(",")[0].strip()
+        for line in firewall_list.splitlines()
+        if line
+    ]
+
+    assert (
+        test_firewall_id in firewall_ids
+    ), f"{test_firewall_id} not found in firewall list"
+
+    new_id = next(
+        fid
+        for fid in firewall_ids
+        if old_default_id is None or fid != old_default_id
+    )
+
+    # Update all default firewall IDs to the new one
+    exec_test_command(
+        BASE_CMD
+        + [
+            "firewall-settings-update",
+            "--default_firewall_ids.linode",
+            new_id,
+            "--default_firewall_ids.nodebalancer",
+            new_id,
+            "--default_firewall_ids.public_interface",
+            new_id,
+            "--default_firewall_ids.vpc_interface",
+            new_id,
+            "--json",
+        ]
+    )
+
+    # Verify update
+    updated_settings = json.loads(
+        exec_test_command(
+            BASE_CMD + ["firewall-settings-list", "--json"]
+        ).stdout.decode()
+    )[0]["default_firewall_ids"]
+
+    for key, val in updated_settings.items():
+        assert (
+            str(val) == new_id
+        ), f"{key} was not updated (expected {new_id}, got {val})"

tests/integration/firewalls/test_firewalls.py

@@ -14,10 +14,8 @@
     delete_target_id,
     exec_failing_test_command,
     exec_test_command,
-    get_random_text,
 )
 
-
 @pytest.mark.smoke
 def test_view_firewall(firewall_id):
     firewall_id = firewall_id
@@ -35,7 +33,7 @@ def test_view_firewall(firewall_id):
     )
 
     assert re.search(firewall_id + "," + FIREWALL_LABEL + ",enabled", result)
-
+    
 
 def test_list_firewall(firewall_id):
     firewall_id = firewall_id

tests/integration/firewalls/test_firewalls_rules.py

@@ -4,13 +4,6 @@
 from tests.integration.firewalls.fixtures import (  # noqa: F401
     firewall_id,
 )
-from tests.integration.helpers import (
-    BASE_CMDS,
-    delete_target_id,
-    exec_test_command,
-    get_random_text,
-)
-
 
 def test_add_rule_to_existing_firewall(firewall_id):
     firewall_id = firewall_id
@@ -81,9 +74,9 @@ def test_swap_rules():
         ]
     )
 
-    swapped_rules = "[" + inbound_rule_2 + "," + inbound_rule_1 + "]"
+    swapped_rules_obj = list(reversed(inbound_rules_obj))
+    swapped_rules = json.dumps(swapped_rules_obj)
 
-    # swapping rules
     result = json.loads(
         exec_test_command(
             BASE_CMDS["firewalls"]
@@ -97,8 +90,8 @@ def test_swap_rules():
         )
     )
 
-    assert result[0]["inbound"][0] == json.loads(inbound_rule_2)
-    assert result[0]["inbound"][1] == json.loads(inbound_rule_1)
+    assert result[0]["inbound"][0] == swapped_rules_obj[0]
+    assert result[0]["inbound"][1] == swapped_rules_obj[1]
 
     delete_target_id(target="firewalls", id=firewall_id)