fix_firewall_test

Author: vshanthe

test/integration/models/firewall/test_firewall.py

@@ -8,18 +8,43 @@
 
 
 @pytest.fixture(scope="session")
-def linode_fw(test_linode_client):
+def linode_fw(test_linode_client, test_firewall):
     client = test_linode_client
     region = get_region(client, {"Linodes", "Cloud Firewall"}, site_type="core")
     label = get_test_label()
 
-    linode_instance, password = client.linode.instance_create(
+    linode_instance, _ = client.linode.instance_create(
         "g6-nanode-1", region, image="linode/debian12", label=label
     )
 
-    yield linode_instance
+    for fw in linode_instance.firewalls():
+        fw_device = next((d for d in fw.devices if d.entity.id == linode_instance.id), None)
+        if fw_device:
+            fw_device.delete()
+
+    firewall = client.networking.firewall_create(
+        label=f"{label}-fw",
+        rules={
+            "inbound": [],
+            "inbound_policy": "DROP",
+            "outbound": [],
+            "outbound_policy": "ACCEPT",
+        },
+    )
+
+    client.post(
+        f"/networking/firewalls/{firewall.id}/devices",
+        data={"id": linode_instance.id, "type": "linode"},
+    )
 
-    linode_instance.delete()
+    try:
+        yield linode_instance, firewall
+    finally:
+        for obj, name in ((firewall, "firewall"), (linode_instance, "Linode")):
+            try:
+                obj.delete()
+            except Exception as e:
+                print(f"Warning: failed to delete {name} {obj.id}: {e}")
 
 
 @pytest.mark.smoke
@@ -64,34 +89,18 @@ def test_update_firewall_rules(test_linode_client, test_firewall):
 
 
 def test_get_devices(test_linode_client, linode_fw):
-    linode = linode_fw
-
-    firewalls = list(linode.firewalls())
-    assert len(firewalls) > 0
-
-    firewall = test_linode_client.load(Firewall, firewalls[0].id)
-
-    devices = list(firewall.devices)
+    linode, firewall = linode_fw
+    devices = list(test_linode_client.load(Firewall, firewall.id).devices)
     assert any(d.entity.id == linode.id for d in devices)
 
 
 def test_get_device(test_linode_client, linode_fw):
-    linode = linode_fw
-
-    firewalls = list(linode.firewalls())
-    assert firewalls, "No firewalls found on Linode"
-
-    firewall = test_linode_client.load(Firewall, firewalls[0].id)
-
-    devices = list(firewall.devices)
+    linode, firewall = linode_fw
+    devices = list(test_linode_client.load(Firewall, firewall.id).devices)
     assert devices, "No devices found on Firewall"
 
-    device = next((d for d in devices if d.entity.id == linode.id), None)
-    assert device is not None, f"No FirewallDevice found for Linode {linode.id}"
-
-    firewall_device = test_linode_client.load(
-        FirewallDevice, device.id, firewall.id
-    )
+    device = next(d for d in devices if d.entity.id == linode.id)
+    fw_device = test_linode_client.load(FirewallDevice, device.id, firewall.id)
 
-    assert firewall_device.entity.type == "linode"
-    assert f"/v4/linode/instances/{linode.id}" in firewall_device.entity.url
+    assert fw_device.entity.type == "linode"
+    assert f"/v4/linode/instances/{linode.id}" in fw_device.entity.url