From 74bdfbae9c64779fded1f70de778dcf248cfe348 Mon Sep 17 00:00:00 2001 From: evangelostsak Date: Sat, 19 Sep 2026 12:07:43 +0200 Subject: [PATCH] test(inventory): pin pruning, and assert mesh membership structurally Adds the mirror case pruning has to survive and its inverse, that an owned mesh is left untouched. The existing assertions searched the file for 'swarm:' and 'data:', which the credential keys mesh_private_key_swarm and mesh_private_key_data also match, so they passed against a file whose mesh block was empty. They now read the mesh names out of the parsed document. --- .../inventory/mesh/test_write.py | 63 ++++++++++++++++--- 1 file changed, 53 insertions(+), 10 deletions(-) diff --git a/tests/unit/python/cli/administration/inventory/mesh/test_write.py b/tests/unit/python/cli/administration/inventory/mesh/test_write.py index 3a9c611b80..6f56e8a1c6 100644 --- a/tests/unit/python/cli/administration/inventory/mesh/test_write.py +++ b/tests/unit/python/cli/administration/inventory/mesh/test_write.py @@ -16,8 +16,10 @@ from cli.administration.inventory.mesh.plan import plan_mesh from cli.administration.inventory.mesh.write import ( existing_public_keys, private_key_name, + prune_foreign_meshes, write_mesh, ) +from utils.cache.yaml import load_yaml_any from . import PROJECT_ROOT @@ -81,6 +83,20 @@ class MeshOnDisk(unittest.TestCase): encoding="utf-8" ) # nocheck: cache-read tempdir fixture rewritten between reads in one test + def _mesh_names(self, host: str) -> list[str]: + """The mesh names a host's file actually declares. + + Read structurally: the credential keys are named mesh_private_key_, + so a substring check for the mesh name matches them too and would pass + against a file whose mesh block is empty. + """ + document = load_yaml_any(str(self.host_vars / f"{host}.yml")) + return sorted( + document.get("applications", {}) + .get("svc-net-wireguard", {}) + .get("meshes", {}) + ) + def _write_all(self, *, rotate: bool = False) -> None: for spec in self.specs: mesh = plan_mesh( @@ -137,10 +153,10 @@ class TestMeshShapeOnDisk(MeshOnDisk, unittest.TestCase): self._write_all() for host in ("swarm-mgr-01", "swarm-wrk-01", "swarm-wrk-02"): with self.subTest(host=host): - self.assertIn("swarm:", self._text(host)) + self.assertIn("swarm", self._mesh_names(host)) for host in ("nfs-server", "swarm-bkp-01"): with self.subTest(host=host): - self.assertIn("data:", self._text(host)) + self.assertIn("data", self._mesh_names(host)) def test_a_worker_is_not_given_the_data_mesh(self): """Workers reach NFS through the hub, not as data-plane members. @@ -149,13 +165,11 @@ class TestMeshShapeOnDisk(MeshOnDisk, unittest.TestCase): directly and bypass the routing the topology depends on. """ self._write_all() - self.assertNotIn("data:", self._text("swarm-wrk-01")) + self.assertEqual(self._mesh_names("swarm-wrk-01"), ["swarm"]) def test_the_hub_carries_both_meshes(self): self._write_all() - text = self._text("swarm-mgr-01") - self.assertIn("swarm:", text) - self.assertIn("data:", text) + self.assertEqual(self._mesh_names("swarm-mgr-01"), ["data", "swarm"]) def test_a_spoke_routes_the_whole_pool_through_the_hub(self): """A worker must reach the data plane it is not a member of. @@ -226,14 +240,43 @@ class TestIdempotence(MeshOnDisk, unittest.TestCase): self._write_all() addresses = { host: plan_mesh( - self.specs[0], self.groups, existing_public_keys( - self.host_vars, list(HOSTS), "swarm" - ) - ).member(host).address + self.specs[0], + self.groups, + existing_public_keys(self.host_vars, list(HOSTS), "swarm"), + ) + .member(host) + .address for host in ("swarm-mgr-01", "swarm-wrk-01", "swarm-wrk-02") } self.assertEqual(len(set(addresses.values())), 3) + def test_a_mesh_a_host_does_not_own_is_pruned(self): + """Rewriting a member's own entry never removes a foreign one. + + After the mirror the NFS server keeps the hub's swarm entry, brings up + an interface claiming 10.100.0.1, and swallows the return path for + every worker -- while its own data mesh looks perfectly healthy. + """ + self._write_all() + hub_text = self._text("swarm-mgr-01") + (self.host_vars / "nfs-server.yml").write_text(hub_text, encoding="utf-8") + self.assertEqual(self._mesh_names("nfs-server"), ["data", "swarm"]) + + removed = prune_foreign_meshes(self.host_vars, list(HOSTS)) + self.assertEqual(removed.get("nfs-server"), ["data", "swarm"]) + self.assertEqual(self._mesh_names("nfs-server"), []) + + self._write_all() + self.assertEqual(self._mesh_names("nfs-server"), ["data"]) + + def test_pruning_leaves_an_owned_mesh_alone(self): + self._write_all() + before = {host: self._text(host) for host in HOSTS} + self.assertEqual(prune_foreign_meshes(self.host_vars, list(HOSTS)), {}) + for host in HOSTS: + with self.subTest(host=host): + self.assertEqual(self._text(host), before[host]) + def test_a_member_whose_credential_vanished_is_reminted(self): """A public key with no private key behind it is worse than no key.