import contextlib import io import json import os import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import deploy import utils def write_config(directory, value): (Path(directory) / "site.yaml").write_text(yaml.safe_dump(value), encoding="utf-8") def v2_config(): return yaml.safe_load((ROOT / "tests" / "fixtures" / "v2-site.yaml").read_text()) class ConfigTests(unittest.TestCase): def parse(self, config): with tempfile.TemporaryDirectory() as directory: write_config(directory, config) return utils.parse_site_yaml(directory, "baseline.fritzlab.net") def reject(self, config, message): with self.assertRaisesRegex(SystemExit, "1"): with contextlib.redirect_stderr(io.StringIO()) as stderr: self.parse(config) self.assertIn(message, stderr.getvalue()) def test_legacy_config_uses_explicit_compatibility_translation(self): config = self.parse({"domain": "example.fritzlab.net", "type": "static", "middlewares": ["authentik-forwardauth"]}) self.assertEqual("single-surface-v1", config["schema"]) self.assertTrue(config["compatibility"]) self.assertEqual("baseline.fritzlab.net", config["artifacts"]["site"]["bucket"]) self.assertEqual("/", config["routes"][0]["path"]) def test_v2_sorts_longest_prefix_and_assigns_cache_contracts(self): config = self.parse(v2_config()) self.assertEqual(["/dist/channels", "/dist/releases", "/"], [route["path"] for route in config["routes"]]) self.assertEqual("public, max-age=31536000, immutable", config["artifacts"]["releases"]["cache_control"]) self.assertEqual("public, max-age=0, must-revalidate", config["artifacts"]["channels"]["cache_control"]) def test_rejects_partial_v2(self): self.reject({"domain": "x.example", "schema": "v2", "routes": []}, "declared together") def test_rejects_public_catch_all(self): config = v2_config() config["routes"][0] = {"path": "/", "artifact": "portal", "access": "public"} config["artifacts"]["portal"]["cache"] = "revalidated-channel" self.reject(config, "public catch-all") def test_rejects_duplicate_paths(self): config = v2_config() config["routes"][1]["path"] = "/dist/releases" self.reject(config, "duplicate route path") def test_rejects_bucket_reuse_across_access(self): config = v2_config() config["artifacts"]["portal"]["bucket"] = "baseline-releases" self.reject(config, "bucket is reused across access classes") def test_rejects_credential_reuse_across_artifacts(self): config = v2_config() config["artifacts"]["portal"]["credentials"] = { "access_key_env": "RELEASES_ACCESS_KEY", "secret_key_env": "RELEASES_SECRET_KEY"} self.reject(config, "publication credentials are reused") def test_requires_explicit_root_policy(self): config = v2_config() config["routes"] = [route for route in config["routes"] if route["path"] != "/"] del config["artifacts"]["portal"] self.reject(config, "explicit / access policy") def test_rejects_overlapping_sources_across_access(self): config = v2_config() config["artifacts"]["portal"]["source"] = "." self.reject(config, "sources overlap across access classes") def test_rejects_unknown_top_level_key(self): config = v2_config() config["surprise"] = True self.reject(config, "unknown site.yaml keys") class ManifestTests(unittest.TestCase): def test_split_routes_render_one_certificate_and_distinct_backends(self): with tempfile.TemporaryDirectory() as config_dir, tempfile.TemporaryDirectory() as out: write_config(config_dir, v2_config()) config = utils.parse_site_yaml(config_dir, "baseline.fritzlab.net") app_dir = Path(out) / "app" deploy.render_site_manifests("baseline.fritzlab.net", ROOT, app_dir, app_dir / "manifests", config) manifests = app_dir / "manifests" self.assertEqual(1, len(list(manifests.glob("certificate*.yaml")))) self.assertEqual(3, len(list(manifests.glob("service-*.yaml")))) self.assertEqual(3, len(list(manifests.glob("ingress-*.yaml")))) for path in manifests.glob("*.yaml"): list(yaml.safe_load_all(path.read_text(encoding="utf-8"))) services = "".join(path.read_text() for path in manifests.glob("service-*.yaml")) self.assertIn("baseline-releases.web.sjc001.fritzlab.net", services) self.assertIn('service.passhostheader: "false"', services) portal_ingress = next(manifests.glob("ingress-*-portal.yaml")).read_text() self.assertIn("authentik-forwardauth@file", portal_ingress) class PublicationTests(unittest.TestCase): def artifact(self, immutable=False): return { "name": "release", "bucket": "release-bucket", "credentials": {"access_key_env": "TEST_ACCESS", "secret_key_env": "TEST_SECRET"}, "cache_control": "public, max-age=31536000, immutable" if immutable else "public, max-age=0, must-revalidate", "immutable": immutable, } @mock.patch.dict(os.environ, {"TEST_ACCESS": "access", "TEST_SECRET": "secret"}) @mock.patch("deploy.urlopen") def test_immutable_put_is_conditional_and_signed_in_process(self, urlopen_mock): response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = b"" with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "bundle.js" path.write_bytes(b"release") deploy._conditional_put("http://garage", self.artifact(True), "dist/releases/bundle.js", path, "application/javascript", "digest") request = urlopen_mock.call_args.args[0] self.assertEqual("*", request.headers["If-none-match"]) self.assertNotIn("secret", request.full_url) @mock.patch.dict(os.environ, {"TEST_ACCESS": "access", "TEST_SECRET": "secret"}) @mock.patch("deploy.run") def test_revalidated_publication_sets_cache_on_sync_and_restamp(self, run_mock): with tempfile.TemporaryDirectory() as directory: deploy._replaceable_sync(Path(directory), self.artifact(), "http://garage", []) commands = [call.args[0] for call in run_mock.call_args_list] self.assertEqual(["sync", "cp"], [command[4] for command in commands]) self.assertTrue(all("public, max-age=0, must-revalidate" in command for command in commands)) @mock.patch.dict(os.environ, {"TEST_ACCESS": "access", "TEST_SECRET": "secret"}) @mock.patch("deploy._conditional_put") @mock.patch("deploy.subprocess.run") def test_immutable_identical_retry_skips_and_changed_key_fails(self, subprocess_mock, put_mock): with tempfile.TemporaryDirectory() as directory: source = Path(directory) content = b"release" (source / "bundle.js").write_bytes(content) digest = __import__("hashlib").sha256(content).hexdigest() subprocess_mock.return_value = subprocess.CompletedProcess( [], 0, stdout=json.dumps({ "Metadata": {"sha256": digest}, "CacheControl": "public, max-age=31536000, immutable"}), stderr="") deploy._immutable_sync(source, self.artifact(True), "http://garage", []) put_mock.assert_not_called() subprocess_mock.return_value = subprocess.CompletedProcess( [], 0, stdout='{"Metadata":{"sha256":"different"}}', stderr="") with self.assertRaises(SystemExit): deploy._immutable_sync(source, self.artifact(True), "http://garage", []) class CredentialSafetyTests(unittest.TestCase): def test_git_askpass_reads_token_only_from_inherited_fd(self): token = "credential-sentinel" with utils.git_auth(token, "ci-bot") as auth: self.assertNotIn(token, str(auth["env"])) result = subprocess.run( [auth["env"]["GIT_ASKPASS"], "Password"], capture_output=True, text=True, check=True, **auth) self.assertEqual(token, result.stdout) if __name__ == "__main__": unittest.main()