import os import sys import tempfile import textwrap import unittest from pathlib import Path from unittest.mock import patch import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) from build import stage_artifacts from deploy import credential_environment, render_site_manifests, routed_cache, s3_sync from utils import k8s_name, parse_site_yaml MULTI_SITE = """ domain: baseline.fritzlab.net artifacts: catalogue: source: apps/catalogue/build bucket: baseline-catalogue credential: catalogue cache: default: private, no-store dist: source: dist bucket: baseline-dist credential: dist cache: default: public, max-age=0, must-revalidate, no-transform rules: - match: releases/* value: public, max-age=31536000, immutable, no-transform cors_origins: ["*"] routes: - name: catalogue path: / artifact: catalogue access: protected - name: dist path: /dist artifact: dist access: public """ class SiteConfigTests(unittest.TestCase): def write_config(self, directory, body): Path(directory, "site.yaml").write_text(textwrap.dedent(body), encoding="utf-8") def test_multi_artifact_config_separates_storage_and_routes(self): with tempfile.TemporaryDirectory() as tmp: self.write_config(tmp, MULTI_SITE) config = parse_site_yaml(tmp) self.assertEqual(config["mode"], "multi") self.assertEqual(config["artifacts"]["dist"]["credential"], "dist") self.assertEqual(config["artifacts"]["dist"]["cache"]["rules"][0]["match"], "releases/*") self.assertEqual(config["routes"][1]["path"], "/dist") def test_legacy_config_remains_supported(self): with tempfile.TemporaryDirectory() as tmp: self.write_config(tmp, """ domain: example.fritzlab.net type: static content_dir: html middlewares: [authentik-forwardauth] """) config = parse_site_yaml(tmp) self.assertEqual(config["mode"], "legacy") self.assertEqual(config["content_dir"], "html") def test_duplicate_bucket_fails_before_sync_delete_can_run(self): with tempfile.TemporaryDirectory() as tmp: self.write_config(tmp, """ domain: example.fritzlab.net artifacts: first: {source: out/first, bucket: shared-bucket, cache: {default: "private, no-store"}} second: {source: out/second, bucket: shared-bucket} routes: - {name: first, path: /, artifact: first, access: protected} - {name: second, path: /second, artifact: second, access: public} """) with self.assertRaises(SystemExit): parse_site_yaml(tmp) def test_unknown_artifact_and_root_strip_fail_closed(self): cases = ( "{name: root, path: /, artifact: missing, access: public}", "{name: root, path: relative, artifact: site, access: public}", ) for routes in cases: with self.subTest(routes=routes), tempfile.TemporaryDirectory() as tmp: self.write_config(tmp, "\n".join(( "domain: example.fritzlab.net", "artifacts:", " site: {source: out/site, bucket: example-site}", "routes:", f" - {routes}", ))) with self.assertRaises(SystemExit): parse_site_yaml(tmp) def test_artifact_build_snapshots_each_output(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) self.write_config(root, MULTI_SITE) (root / "apps/catalogue/build").mkdir(parents=True) (root / "apps/catalogue/build/index.html").write_text("catalogue", encoding="utf-8") (root / "dist").mkdir() (root / "dist/baseline.css").write_text("tokens", encoding="utf-8") config = parse_site_yaml(root) stage_artifacts(root, config) self.assertEqual((root / ".site-publish/catalogue/index.html").read_text(), "catalogue") self.assertEqual((root / ".site-publish/dist/dist/baseline.css").read_text(), "tokens") def test_rendered_routes_have_independent_middleware_chains(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) self.write_config(root, MULTI_SITE) config = parse_site_yaml(root) app_dir = root / "app" manifests = app_dir / "manifests" app_dir.mkdir() render_site_manifests("baseline", ROOT, app_dir, manifests, config) ingress = (manifests / "ingress.yaml").read_text(encoding="utf-8") service = (manifests / "service.yaml").read_text(encoding="utf-8") self.assertIn("authentik-forwardauth@file", ingress) self.assertNotIn("kubernetescrd", ingress) self.assertIn("baseline-dist.web.sjc001.fritzlab.net", service) self.assertIn("service.passhostheader: \"false\"", service) for manifest in manifests.glob("*.yaml"): documents = list(yaml.safe_load_all(manifest.read_text(encoding="utf-8"))) self.assertTrue(documents) self.assertNotIn(None, documents, manifest.name) def test_legacy_render_does_not_emit_crd_middleware(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) self.write_config(root, """ domain: example.fritzlab.net type: static content_dir: html """) config = parse_site_yaml(root) app_dir = root / "app" manifests = app_dir / "manifests" app_dir.mkdir() render_site_manifests("example.fritzlab.net", ROOT, app_dir, manifests, config) self.assertFalse((manifests / "middleware.yaml").exists()) self.assertIn("name: example-fritzlab-net", (manifests / "ingress.yaml").read_text()) def test_protected_artifact_rejects_shared_cache_and_wildcard_cors(self): invalid_configs = ( """ domain: example.fritzlab.net artifacts: site: source: out/site bucket: example-site cache: {default: "public, max-age=0"} routes: - {name: site, path: /, artifact: site, access: protected} """, """ domain: example.fritzlab.net artifacts: site: source: out/site bucket: example-site cache: {default: "private, no-store"} cors_origins: ["*"] routes: - {name: site, path: /, artifact: site, access: protected} """, ) for body in invalid_configs: with self.subTest(body=body), tempfile.TemporaryDirectory() as tmp: self.write_config(tmp, body) with self.assertRaises(SystemExit): parse_site_yaml(tmp) def test_credential_profile_cannot_cross_access_boundaries(self): with tempfile.TemporaryDirectory() as tmp: self.write_config(tmp, """ domain: example.fritzlab.net artifacts: private: source: out/private bucket: example-private credential: shared cache: {default: "private, no-store"} public: source: out/public bucket: example-public credential: shared routes: - {name: private, path: /, artifact: private, access: protected} - {name: public, path: /public, artifact: public, access: public} """) with self.assertRaises(SystemExit): parse_site_yaml(tmp) def test_unknown_modern_field_fails_closed(self): with tempfile.TemporaryDirectory() as tmp: self.write_config(tmp, """ domain: example.fritzlab.net artifacts: site: source: out/site storage_bucket: misspelled bucket: example-site routes: - {name: site, path: /, artifact: site, access: public} """) with self.assertRaises(SystemExit): parse_site_yaml(tmp) def test_empty_artifact_fails_before_staging(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) self.write_config(root, MULTI_SITE) (root / "apps/catalogue/build").mkdir(parents=True) (root / "dist").mkdir() config = parse_site_yaml(root) with self.assertRaises(SystemExit): stage_artifacts(root, config) def test_cache_rules_are_applied_after_default_metadata(self): with tempfile.TemporaryDirectory() as tmp: source = Path(tmp) (source / "release.css").write_text("css", encoding="utf-8") cache = { "default": "public, max-age=0, must-revalidate", "rules": [{ "match": "releases/*", "value": "public, max-age=31536000, immutable", }], } with patch("deploy.credential_environment", return_value={}), \ patch("deploy.configure_cors"), patch("deploy.run_args") as run_args: s3_sync("example-public", source, cache=cache) commands = [call.args[0] for call in run_args.call_args_list] self.assertEqual(len(commands), 3) self.assertIn("--include", commands[2]) self.assertIn("releases/*", commands[2]) def test_cache_patterns_follow_the_public_route_prefix(self): cache = { "default": "public, max-age=0, must-revalidate", "rules": [{"match": "releases/*", "value": "public, immutable"}], } routed = routed_cache(cache, [{"path": "/dist"}]) self.assertEqual(routed["rules"][0]["match"], "dist/releases/*") def test_named_credentials_are_selected_without_changing_parent_environment(self): with patch.dict(os.environ, { "SITE_PUBLISH_DIST_S3_ACCESS_KEY_ID": "access", "SITE_PUBLISH_DIST_S3_SECRET_ACCESS_KEY": "secret", }, clear=False): child = credential_environment("dist") self.assertEqual(child["AWS_ACCESS_KEY_ID"], "access") self.assertEqual(child["AWS_SECRET_ACCESS_KEY"], "secret") def test_long_resource_names_are_stable_dns_labels(self): first = k8s_name("a" * 90) second = k8s_name("a" * 89 + "b") self.assertLessEqual(len(first), 63) self.assertNotEqual(first, second) if __name__ == "__main__": unittest.main()