From 777d889980ac5d0c31f924abd836a32ffac2c832 Mon Sep 17 00:00:00 2001 From: BirSite Automation Date: Sun, 19 Jul 2026 17:46:16 +0000 Subject: [PATCH] test: require portable Traefik router priorities --- tests/test_compose_contract.py | 77 ++++++++++++++++++++++++++++++++-- 1 file changed, 73 insertions(+), 4 deletions(-) diff --git a/tests/test_compose_contract.py b/tests/test_compose_contract.py index 37c2a74..e742621 100644 --- a/tests/test_compose_contract.py +++ b/tests/test_compose_contract.py @@ -15,6 +15,16 @@ ROOT = Path(__file__).resolve().parents[1] COMPOSE_FILE = ROOT / "compose.yaml" DOCKERFILE = ROOT / "Dockerfile" COMPOSE_BIN = os.environ.get("COMPOSE_BIN", "/opt/data/tmp/docker-compose-birweb") +SYNTHETIC_RESOURCE_UUID = "contract-test-resource" + +ROUTER_PRIORITY_LABELS = { + f"traefik.http.routers.http-0-{SYNTHETIC_RESOURCE_UUID}-site.priority": "500", + f"traefik.http.routers.https-0-{SYNTHETIC_RESOURCE_UUID}-site.priority": "500", +} +SOURCE_ROUTER_PRIORITY_TEMPLATES = { + "traefik.http.routers.http-0-${UUID}-site.priority=500", + "traefik.http.routers.https-0-${UUID}-site.priority=500", +} DOCKERFILE_SHA256 = "748bd362b43475779d241074a1955ae880967c923a7f68c693cb1c67f8b242dd" BASE_IMAGE = ( @@ -25,10 +35,13 @@ HEALTH_COMMAND = "wget -qO- http://127.0.0.1:8080/health.json || exit 1" TMPFS_TARGETS = {"/tmp", "/var/cache/nginx", "/var/run"} -def run(command: list[str]) -> subprocess.CompletedProcess[str]: +def run( + command: list[str], env: dict[str, str] | None = None +) -> subprocess.CompletedProcess[str]: return subprocess.run( command, cwd=ROOT, + env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, @@ -80,9 +93,12 @@ class ComposeContract(unittest.TestCase): ) if not COMPOSE_FILE.is_file(): raise AssertionError(f"RED: required Compose file is missing: {COMPOSE_FILE}") - rendered = run( - [COMPOSE_BIN, "-f", str(COMPOSE_FILE), "config", "--format", "json"] - ) + cls.render_command = [ + COMPOSE_BIN, "-f", str(COMPOSE_FILE), "config", "--format", "json" + ] + cls.render_env = os.environ.copy() + cls.render_env["COOLIFY_RESOURCE_UUID"] = SYNTHETIC_RESOURCE_UUID + rendered = run(cls.render_command, env=cls.render_env) if rendered.returncode != 0: raise AssertionError(f"Compose config rendering failed:\n{rendered.stderr}") try: @@ -160,6 +176,59 @@ class ComposeContract(unittest.TestCase): self.assertGreaterEqual(memory, 16 * 1024**2) self.assertLessEqual(memory, 512 * 1024**2) + def test_rendered_traefik_router_priorities_are_exact(self) -> None: + rendered_labels = self.service().get("labels") + self.assertIsInstance(rendered_labels, dict) + labels = rendered_labels if isinstance(rendered_labels, dict) else {} + self.assertEqual(labels, ROUTER_PRIORITY_LABELS) + self.assertTrue(all(isinstance(value, str) for value in labels.values())) + + def test_priority_labels_use_source_list_syntax_and_uuid_interpolation(self) -> None: + source_label_items = { + match.group(1) + for match in re.finditer( + r'^\s*-\s*["\']?(traefik\.(?:http\.)?[^"\']+?)["\']?\s*$', + self.source, + re.MULTILINE, + ) + } + required_interpolation = re.compile( + r"\$\{COOLIFY_RESOURCE_UUID:\?[^}\r\n]+\}" + ) + normalized_items = { + required_interpolation.sub("${UUID}", item) + for item in source_label_items + } + self.assertEqual(normalized_items, SOURCE_ROUTER_PRIORITY_TEMPLATES) + self.assertTrue( + all(required_interpolation.search(item) for item in source_label_items), + "each Traefik label must use required COOLIFY_RESOURCE_UUID interpolation", + ) + self.assertNotIn(SYNTHETIC_RESOURCE_UUID, self.source) + + mapping_labels = [ + (key, line) + for _, key, line in source_mapping_keys(self.source) + if key.startswith("traefik.") + ] + self.assertEqual(mapping_labels, [], "Traefik labels must use Compose list syntax") + + def test_resource_uuid_interpolation_fails_closed_when_missing_or_empty(self) -> None: + missing_env = os.environ.copy() + missing_env.pop("COOLIFY_RESOURCE_UUID", None) + empty_env = os.environ.copy() + empty_env["COOLIFY_RESOURCE_UUID"] = "" + + for case, env in (("missing", missing_env), ("empty", empty_env)): + with self.subTest(case=case): + rendered = run(self.render_command, env=env) + self.assertNotEqual( + rendered.returncode, + 0, + f"Compose config accepted a {case} COOLIFY_RESOURCE_UUID", + ) + self.assertIn("COOLIFY_RESOURCE_UUID", rendered.stderr) + def test_no_secrets_bind_mounts_or_forbidden_source_keys(self) -> None: service = self.service() self.assertNotIn("secrets", self.config)