test: Add Linode Cloud Firewall for integration tests (#408)

* add cloud firewall to integration tests

* pr comments

* update fixture name

* lint

test/integration/conftest.py

@@ -1,9 +1,11 @@
+import ipaddress
 import os
 import random
 import time
 from typing import Set
 
 import pytest
+import requests
 
 from linode_api4 import ApiError
 from linode_api4.linode_client import LinodeClient
@@ -50,16 +52,90 @@ def run_long_tests():
     return os.environ.get(RUN_LONG_TESTS, None)
 
 
+@pytest.fixture(autouse=True, scope="session")
+def e2e_test_firewall(test_linode_client):
+    def is_valid_ipv4(address):
+        try:
+            ipaddress.IPv4Address(address)
+            return True
+        except ipaddress.AddressValueError:
+            return False
+
+    def is_valid_ipv6(address):
+        try:
+            ipaddress.IPv6Address(address)
+            return True
+        except ipaddress.AddressValueError:
+            return False
+
+    def get_public_ip(ip_version="ipv4"):
+        url = (
+            f"https://api64.ipify.org?format=json"
+            if ip_version == "ipv6"
+            else f"https://api.ipify.org?format=json"
+        )
+        response = requests.get(url)
+        return str(response.json()["ip"])
+
+    def create_inbound_rule(ipv4_address, ipv6_address):
+        rule = [
+            {
+                "protocol": "TCP",
+                "ports": "22",
+                "addresses": {},
+                "action": "ACCEPT",
+            }
+        ]
+        if is_valid_ipv4(ipv4_address):
+            rule[0]["addresses"]["ipv4"] = [f"{ipv4_address}/32"]
+
+        if is_valid_ipv6(ipv6_address):
+            rule[0]["addresses"]["ipv6"] = [f"{ipv6_address}/128"]
+
+        return rule
+
+        # Fetch the public IP addresses
+
+    ipv4_address = get_public_ip("ipv4")
+    ipv6_address = get_public_ip("ipv6")
+
+    inbound_rule = create_inbound_rule(ipv4_address, ipv6_address)
+
+    client = test_linode_client
+
+    rules = {
+        "outbound": [],
+        "outbound_policy": "ACCEPT",
+        "inbound": inbound_rule,
+        "inbound_policy": "DROP",
+    }
+
+    label = "cloud_firewall_" + str(int(time.time()))
+
+    firewall = client.networking.firewall_create(
+        label=label, rules=rules, status="enabled"
+    )
+
+    yield firewall
+
+    firewall.delete()
+
+
 @pytest.fixture(scope="session")
-def create_linode(test_linode_client):
+def create_linode(test_linode_client, e2e_test_firewall):
     client = test_linode_client
+
     available_regions = client.regions()
     chosen_region = available_regions[4]
     timestamp = str(time.time_ns())
     label = "TestSDK-" + timestamp
 
     linode_instance, password = client.linode.instance_create(
-        "g6-nanode-1", chosen_region, image="linode/debian10", label=label
+        "g6-nanode-1",
+        chosen_region,
+        image="linode/debian12",
+        label=label,
+        firewall=e2e_test_firewall,
     )
 
     yield linode_instance
@@ -68,15 +144,20 @@ def create_linode(test_linode_client):
 
 
 @pytest.fixture
-def create_linode_for_pass_reset(test_linode_client):
+def create_linode_for_pass_reset(test_linode_client, e2e_test_firewall):
     client = test_linode_client
+
     available_regions = client.regions()
     chosen_region = available_regions[4]
     timestamp = str(time.time_ns())
     label = "TestSDK-" + timestamp
 
     linode_instance, password = client.linode.instance_create(
-        "g6-nanode-1", chosen_region, image="linode/debian10", label=label
+        "g6-nanode-1",
+        chosen_region,
+        image="linode/debian10",
+        label=label,
+        firewall=e2e_test_firewall,
     )
 
     yield linode_instance, password
@@ -303,15 +384,19 @@ def create_vpc_with_subnet(test_linode_client, create_vpc):
 
 @pytest.fixture(scope="session")
 def create_vpc_with_subnet_and_linode(
-    test_linode_client, create_vpc_with_subnet
+    test_linode_client, create_vpc_with_subnet, e2e_test_firewall
 ):
     vpc, subnet = create_vpc_with_subnet
 
     timestamp = str(int(time.time()))
     label = "TestSDK-" + timestamp
 
     instance, password = test_linode_client.linode.instance_create(
-        "g6-standard-1", vpc.region, image="linode/debian11", label=label
+        "g6-standard-1",
+        vpc.region,
+        image="linode/debian11",
+        label=label,
+        firewall=e2e_test_firewall,
     )
 
     yield vpc, subnet, instance, password

test/integration/linode_client/test_linode_client.py

@@ -8,15 +8,19 @@
 from linode_api4.objects import ConfigInterface, ObjectStorageKeys, Region
 
 
-@pytest.fixture(scope="session", autouse=True)
-def setup_client_and_linode(test_linode_client):
+@pytest.fixture(scope="session")
+def setup_client_and_linode(test_linode_client, e2e_test_firewall):
     client = test_linode_client
     available_regions = client.regions()
     chosen_region = available_regions[4]  # us-ord (Chicago)
     label = get_test_label()
 
     linode_instance, password = client.linode.instance_create(
-        "g6-nanode-1", chosen_region, image="linode/debian10", label=label
+        "g6-nanode-1",
+        chosen_region,
+        image="linode/debian10",
+        label=label,
+        firewall=e2e_test_firewall,
     )
 
     yield client, linode_instance

test/integration/models/linode/test_linode.py

@@ -32,7 +32,7 @@ def linode_with_volume_firewall(test_linode_client):
         "outbound": [],
         "outbound_policy": "DROP",
         "inbound": [],
-        "inbound_policy": "ACCEPT",
+        "inbound_policy": "DROP",
     }
 
     linode_instance, password = client.linode.instance_create(
@@ -68,15 +68,19 @@ def linode_with_volume_firewall(test_linode_client):
 
 
 @pytest.fixture(scope="session")
-def linode_for_network_interface_tests(test_linode_client):
+def linode_for_network_interface_tests(test_linode_client, e2e_test_firewall):
     client = test_linode_client
     available_regions = client.regions()
     chosen_region = available_regions[4]
     timestamp = str(time.time_ns())
     label = "TestSDK-" + timestamp
 
     linode_instance, password = client.linode.instance_create(
-        "g6-nanode-1", chosen_region, image="linode/debian10", label=label
+        "g6-nanode-1",
+        chosen_region,
+        image="linode/debian10",
+        label=label,
+        firewall=e2e_test_firewall,
     )
 
     yield linode_instance
@@ -85,7 +89,7 @@ def linode_for_network_interface_tests(test_linode_client):
 
 
 @pytest.fixture
-def linode_for_disk_tests(test_linode_client):
+def linode_for_disk_tests(test_linode_client, e2e_test_firewall):
     client = test_linode_client
     available_regions = client.regions()
     chosen_region = available_regions[4]
@@ -96,6 +100,7 @@ def linode_for_disk_tests(test_linode_client):
         chosen_region,
         image="linode/alpine3.19",
         label=label + "_long_tests",
+        firewall=e2e_test_firewall,
     )
 
     # Provisioning time
