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"]) 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", ): 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 _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) 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_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_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_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, "publish_route_immutables", side_effect=[None, RuntimeError("immutable failed")], ) as immutable_publish, patch.object( deploy, "configure_artifact_cors" ) as cors_configure, 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_configure.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, "configure_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", "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()