514 lines
23 KiB
Python
514 lines
23 KiB
Python
|
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 = []
|
||
|
|
|
||
|
|
def capture(command, **kwargs):
|
||
|
|
commands.append((command, kwargs["env"]))
|
||
|
|
|
||
|
|
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") 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.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/", rendered[0])
|
||
|
|
self.assertIn("dist/releases/*", rendered[0])
|
||
|
|
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()
|