@@ -118,7 +123,7 @@ def linode_for_disk_tests(test_linode_client):
 
 
 @pytest.fixture
-def create_linode_for_long_running_tests(test_linode_client):
+def create_linode_for_long_running_tests(test_linode_client, e2e_test_firewall):
     client = test_linode_client
     available_regions = client.regions()
     chosen_region = available_regions[4]
@@ -129,6 +134,7 @@ def create_linode_for_long_running_tests(test_linode_client):
         chosen_region,
         image="linode/debian10",
         label=label + "_long_tests",
+        firewall=e2e_test_firewall,
     )
 
     yield linode_instance
@@ -411,7 +417,6 @@ def test_disk_resize_and_duplicate(test_linode_client, linode_for_disk_tests):
     time.sleep(40)
 
     wait_for_disk_status(dup_disk, 120)
-
     assert dup_disk.linode_id == linode.id
 
 

test/integration/models/nodebalancer/test_nodebalancer.py

@@ -7,7 +7,7 @@
 
 
 @pytest.fixture(scope="session")
-def linode_with_private_ip(test_linode_client):
+def linode_with_private_ip(test_linode_client, e2e_test_firewall):
     client = test_linode_client
     available_regions = client.regions()
     chosen_region = available_regions[4]
@@ -19,6 +19,7 @@ def linode_with_private_ip(test_linode_client):
         image="linode/debian10",
         label=label,
         private_ip=True,
+        firewall=e2e_test_firewall,
     )
 
     yield linode_instance
@@ -27,13 +28,15 @@ def linode_with_private_ip(test_linode_client):
 
 
 @pytest.fixture(scope="session")
-def create_nb_config(test_linode_client):
+def create_nb_config(test_linode_client, e2e_test_firewall):
     client = test_linode_client
     available_regions = client.regions()
     chosen_region = available_regions[4]
     label = "nodebalancer_test"
 
-    nb = client.nodebalancer_create(region=chosen_region, label=label)
+    nb = client.nodebalancer_create(
+        region=chosen_region, label=label, firewall=e2e_test_firewall.id
+    )
 
     config = nb.config_create()
 

test/integration/models/volume/test_volume.py

@@ -13,15 +13,19 @@
 
 
 @pytest.fixture(scope="session")
-def linode_for_volume(test_linode_client):
+def linode_for_volume(test_linode_client, e2e_test_firewall):
     client = test_linode_client
     available_regions = client.regions()
     chosen_region = available_regions[4]
     timestamp = str(time.time_ns())
     label = "TestSDK-" + timestamp
 
     linode_instance, password = client.linode.instance_create(
-        "g6-nanode-1", chosen_region, image="linode/debian10", label=label
+        "g6-nanode-1",
+        chosen_region,
+        image="linode/debian10",
+        label=label,
+        firewall=e2e_test_firewall,
     )
 
     yield linode_instance