site-publish/tests/test_contract.py

import base64
import copy
import hashlib
import io
import json
import os
import socket
import subprocess
import sys
import tempfile
import threading
import unittest
from contextlib import redirect_stderr, redirect_stdout
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit
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,
    parse_site_yaml,
    validate_artifact_inputs,
)


def fixture(name):
    return yaml.safe_load((ROOT / "tests" / "fixtures" / name).read_text())


class IPv6GitHTTPServer(ThreadingHTTPServer):
    address_family = socket.AF_INET6


class AuthenticatedGitHandler(BaseHTTPRequestHandler):
    project_root = None
    expected_authorization = None

    def log_message(self, _format, *_args):
        pass

    def do_GET(self):
        self._git_backend()

    def do_POST(self):
        self._git_backend()

    def _git_backend(self):
        if self.headers.get("Authorization") != self.expected_authorization:
            self.send_response(401)
            self.send_header("WWW-Authenticate", 'Basic realm="test"')
            self.end_headers()
            return
        parsed = urlsplit(self.path)
        length = int(self.headers.get("Content-Length", "0"))
        request_body = self.rfile.read(length) if length else b""
        backend_env = os.environ.copy()
        backend_env.update({
            "GIT_PROJECT_ROOT": str(self.project_root),
            "GIT_HTTP_EXPORT_ALL": "1",
            "PATH_INFO": parsed.path,
            "QUERY_STRING": parsed.query,
            "REQUEST_METHOD": self.command,
            "CONTENT_TYPE": self.headers.get("Content-Type", ""),
            "CONTENT_LENGTH": str(length),
            "REMOTE_USER": "ci-bot",
            "REMOTE_ADDR": "::1",
            "GATEWAY_INTERFACE": "CGI/1.1",
            "SERVER_PROTOCOL": "HTTP/1.1",
        })
        result = subprocess.run(
            ["git", "http-backend"], input=request_body, env=backend_env,
            capture_output=True, check=True,
        )
        raw_headers, response_body = result.stdout.split(b"\r\n\r\n", 1)
        headers, status = [], 200
        for line in raw_headers.decode().split("\r\n"):
            name, value = line.split(":", 1)
            if name.lower() == "status":
                status = int(value.strip().split(" ", 1)[0])
            else:
                headers.append((name, value.strip()))
        self.send_response(status)
        for name, value in headers:
            self.send_header(name, value)
        self.end_headers()
        self.wfile.write(response_body)


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"])
        distributions = next(
            artifact for artifact in cfg["artifacts"] if artifact["name"] == "distributions"
        )
        self.assertEqual(["*"], distributions["cors_origins"])

    def test_cors_origins_are_https_origins_or_wildcard(self):
        self.assert_invalid(
            lambda raw: raw["artifacts"][1].__setitem__(
                "cors_origins", ["http://consumer.example"]
            ),
            "must contain '\\*' or canonical HTTPS origins",
        )
        for origin in (
            "https://consumer example",
            "https://Consumer.example",
            "https://consumer.example:443",
            "https://consumer.example/",
            "https://consumer.example:invalid",
            "https://[fe80::1%eth0]",
            "https://[fe80::1%25eth0]",
        ):
            with self.subTest(origin=origin):
                self.assert_invalid(
                    lambda raw, origin=origin: raw["artifacts"][1].__setitem__(
                        "cors_origins", [origin]
                    ),
                    "must contain '\\*' or canonical HTTPS origins",
                )
        self.assert_invalid(
            lambda raw: raw["artifacts"][1].__setitem__(
                "cors_origins", ["https://consumer.example", "https://consumer.example:443"]
            ),
            "must not contain duplicate canonical origins",
        )
        self.assert_invalid(
            lambda raw: raw["artifacts"][1].__setitem__(
                "cors_origins", ["*", "https://consumer.example"]
            ),
            "wildcard must be the only origin",
        )
        raw = copy.deepcopy(self.raw)
        raw["artifacts"][1]["cors_origins"] = ["https://[2602:817:3000::1]:8443"]
        cfg = normalize_site_config(raw, "baseline.fritzlab.net")
        self.assertEqual(
            ["https://[2602:817:3000::1]:8443"],
            next(a for a in cfg["artifacts"] if a["name"] == "distributions")["cors_origins"],
        )

    def test_declared_cors_origins_must_be_a_list(self):
        for value in (None, False, 0, "", {}):
            with self.subTest(value=value):
                self.assert_invalid(
                    lambda raw, value=value: raw["artifacts"][1].__setitem__(
                        "cors_origins", value
                    ),
                    "cors_origins must be a list",
                )

    def test_protected_route_rejects_wildcard_cors(self):
        self.assert_invalid(
            lambda raw: raw["artifacts"][0].__setitem__("cors_origins", ["*"]),
            "protected route portal cannot allow wildcard CORS",
        )

    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"][1]["cors_origins"] = []
            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):
        def mutate(raw):
            raw["artifacts"][0]["publish"]["credentials"] = copy.deepcopy(
                raw["artifacts"][1]["publish"]["credentials"]
            )
        self.assert_invalid(mutate, "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_cache_policy_cannot_nest_below_immutable_path(self):
        def mutate(raw):
            raw["artifacts"][1]["cache"]["rules"].append({
                "path": "releases/candidates",
                "cache_control": "public, max-age=0, must-revalidate",
            })
        self.assert_invalid(mutate, "cannot nest another policy under immutable /releases")

    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",
        )

    def test_backend_authority_and_endpoint_are_derived(self):
        for field, value in (
            ("endpoint", "https://attacker.example"),
            ("website_authority", "internal-api.default.svc.k8s.sjc001.fritzlab.net"),
        ):
            with self.subTest(field=field):
                self.assert_invalid(
                    lambda raw, field=field, value=value: raw["artifacts"][0]["publish"].__setitem__(field, value),
                    f"unknown fields: {field}",
                )

    def test_publication_credentials_use_dedicated_matched_names(self):
        self.assert_invalid(
            lambda raw: raw["artifacts"][0]["publish"]["credentials"].__setitem__(
                "access_key_env", "CI_BOT_TOKEN"
            ),
            "must be a matched <NAME>_S3_ACCESS_KEY",
        )

    def test_resolved_build_inputs_must_be_pairwise_disjoint(self):
        raw = copy.deepcopy(self.raw)
        raw["artifacts"][1]["content_dir"] = ""
        cfg = normalize_site_config(raw, "baseline.fritzlab.net")
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "portal" / "build").mkdir(parents=True)
            with self.assertRaisesRegex(ConfigError, "build inputs overlap after resolution"):
                validate_artifact_inputs(root, cfg)

    def test_descendant_symlink_cannot_cross_artifact_boundary(self):
        cfg = normalize_site_config(self.raw, "baseline.fritzlab.net")
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "dist").mkdir()
            (root / "portal" / "build").mkdir(parents=True)
            (root / "portal" / "build" / "private.txt").write_text("private")
            (root / "dist" / "portal-link").symlink_to(root / "portal" / "build")
            with self.assertRaisesRegex(ConfigError, "build input contains symlink"):
                validate_artifact_inputs(root, cfg)

    def test_artifact_root_symlink_is_rejected_before_resolution(self):
        cfg = normalize_site_config(self.raw, "baseline.fritzlab.net")
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "dist-real").mkdir()
            (root / "dist").symlink_to(root / "dist-real")
            (root / "portal" / "build").mkdir(parents=True)
            with self.assertRaisesRegex(ConfigError, "content_dir contains symlink component"):
                validate_artifact_inputs(root, cfg)


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",
            "site-publish-history.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_yaml_ambiguous_artifact_name_stays_a_string_annotation(self):
        raw = fixture("split-site.yaml")
        raw["artifacts"][0]["name"] = "yes"
        raw["routes"][0]["artifact"] = "yes"
        tmp, _, files = self.render(raw)
        self.addCleanup(tmp.cleanup)
        ingress = yaml.safe_load(files["manifests/ingress-portal.yaml"])
        self.assertEqual(
            "yes", ingress["metadata"]["annotations"]["site-publish.fritzlab.net/artifact"],
        )

    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, app_dir, 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"])
        self.assertNotIn("site-publish.fritzlab.net", files["manifests/ingress.yaml"])
        self.assertEqual({}, deploy.previous_route_contracts(app_dir))


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 ArtifactSelectionTests(unittest.TestCase):
    """The `artifacts` input scopes publication, never the route contract."""

    def parse(self, selection):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "site.yaml").write_text(
                (ROOT / "tests" / "fixtures" / "split-site.yaml").read_text()
            )
            environment = {"SITE_REPO": "fritzlab/baseline"}
            if selection is not None:
                environment["SITE_ARTIFACTS"] = selection
            with patch.dict(os.environ, environment, clear=False), \
                    redirect_stdout(io.StringIO()):
                if selection is None:
                    os.environ.pop("SITE_ARTIFACTS", None)
                return parse_site_yaml(root)

    def test_absent_selection_publishes_every_declared_artifact(self):
        self.assertEqual(["distributions", "portal"], self.parse(None)["selected"])
        self.assertEqual(["distributions", "portal"], self.parse("  ")["selected"])

    def test_selection_accepts_comma_and_whitespace_separated_names(self):
        self.assertEqual(["portal"], self.parse("portal")["selected"])
        self.assertEqual(
            ["distributions", "portal"], self.parse("portal, distributions")["selected"]
        )
        self.assertEqual(
            ["distributions", "portal"], self.parse("portal distributions")["selected"]
        )

    def test_undeclared_selection_fails_before_any_publication(self):
        stderr = io.StringIO()
        with redirect_stderr(stderr), self.assertRaises(SystemExit):
            self.parse("catalogue")
        self.assertIn("undeclared artifact", stderr.getvalue())
        self.assertIn("catalogue", stderr.getvalue())

    def test_build_skips_unselected_artifacts(self):
        raw = fixture("split-site.yaml")
        for artifact in raw["artifacts"]:
            artifact["tidy"] = False
        cfg = normalize_site_config(raw, "baseline.fritzlab.net")
        cfg["selected"] = ["portal"]
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "portal" / "build").mkdir(parents=True)
            (root / "portal" / "build" / "index.html").write_text("portal")
            built = []
            with patch.object(build, "build_artifact",
                              side_effect=lambda _root, artifact: built.append(artifact["name"])), \
                    patch.dict(os.environ, {"SITE_DIR": str(root)}, clear=False), \
                    patch.object(build, "parse_site_yaml", return_value=cfg), \
                    redirect_stdout(io.StringIO()):
                build.cmd_build()
            self.assertEqual(["portal"], built)

    def deploy_selection(self, selection, environment):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        cfg["selected"] = selection
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        root = Path(tmp.name)
        apps_dir = root / "apps"
        app_dir = apps_dir / "sjc001" / "websites" / "baseline.fritzlab.net"
        app_dir.mkdir(parents=True)
        # A scoped run refuses an artifact it has no published history for, so
        # the fixture starts from the whole publication this one narrows.
        (app_dir / "site-publish-history.yaml").write_text(yaml.safe_dump({
            "version": 1,
            "buckets": {
                artifact["bucket"]: {
                    "path": route["path"],
                    "access": "public" if route["access"] == "legacy" else route["access"],
                    "artifact": route["artifact"],
                    "immutable_paths": deploy.immutable_key_prefixes(artifact, route),
                }
                for route in cfg["routes"]
                for artifact in [next(item for item in cfg["artifacts"]
                                      if item["name"] == route["artifact"])]
            },
        }, sort_keys=True))
        for artifact in cfg["artifacts"]:
            if artifact["name"] not in set(selection):
                continue
            html = root / artifact["build_dir"]
            html.mkdir(parents=True)
            (html / "index.html").write_text(artifact["name"])
            for rule in artifact["cache_rules"]:
                if rule["path"]:
                    (html / rule["path"]).mkdir(parents=True, exist_ok=True)
                    (html / rule["path"] / "keep.json").write_text("{}")
        synced, immutables, cors = [], [], []
        with patch.dict(os.environ, environment, clear=False), \
                patch.object(deploy, "clone_apps", return_value=apps_dir), \
                patch.object(deploy, "commit_and_push"), \
                patch.object(deploy, "publish_route_immutables",
                             side_effect=lambda artifact, *_a, **_k: immutables.append(artifact["name"])), \
                patch.object(deploy, "reconcile_artifact_cors",
                             side_effect=lambda artifacts, *_a: cors.extend(item["name"] for item in artifacts)), \
                patch.object(deploy, "s3_sync",
                             side_effect=lambda artifact, *_a, **_k: synced.append(artifact["name"])), \
                redirect_stdout(io.StringIO()):
            deploy.deploy_static("baseline.fritzlab.net", root, ROOT, "token", cfg)
        files = {path.relative_to(app_dir).as_posix(): path.read_text()
                 for path in sorted(app_dir.rglob("*.yaml"))}
        return {"synced": synced, "immutables": immutables, "cors": cors, "files": files}

    def test_selected_publication_leaves_the_whole_route_contract_rendered(self):
        # Only the portal credentials exist: a scoped run must not demand the
        # secrets of an artifact it is not publishing.
        environment = {"PORTAL_S3_ACCESS_KEY": "portal-key",
                       "PORTAL_S3_SECRET_KEY": "portal-secret"}
        for name in ("DIST_S3_ACCESS_KEY", "DIST_S3_SECRET_KEY"):
            os.environ.pop(name, None)
        result = self.deploy_selection(["portal"], environment)
        self.assertEqual(["portal"], result["synced"])
        self.assertEqual(["portal"], result["immutables"])
        self.assertEqual(["portal"], result["cors"])
        self.assertIn("manifests/ingress-distributions.yaml", result["files"])
        self.assertIn("manifests/service-distributions.yaml", result["files"])
        history = yaml.safe_load(result["files"]["site-publish-history.yaml"])
        self.assertEqual({"baseline-dist", "baseline-portal"}, set(history["buckets"]))
        self.assertEqual(
            ["dist/releases"], history["buckets"]["baseline-dist"]["immutable_paths"]
        )

    def test_scoped_run_refuses_to_record_a_contract_it_did_not_publish(self):
        """The reviewer's scenario: distributions goes protected in site.yaml while
        a catalogue-only run publishes. Recording that access would make the
        bucket unreturnable through validate_route_migrations."""
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        published = {
            "baseline-portal": {
                "path": "/", "access": "protected", "artifact": "portal", "immutable_paths": [],
            },
            "baseline-dist": {
                "path": "/dist", "access": "public", "artifact": "distributions",
                "immutable_paths": ["dist/releases"],
            },
        }
        # A whole publish agrees with what is recorded, scoped or not.
        deploy.validate_scoped_history(cfg, published)
        cfg["selected"] = ["portal"]
        deploy.validate_scoped_history(cfg, published)

        moved = copy.deepcopy(cfg)
        route = next(item for item in moved["routes"] if item["artifact"] == "distributions")
        route["access"] = "protected"
        route["access_middleware"] = "authentik-forwardauth"
        with self.assertRaises(RuntimeError) as refused:
            deploy.validate_scoped_history(moved, published)
        self.assertIn("may not change unselected artifact distributions", str(refused.exception))
        # The same change published in the same run is allowed to proceed.
        moved["selected"] = ["distributions", "portal"]
        deploy.validate_scoped_history(moved, published)

    def test_scoped_run_refuses_an_artifact_with_no_published_history(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        cfg["selected"] = ["portal"]
        with self.assertRaises(RuntimeError) as refused:
            deploy.validate_scoped_history(cfg, {})
        self.assertIn("cannot introduce artifact distributions", str(refused.exception))

    def test_scoped_deploy_runs_the_history_guard_before_touching_a_bucket(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        cfg["selected"] = ["portal"]
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        root = Path(tmp.name)
        apps_dir = root / "apps"
        app_dir = apps_dir / "sjc001" / "websites" / "baseline.fritzlab.net"
        app_dir.mkdir(parents=True)
        html = root / next(
            item for item in cfg["artifacts"] if item["name"] == "portal"
        )["build_dir"]
        html.mkdir(parents=True)
        (html / "index.html").write_text("portal")
        with patch.dict(os.environ, {
            "PORTAL_S3_ACCESS_KEY": "portal-key", "PORTAL_S3_SECRET_KEY": "portal-secret",
        }, clear=False), \
                patch.object(deploy, "clone_apps", return_value=apps_dir), \
                patch.object(deploy, "commit_and_push"), \
                patch.object(deploy, "publish_route_immutables") as immutables, \
                patch.object(deploy, "reconcile_artifact_cors") as cors, \
                patch.object(deploy, "s3_sync") as sync, \
                redirect_stdout(io.StringIO()), \
                self.assertRaises(RuntimeError):
            deploy.deploy_static("baseline.fritzlab.net", root, ROOT, "token", cfg)
        immutables.assert_not_called()
        cors.assert_not_called()
        sync.assert_not_called()

    def test_disabled_site_refuses_a_partial_selection(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        cfg["enabled"] = False
        cfg["selected"] = ["portal"]
        with patch.object(deploy, "parse_site_yaml", return_value=cfg), \
                patch.dict(os.environ, {
                    "SITE_REPO": "fritzlab/baseline.fritzlab.net", "SITE_DIR": ".",
                    "ACTION_DIR": str(ROOT), "CI_BOT_TOKEN": "token",
                }, clear=False), \
                patch.object(deploy, "decommission") as decommission, \
                redirect_stderr(io.StringIO()), redirect_stdout(io.StringIO()), \
                self.assertRaises(SystemExit):
            deploy.cmd_deploy()
        decommission.assert_not_called()


class PublishingTests(unittest.TestCase):
    def test_cors_policy_is_reconciled_as_read_only_browser_access(self):
        captured = {}

        def capture(command, **_kwargs):
            config_path = command[command.index("--cors-configuration") + 1]
            captured["command"] = command
            captured["config"] = json.loads(Path(config_path.removeprefix("file://")).read_text())

        with patch.object(deploy, "run", side_effect=capture):
            deploy.configure_cors(
                "baseline-dist", ["*"], "http://garage-s3.storage.svc:3900", {}
            )

        self.assertIn("put-bucket-cors", captured["command"])
        self.assertEqual(["GET", "HEAD"], captured["config"]["CORSRules"][0]["AllowedMethods"])
        self.assertEqual(["*"], captured["config"]["CORSRules"][0]["AllowedOrigins"])

    def test_empty_cors_policy_removes_stale_bucket_cors(self):
        with patch.object(deploy, "run") as request:
            deploy.configure_cors(
                "baseline-catalogue", [], "http://garage-s3.storage.svc:3900", {}
            )
        self.assertIn("delete-bucket-cors", request.call_args.args[0])

    def test_cors_reconciliation_restores_prior_policies_on_failure(self):
        artifacts = [
            {
                "bucket": "first", "cors_origins": ["https://new.example"],
                "s3_endpoint": "http://garage-s3.storage.svc:3900", "credentials": {},
            },
            {
                "bucket": "second", "cors_origins": [],
                "s3_endpoint": "http://garage-s3.storage.svc:3900", "credentials": {},
            },
        ]
        prior = [
            {"CORSRules": [{"AllowedOrigins": ["https://old.example"]}]},
            None,
        ]
        writes = []

        def write(bucket, config, *_args):
            writes.append((bucket, config))
            if bucket == "second" and len(writes) == 2:
                raise RuntimeError("write failed")

        with patch.object(deploy, "publication_aws_env", return_value={}), patch.object(
            deploy, "get_cors_configuration", side_effect=prior,
        ) as read, patch.object(deploy, "set_cors_configuration", side_effect=write), \
                self.assertRaisesRegex(RuntimeError, "write failed"):
            deploy.reconcile_artifact_cors(artifacts)

        self.assertEqual(2, read.call_count)
        self.assertEqual(
            [
                ("first", {"CORSRules": [{
                    "AllowedOrigins": ["https://new.example"],
                    "AllowedMethods": ["GET", "HEAD"],
                    "AllowedHeaders": ["*"],
                    "ExposeHeaders": ["ETag"],
                    "MaxAgeSeconds": 3600,
                }]}),
                ("second", None),
                ("second", None),
                ("first", prior[0]),
            ],
            writes,
        )

    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_askpass_authenticates_clone_and_push_round_trip(self):
        token = "round-trip-token"
        expected = "Basic " + base64.b64encode(f"ci-bot:{token}".encode()).decode()
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            bare = root / "repo.git"
            seed = root / "seed"
            checkout = root / "checkout"
            subprocess.run(["git", "init", "--bare", "--initial-branch=main", str(bare)], check=True,
                           stdout=subprocess.DEVNULL)
            subprocess.run(["git", "-C", str(bare), "config", "http.receivepack", "true"], check=True)
            subprocess.run(["git", "init", "--initial-branch=main", str(seed)], check=True,
                           stdout=subprocess.DEVNULL)
            subprocess.run(["git", "-C", str(seed), "config", "user.name", "Test"], check=True)
            subprocess.run(["git", "-C", str(seed), "config", "user.email", "test@example.invalid"], check=True)
            (seed / "README.md").write_text("seed\n")
            subprocess.run(["git", "-C", str(seed), "add", "README.md"], check=True)
            subprocess.run(["git", "-C", str(seed), "commit", "-m", "seed"], check=True,
                           stdout=subprocess.DEVNULL)
            subprocess.run(["git", "-C", str(seed), "push", str(bare), "main"], check=True,
                           stdout=subprocess.DEVNULL)

            handler = type("GitHandler", (AuthenticatedGitHandler,), {
                "project_root": root, "expected_authorization": expected,
            })
            server = IPv6GitHTTPServer(("::1", 0), handler)
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()
            try:
                url = f"http://ci-bot@[::1]:{server.server_port}/repo.git"
                with patch.dict(os.environ, {"NO_PROXY": "::1,[::1]", "no_proxy": "::1,[::1]"}, clear=False):
                    subprocess.run(
                        ["git", "clone", url, str(checkout)], env=utils.git_auth_env(token),
                        check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                    )
                    subprocess.run(["git", "-C", str(checkout), "config", "user.name", "Test"], check=True)
                    subprocess.run(["git", "-C", str(checkout), "config", "user.email", "test@example.invalid"],
                                   check=True)
                    (checkout / "roundtrip.txt").write_text("authenticated\n")
                    utils.commit_and_push(checkout, "authenticated round trip", token)
            finally:
                server.shutdown()
                server.server_close()
                thread.join(timeout=5)
            result = subprocess.run(
                ["git", "--git-dir", str(bare), "show", "main:roundtrip.txt"],
                check=True, text=True, capture_output=True,
            )
            self.assertEqual("authenticated\n", result.stdout)

    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, events = [], []

            def capture(command, **kwargs):
                commands.append((command, kwargs["env"]))
                events.append("mutable")

            def publish_immutable(*_args):
                events.append("immutable")

            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), \
                    patch.object(deploy, "publish_immutable_rule", side_effect=publish_immutable) as immutable_publish, \
                    redirect_stdout(output):
                deploy.publish_route_immutables(artifact, route, root)
                deploy.s3_sync(artifact, route, root)

            self.assertTrue(all(secret not in " ".join(command) for command, _ in commands))
            self.assertEqual("immutable", events[0])
            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/dist/", rendered[0])
            self.assertIn("releases/*", rendered[0])
            self.assertNotIn("--delete", rendered[0])
            self.assertIn("--delete", rendered[1])
            self.assertTrue(all("s3://baseline-dist/dist/" in command for command in rendered[1:]))
            self.assertTrue(any("channels/" in command and
                                "public, max-age=0, must-revalidate" in command
                                for command in rendered))
            immutable_publish.assert_called_once()
            self.assertEqual(
                "public, max-age=31536000, immutable",
                immutable_publish.call_args.args[2]["cache_control"],
            )

    def test_root_move_preserves_only_actual_retired_immutable_prefix(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"),
                 "path": "/"}
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            html = root / artifact["build_dir"]
            (html / "releases").mkdir(parents=True)
            (html / "channels").mkdir()
            (html / "docs" / "releases").mkdir(parents=True)
            (html / "channels" / "stable.json").write_text("channel")
            (html / "docs" / "releases" / "index.html").write_text("mutable")
            commands = []
            with patch.dict(os.environ, {
                "DIST_S3_ACCESS_KEY": "dist-key", "DIST_S3_SECRET_KEY": "dist-secret"
            }, clear=False), patch.object(
                deploy, "run", side_effect=lambda command, **_: commands.append(command)
            ):
                deploy.s3_sync(artifact, route, root, previous_contract={
                    "path": "/foo", "access": "public", "artifact": "distributions",
                    "immutable_paths": ["foo/releases"],
                })
            rendered = [" ".join(command) for command in commands]
            self.assertTrue(all("foo/releases/*" in command for command in rendered[:2]))
            self.assertTrue(all("*/releases/*" not in command for command in rendered))

    def test_protected_bucket_cannot_become_public_across_deployments(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        previous = {
            "baseline-dist": {
                "path": "/dist", "access": "protected", "artifact": "old-name",
                "immutable_paths": ["dist/releases"],
            }
        }
        with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"):
            deploy.validate_route_migrations(cfg, previous)
        renamed = copy.deepcopy(cfg)
        artifact = next(item for item in renamed["artifacts"] if item["name"] == "distributions")
        artifact["name"] = "downloads"
        route = next(item for item in renamed["routes"] if item["artifact"] == "distributions")
        route["artifact"] = "downloads"
        with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"):
            deploy.validate_route_migrations(renamed, previous)
        previous = {"retired-protected-bucket": previous["baseline-dist"]}
        deploy.validate_route_migrations(cfg, previous)

    def test_protected_split_bucket_cannot_become_legacy_public(self):
        cfg = normalize_site_config(fixture("legacy-site.yaml"), "baseline.fritzlab.net")
        previous = {
            "baseline.fritzlab.net": {
                "path": "/portal", "access": "protected", "artifact": "portal",
                "immutable_paths": [],
            }
        }
        with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"):
            deploy.validate_route_migrations(cfg, previous)

    def test_removed_immutable_rule_preserves_prior_keys(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions")
        artifact["cache_rules"] = [
            rule for rule in artifact["cache_rules"] if rule["path"] != "releases"
        ]
        route = next(item for item in cfg["routes"] if item["artifact"] == "distributions")
        with tempfile.TemporaryDirectory() as tmp:
            html = Path(tmp)
            filters = deploy.retired_immutable_filters(artifact, route, html, {
                "path": "/dist", "access": "public", "artifact": "distributions",
                "immutable_paths": ["dist/releases"],
            })
            self.assertEqual(["--exclude", "releases/*"], filters)
            (html / "releases").mkdir()
            (html / "releases" / "replacement.js").write_text("mutable")
            with self.assertRaisesRegex(RuntimeError, "collides with retired immutable"):
                deploy.retired_immutable_filters(artifact, route, html, {
                    "path": "/dist", "access": "public", "artifact": "distributions",
                    "immutable_paths": ["dist/releases"],
                })

    def test_removed_immutable_rule_remains_in_next_manifest_history(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions")
        artifact["cache_rules"] = [
            rule for rule in artifact["cache_rules"] if rule["path"] != "releases"
        ]
        previous = {
            "baseline-dist": {
                "path": "/dist", "access": "public", "artifact": "distributions",
                "immutable_paths": ["dist/releases"],
            }
        }
        with tempfile.TemporaryDirectory() as tmp:
            app_dir = Path(tmp) / "app"
            manifests = app_dir / "manifests"
            deploy.render_site_manifests(
                "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg, previous,
            )
            first = deploy.previous_route_contracts(app_dir)
            self.assertEqual(["dist/releases"], first["baseline-dist"]["immutable_paths"])
            deploy.render_site_manifests(
                "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg, first,
            )
            second = deploy.previous_route_contracts(app_dir)
            self.assertEqual(first, second)

    def test_protected_split_bucket_survives_absence_and_blocks_legacy(self):
        raw = fixture("split-site.yaml")
        raw["artifacts"] = [
            item for item in raw["artifacts"] if item["name"] == "distributions"
        ]
        raw["routes"] = [
            item for item in raw["routes"] if item["artifact"] == "distributions"
        ]
        raw["routes"][0]["path"] = "/"
        cfg = normalize_site_config(raw, "baseline.fritzlab.net")
        previous = {
            "baseline.fritzlab.net": {
                "path": "/", "access": "protected", "artifact": "portal",
                "immutable_paths": [],
            }
        }
        with tempfile.TemporaryDirectory() as tmp:
            app_dir = Path(tmp) / "app"
            deploy.render_site_manifests(
                "baseline.fritzlab.net", ROOT, app_dir, app_dir / "manifests", cfg, previous,
            )
            retained = deploy.previous_route_contracts(app_dir)
            self.assertEqual(
                previous["baseline.fritzlab.net"], retained["baseline.fritzlab.net"],
            )
            legacy = normalize_site_config(
                fixture("legacy-site.yaml"), "baseline.fritzlab.net",
            )
            with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"):
                deploy.validate_route_migrations(legacy, retained)

    def test_malformed_persistent_history_fails_closed(self):
        with tempfile.TemporaryDirectory() as tmp:
            app_dir = Path(tmp)
            (app_dir / "site-publish-history.yaml").write_text(
                "version: 1\nbuckets:\n  bucket:\n    path: /\n"
                "    access: public\n    artifact: site\n"
                "    immutable_paths: [../releases]\n"
            )
            with self.assertRaisesRegex(RuntimeError, "invalid site-publish route history"):
                deploy.previous_route_contracts(app_dir)

    def test_retired_immutable_history_survives_a_route_move(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions")
        artifact["cache_rules"] = [
            rule for rule in artifact["cache_rules"] if rule["path"] != "releases"
        ]
        route = next(item for item in cfg["routes"] if item["artifact"] == "distributions")
        route["path"] = "/"
        previous = {
            "baseline-dist": {
                "path": "/foo", "access": "public", "artifact": "distributions",
                "immutable_paths": ["foo/releases"],
            }
        }
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            app_dir = root / "app"
            manifests = app_dir / "manifests"
            app_dir.mkdir()
            deploy.render_site_manifests(
                "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg, previous,
            )
            following = deploy.previous_route_contracts(app_dir)
            self.assertEqual(["foo/releases"], following["baseline-dist"]["immutable_paths"])
            html = root / "html"
            html.mkdir()
            first_filters = deploy.retired_immutable_filters(
                artifact, route, html, previous["baseline-dist"],
            )
            following_filters = deploy.retired_immutable_filters(
                artifact, route, html, following["baseline-dist"],
            )
            self.assertEqual(["--exclude", "foo/releases/*"], first_filters)
            self.assertEqual(first_filters, following_filters)

    def test_later_route_immutable_failure_stops_all_mutable_publication(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            with patch.object(deploy, "validate_publication_environment"), \
                    patch.object(deploy, "validate_artifact_output"), \
                    patch.object(deploy, "clone_apps", return_value=root / "apps"), patch.object(
                        deploy, "publish_route_immutables",
                        side_effect=[None, RuntimeError("immutable failed")],
                    ) as immutable_publish, patch.object(
                        deploy, "reconcile_artifact_cors"
                    ) as cors_reconcile, patch.object(deploy, "s3_sync") as mutable_sync, \
                    self.assertRaisesRegex(
                RuntimeError, "immutable failed"
            ):
                deploy.deploy_static("baseline", root, root, "token", cfg)

            self.assertEqual(2, immutable_publish.call_count)
            cors_reconcile.assert_not_called()
            mutable_sync.assert_not_called()

    def test_all_cors_policies_complete_before_mutable_publication(self):
        cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net")
        events = []
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            apps = root / "apps"
            apps.mkdir()
            with patch.object(deploy, "validate_publication_environment"), patch.object(
                deploy, "validate_artifact_output"
            ), patch.object(
                deploy, "publish_route_immutables", side_effect=lambda *_args: events.append("immutable")
            ), patch.object(
                deploy, "reconcile_artifact_cors", side_effect=lambda *_args: events.append("cors")
            ), patch.object(
                deploy, "s3_sync", side_effect=lambda *_args: events.append("mutable")
            ), patch.object(deploy, "clone_apps", return_value=apps), patch.object(
                deploy, "render_site_manifests"
            ), patch.object(deploy, "commit_and_push"):
                deploy.deploy_static("baseline", root, root, "token", cfg)

        self.assertEqual(
            ["immutable", "immutable", "cors", "mutable", "mutable"], events
        )


    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)


class ImmutablePublicationTests(unittest.TestCase):
    CACHE = "public, max-age=31536000, immutable"

    def result(self, returncode, stdout="", stderr=""):
        return subprocess.CompletedProcess([], returncode, stdout, stderr)

    def key_and_digests(self, source):
        content_type = "text/javascript"
        content_digest, publication_digest = deploy._immutable_digests(
            source, self.CACHE, content_type,
        )
        return f"dist/releases/release-{publication_digest}.js", content_digest, publication_digest

    def test_new_key_uses_content_address_and_digest_metadata(self):
        with tempfile.TemporaryDirectory() as tmp:
            source = Path(tmp) / "release.js"
            source.write_text("fixed release")
            key, content_digest, publication_digest = self.key_and_digests(source)
            calls = []

            def capture(args, _env):
                calls.append(args)
                return self.result(1, stderr="404 Not Found") if len(calls) == 1 else self.result(0, "{}")

            with patch.object(deploy, "_aws_capture", side_effect=capture):
                created = deploy.publish_immutable_file(
                    "http://garage-s3.storage.svc:3900", "dist-bucket",
                    key, source, self.CACHE, {},
                )
            self.assertTrue(created)
            put = calls[1]
            self.assertEqual("put-object", put[4])
            self.assertNotIn("--if-none-match", put)
            self.assertEqual(
                f"sha256={content_digest},publication-sha256={publication_digest}",
                put[put.index("--metadata") + 1],
            )

    def test_identical_retry_converges_without_put(self):
        with tempfile.TemporaryDirectory() as tmp:
            source = Path(tmp) / "release.js"
            source.write_text("fixed release")
            key, content_digest, publication_digest = self.key_and_digests(source)
            head = json.dumps({
                "Metadata": {"sha256": content_digest, "publication-sha256": publication_digest},
                "CacheControl": self.CACHE,
                "ContentType": "text/javascript",
            })
            with patch.object(deploy, "_aws_capture", return_value=self.result(0, head)) as request:
                created = deploy.publish_immutable_file(
                    "http://garage-s3.storage.svc:3900", "dist-bucket",
                    key, source, self.CACHE, {},
                )
            self.assertFalse(created)
            self.assertEqual(1, request.call_count)

    def test_changed_immutable_key_is_refused(self):
        with tempfile.TemporaryDirectory() as tmp:
            source = Path(tmp) / "release.js"
            source.write_text("changed release")
            key, _, _ = self.key_and_digests(source)
            head = json.dumps({"Metadata": {"sha256": "different"}, "CacheControl": self.CACHE})
            with patch.object(deploy, "_aws_capture", return_value=self.result(0, head)), \
                    self.assertRaisesRegex(RuntimeError, "immutable object differs"):
                deploy.publish_immutable_file(
                    "http://garage-s3.storage.svc:3900", "dist-bucket",
                    key, source, self.CACHE, {},
                )

    def test_immutable_key_without_publication_digest_is_refused_before_s3(self):
        with tempfile.TemporaryDirectory() as tmp:
            source = Path(tmp) / "release.js"
            source.write_text("fixed release")
            with patch.object(deploy, "_aws_capture") as request, \
                    self.assertRaisesRegex(RuntimeError, "must contain its one publication SHA-256"):
                deploy.publish_immutable_file(
                    "http://garage-s3.storage.svc:3900", "dist-bucket",
                    "dist/releases/release.js", source, self.CACHE, {},
                )
            request.assert_not_called()


if __name__ == "__main__":
    unittest.main()