test: add integration test

src-docs/firewall.py.md

@@ -50,7 +50,7 @@ Get the host IP address for the corresponding LXD network.
 
 ---
 
-<a href="../src/firewall.py#L83"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
+<a href="../src/firewall.py#L96"><img align="right" style="float:right;" src="https://img.shields.io/badge/-source-cccccc?style=flat-square"></a>
 
 ### <kbd>function</kbd> `refresh_firewall`
 

src/charm.py

@@ -914,7 +914,7 @@ def _refresh_firewall(self):
             FirewallEntry.decode(str(entry.host)) for entry in self._state.ssh_debug_infos
         ]
         firewall = Firewall("lxdbr0")
-        firewall.refresh_firewall(denylist, allowlist)
+        firewall.refresh_firewall(denylist=denylist, allowlist=allowlist)
         logger.debug(
             "firewall update, current firewall: %s",
             execute_command(["/usr/sbin/nft", "list", "ruleset"]),

src/firewall.py

@@ -71,14 +71,27 @@ def _exclude_network(
         networks: typing.List[ipaddress.IPv4Network],
         excludes: typing.List[ipaddress.IPv4Network],
     ) -> typing.Set[ipaddress.IPv4Network]:
-        total: typing.Set[ipaddress.IPv4Network] = set()
+        """Excludes the network segment from a pool of networks.
+
+        Args:
+            networks: The network pool to apply.
+            excludes: The networks to exclude from the pool.
+
+        Returns:
+            The network pool without the network segments in excludes.
+        """
+        total_excluded_networks = set(networks)
+
         for exclude in excludes:
-            for network in networks:
-                try:
-                    total.update(network.address_exclude(exclude))
-                except ValueError:
-                    total.update([network])
-        return total  # this currently doesn't work
+            scoped_excluded_networks: set[ipaddress.IPv4Network] = set()
+            for net in total_excluded_networks:
+                if net.overlaps(exclude):
+                    scoped_excluded_networks.update(net.address_exclude(exclude))
+                else:
+                    scoped_excluded_networks.add(net)
+            total_excluded_networks = scoped_excluded_networks
+
+        return total_excluded_networks
 
     def refresh_firewall(
         self,
@@ -147,13 +160,12 @@ def refresh_firewall(
             },
         ]
 
-        host_network = ipaddress.IPv4Network(host_ip)
         allowed_ips = [
-            host_network,
+            ipaddress.IPv4Network(host_ip),
             *[ipaddress.IPv4Network(entry.ip_range) for entry in (allowlist or [])],
         ]
         ips_to_deny = [ipaddress.IPv4Network(entry.ip_range) for entry in denylist]
-        denied_ips = self._exclude_network(ips_to_deny, allowed_ips)
+        denied_ips = self._exclude_network(networks=ips_to_deny, excludes=allowed_ips)
         egress_rules.extend(
             [
                 {

tests/integration/conftest.py

@@ -294,7 +294,7 @@ async def tmate_ssh_server_app_fixture(
     """tmate-ssh-server charm application related to GitHub-Runner app charm."""
     tmate_app: Application = await model.deploy("tmate-ssh-server", channel="edge")
     await app_no_wait.relate("debug-ssh", f"{tmate_app.name}:debug-ssh")
-    await model.wait_for_idle(status=ACTIVE, timeout=60 * 30)
+    await model.wait_for_idle(apps=[tmate_app.name], status=ACTIVE, timeout=60 * 30)
 
     return tmate_app
 

tests/integration/test_debug_ssh.py

@@ -9,25 +9,41 @@
 from github.Repository import Repository
 from github.WorkflowRun import WorkflowRun
 from juju.application import Application
+from juju.model import Model
 
 from tests.integration.helpers import dispatch_workflow, get_job_logs, get_workflow_runs
+from tests.status_name import ACTIVE
 
 logger = logging.getLogger(__name__)
 
 SSH_DEBUG_WORKFLOW_FILE_NAME = "workflow_dispatch_ssh_debug.yaml"
 
 
 async def test_ssh_debug(
+    model: Model,
     app_no_wait: Application,
     github_repository: Repository,
     test_github_branch: Branch,
     tmate_ssh_server_unit_ip: str,
 ):
     """
-    arrange: given an integrated GitHub-Runner charm and tmate-ssh-server charm.
+    arrange: given an integrated GitHub-Runner charm and tmate-ssh-server charm with a denylist
+        covering ip ranges of tmate-ssh-server.
     act: when canonical/action-tmate is triggered.
     assert: the ssh connection info from action-log and tmate-ssh-server matches.
     """
+    await app_no_wait.set_config(
+        {
+            "denylist": (
+                "0.0.0.0/8,10.0.0.0/8,100.64.0.0/10,169.254.0.0/16,"
+                "172.16.0.0/12,192.0.0.0/24,192.0.2.0/24,192.88.99.0/24,192.168.0.0/16,"
+                "198.18.0.0/15,198.51.100.0/24,203.0.113.0/24,224.0.0.0/4,233.252.0.0/24,"
+                "240.0.0.0/4"
+            ),
+        }
+    )
+    await model.wait_for_idle(status=ACTIVE, timeout=60 * 30)
+
     # trigger tmate action
     logger.info("Dispatching workflow_dispatch_ssh_debug.yaml workflow.")
     start_time = datetime.now(timezone.utc)

tests/unit/test_charm.py

@@ -446,7 +446,7 @@ def test__refresh_firewall(self, mock_firewall, *args):
 
         harness.charm._refresh_firewall()
         mocked_firewall_instance = mock_firewall.return_value
-        allowlist = mocked_firewall_instance.refresh_firewall.call_args_list[0][0][1]
+        allowlist = mocked_firewall_instance.refresh_firewall.call_args_list[0][1]["allowlist"]
         assert all(
             FirewallEntry(ip) in allowlist for ip in test_unit_ip_addresses
         ), "Expected IP firewall entry not found in allowlist arg."