feat(site-publish): add split-surface publishing

Authored-By: OpenAI (GPT-5) <noreply@openai.com>

tests/fixtures/legacy-site.yaml

@@ -0,0 +1,9 @@
+domain: example.fritzlab.net
+type: static
+content_dir: html
+aliases:
+  - www.example.fritzlab.net
+middlewares:
+  - response-headers
+excludes:
+  - media/*

tests/fixtures/split-site.yaml

@@ -0,0 +1,42 @@
+domain: baseline.fritzlab.net
+artifacts:
+  - name: portal
+    type: static
+    content_dir: portal/build
+    publish:
+      bucket: baseline-portal
+      credentials:
+        access_key_env: PORTAL_S3_ACCESS_KEY
+        secret_key_env: PORTAL_S3_SECRET_KEY
+    cache:
+      rules:
+        - path: /
+          cache_control: private, no-store
+  - name: distributions
+    type: static
+    content_dir: dist
+    publish:
+      bucket: baseline-dist
+      credentials:
+        access_key_env: DIST_S3_ACCESS_KEY
+        secret_key_env: DIST_S3_SECRET_KEY
+    cache:
+      rules:
+        - path: /
+          cache_control: public, max-age=0, must-revalidate
+        - path: releases
+          cache_control: public, max-age=31536000, immutable
+        - path: channels
+          cache_control: public, max-age=0, must-revalidate
+routes:
+  - name: portal
+    path: /
+    artifact: portal
+    access:
+      mode: protected
+      middleware: authentik-forwardauth
+  - name: distributions
+    path: /dist
+    artifact: distributions
+    access:
+      mode: public

tests/test_contract.py

@@ -0,0 +1,521 @@
+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, 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"])
+
+    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):
+        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)
+
+
+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_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.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_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()