import copy import io import os import sys import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from pathlib import Path from unittest.mock import patch import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import deploy import build import utils from utils import ConfigError, normalize_site_config def fixture(name): return yaml.safe_load((ROOT / "tests" / "fixtures" / name).read_text()) class ConfigContractTests(unittest.TestCase): def setUp(self): self.raw = fixture("split-site.yaml") def assert_invalid(self, mutate, message): raw = copy.deepcopy(self.raw) mutate(raw) with self.assertRaisesRegex(ConfigError, message): normalize_site_config(raw, "baseline.fritzlab.net") def test_legacy_normalizes_to_explicit_compatibility_surface(self): cfg = normalize_site_config(fixture("legacy-site.yaml"), "example.fritzlab.net") self.assertEqual("single-surface-v1", cfg["compatibility"]) self.assertEqual("build/html", cfg["artifacts"][0]["build_dir"]) self.assertEqual("example.fritzlab.net", cfg["artifacts"][0]["bucket"]) self.assertEqual("/", cfg["routes"][0]["path"]) self.assertEqual("legacy", cfg["routes"][0]["access"]) def test_routes_are_sorted_longest_prefix_first(self): cfg = normalize_site_config(self.raw, "baseline.fritzlab.net") self.assertEqual(["/dist", "/"], [route["path"] for route in cfg["routes"]]) credentials = {artifact["name"]: artifact["credentials"] for artifact in cfg["artifacts"]} self.assertNotEqual(credentials["distributions"], credentials["portal"]) def test_equivalent_route_paths_are_ambiguous(self): self.assert_invalid(lambda raw: raw["routes"].append({ "name": "duplicate", "path": "/dist/", "artifact": "portal", "access": {"mode": "protected", "middleware": "authentik-forwardauth"}, }), "ambiguous") def test_catch_all_is_required(self): self.assert_invalid(lambda raw: raw["routes"].__setitem__(0, { **raw["routes"][0], "path": "/portal" }), "catch-all") def test_public_catch_all_is_rejected_when_any_route_is_protected(self): def mutate(raw): raw["routes"][0]["access"] = {"mode": "public"} raw["routes"][1]["access"] = { "mode": "protected", "middleware": "authentik-forwardauth" } raw["artifacts"][0]["cache"]["rules"][0]["cache_control"] = ( "public, max-age=0, must-revalidate" ) for rule in raw["artifacts"][1]["cache"]["rules"]: rule["cache_control"] = "private, no-store" self.assert_invalid(mutate, "public '/' catch-all") def test_protected_route_requires_middleware(self): self.assert_invalid( lambda raw: raw["routes"][0].__setitem__("access", {"mode": "protected"}), "middleware is required", ) def test_bucket_cannot_cross_access_boundary(self): self.assert_invalid( lambda raw: raw["artifacts"][0]["publish"].__setitem__("bucket", "baseline-dist"), "protected and public", ) def test_publication_credentials_cannot_be_reused(self): self.assert_invalid( lambda raw: raw["artifacts"][0]["publish"]["credentials"].__setitem__( "access_key_env", "DIST_S3_ACCESS_KEY" ), "publication credential DIST_S3_ACCESS_KEY is reused", ) def test_cache_directive_contradiction_is_rejected(self): self.assert_invalid( lambda raw: raw["artifacts"][1]["cache"]["rules"][1].__setitem__( "cache_control", "public, max-age=31536000, immutable, must-revalidate" ), "immutable requires", ) def test_public_cache_is_rejected_on_protected_route(self): self.assert_invalid( lambda raw: raw["artifacts"][0]["cache"]["rules"][0].__setitem__( "cache_control", "public, max-age=0, must-revalidate" ), "protected route portal", ) def test_legacy_and_split_fields_cannot_mix(self): self.assert_invalid(lambda raw: raw.__setitem__("type", "static"), "cannot be mixed") def test_unknown_split_field_is_rejected(self): self.assert_invalid( lambda raw: raw["artifacts"][0].__setitem__("storage_bucket", "typo"), "unknown fields: storage_bucket", ) class GenerationTests(unittest.TestCase): def render(self, raw): cfg = normalize_site_config(raw, "baseline.fritzlab.net") tmp = tempfile.TemporaryDirectory() root = Path(tmp.name) app_dir = root / "app" manifests = app_dir / "manifests" app_dir.mkdir(parents=True) deploy.render_site_manifests( "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg ) files = {path.relative_to(app_dir).as_posix(): path.read_text() for path in sorted(app_dir.rglob("*.yaml"))} return tmp, app_dir, files def test_split_fixture_generates_per_route_resources_and_one_certificate(self): tmp, _, files = self.render(fixture("split-site.yaml")) self.addCleanup(tmp.cleanup) self.assertEqual({ "app.yaml", "manifests/certificate.yaml", "manifests/ingress-distributions.yaml", "manifests/ingress-portal.yaml", "manifests/kustomization.yaml", "manifests/service-distributions.yaml", "manifests/service-portal.yaml", }, set(files)) for content in files.values(): self.assertIsNotNone(yaml.safe_load(content)) self.assertIn("path: /dist", files["manifests/ingress-distributions.yaml"]) self.assertNotIn("authentik-forwardauth", files["manifests/ingress-distributions.yaml"]) self.assertIn("authentik-forwardauth@file", files["manifests/ingress-portal.yaml"]) self.assertIn('service.passhostheader: "false"', files["manifests/service-portal.yaml"]) self.assertNotIn("passhostheader", files["manifests/ingress-portal.yaml"]) self.assertIn("baseline-dist.web.sjc001.fritzlab.net", files["manifests/service-distributions.yaml"]) def test_generation_is_deterministic_when_input_lists_are_reversed(self): raw = fixture("split-site.yaml") first_tmp, _, first = self.render(raw) self.addCleanup(first_tmp.cleanup) raw["artifacts"].reverse() raw["routes"].reverse() second_tmp, _, second = self.render(raw) self.addCleanup(second_tmp.cleanup) self.assertEqual(first, second) def test_stale_route_manifests_are_removed(self): raw = fixture("split-site.yaml") cfg = normalize_site_config(raw, "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: app_dir = Path(tmp) / "app" manifests = app_dir / "manifests" manifests.mkdir(parents=True) stale = manifests / "ingress-removed.yaml" stale.write_text("stale\n") deploy.render_site_manifests("baseline.fritzlab.net", ROOT, app_dir, manifests, cfg) self.assertFalse(stale.exists()) def test_legacy_names_and_garage_s3_target_are_preserved(self): tmp, _, files = self.render(fixture("legacy-site.yaml")) self.addCleanup(tmp.cleanup) self.assertIn("manifests/service.yaml", files) self.assertIn("manifests/ingress.yaml", files) self.assertIn("garage-s3.storage.svc.k8s.sjc001.fritzlab.net", files["manifests/service.yaml"]) self.assertNotIn("passhostheader", files["manifests/ingress.yaml"]) class BuildTests(unittest.TestCase): def test_artifacts_build_independently_without_clobbering_siblings(self): raw = fixture("split-site.yaml") for artifact in raw["artifacts"]: artifact["tidy"] = False cfg = normalize_site_config(raw, "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "dist").mkdir() (root / "dist" / "bundle.js").write_text("bundle") (root / "portal" / "build").mkdir(parents=True) (root / "portal" / "build" / "index.html").write_text("portal") for artifact in cfg["artifacts"]: build.build_artifact(root, artifact) self.assertEqual( "bundle", (root / ".site-publish/distributions/html/bundle.js").read_text() ) self.assertEqual( "portal", (root / ".site-publish/portal/html/index.html").read_text() ) class PublishingTests(unittest.TestCase): def test_apps_clone_never_places_token_in_argv_or_log(self): calls = [] secret = "clone-secret-must-not-appear" with patch.dict(os.environ, {"CI_BOT_USER": "ci-bot"}, clear=False), \ patch.object(utils, "run", side_effect=lambda command, **kwargs: calls.append((command, kwargs))), \ patch.object(utils.shutil, "rmtree"), redirect_stdout(io.StringIO()) as output: utils.clone_apps(secret) self.assertNotIn(secret, output.getvalue()) for command, kwargs in calls: self.assertNotIn(secret, " ".join(command)) self.assertNotIn(secret, kwargs.get("display", "")) def test_cache_headers_credentials_and_route_prefix_are_separate(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") route = next(item for item in cfg["routes"] if item["artifact"] == "distributions") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) html = root / artifact["build_dir"] (html / "releases").mkdir(parents=True) (html / "channels").mkdir() (html / "releases" / "1.0.js").write_text("release") (html / "channels" / "stable.json").write_text("channel") commands = [] def capture(command, **kwargs): commands.append((command, kwargs["env"])) secret = "secret-must-not-appear" output = io.StringIO() with patch.dict(os.environ, { "DIST_S3_ACCESS_KEY": "dist-key", "DIST_S3_SECRET_KEY": secret }, clear=False), patch.object(deploy, "run", side_effect=capture), redirect_stdout(output): deploy.s3_sync(artifact, route, root) self.assertTrue(all(secret not in " ".join(command) for command, _ in commands)) self.assertNotIn(secret, output.getvalue()) self.assertTrue(all(call_env["AWS_ACCESS_KEY_ID"] == "dist-key" for _, call_env in commands)) self.assertTrue(all("DIST_S3_SECRET_KEY" not in call_env for _, call_env in commands)) rendered = [" ".join(command) for command, _ in commands] self.assertIn("s3://baseline-dist/", rendered[0]) self.assertTrue(all("s3://baseline-dist/dist/" in command for command in rendered[1:])) self.assertTrue(any("releases/" in command and "public, max-age=31536000, immutable" in command for command in rendered)) self.assertTrue(any("channels/" in command and "public, max-age=0, must-revalidate" in command for command in rendered)) def test_absent_artifact_is_detected_before_publish(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp, redirect_stderr(io.StringIO()), \ self.assertRaises(SystemExit): deploy.validate_artifact_output(Path(tmp), cfg["artifacts"][0]) def test_absent_cache_prefix_is_detected(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") with tempfile.TemporaryDirectory() as tmp: html = Path(tmp) / artifact["build_dir"] html.mkdir(parents=True) (html / "index.html").write_text("content") with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): deploy.validate_artifact_output(Path(tmp), artifact) if __name__ == "__main__": unittest.main()