test: require portable Traefik router priorities

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(
+        cls.render_command = [
-            [COMPOSE_BIN, "-f", str(COMPOSE_FILE), "config", "--format", "json"]
+            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)