from __future__ import annotations import json import os import stat import tempfile import textwrap import unittest from pathlib import Path from typing import Any from fastapi.testclient import TestClient from build_api.app import create_application from build_api.service import BuildService, Settings CHANNEL = "a" * 32 TOKEN = "test-secret" class BuildApiTest(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.base = Path(self.temporary.name) self.builder = self.base / "fake_builder.py" self.builder.write_text( textwrap.dedent( """\ #!/usr/bin/env python3 import argparse import pathlib import sys parser = argparse.ArgumentParser() parser.add_argument("--root", required=True) parser.add_argument("--channel-id", required=True) parser.add_argument("--force", action="store_true") parser.add_argument("--deployment-domains", action="append", default=[]) parser.add_argument("--reporting-domains", action="append", default=[]) parser.add_argument("--support-template", default="test") args = parser.parse_args() counter = pathlib.Path(__file__).with_name("build-count") count = int(counter.read_text()) if counter.exists() else 0 counter.write_text(str(count + 1)) if "fail.example" in args.deployment_domains: print("deliberate builder failure", file=sys.stderr) raise SystemExit(3) root = pathlib.Path(args.root) if root.exists() and not args.force: raise SystemExit("root exists; pass --force") (root / "web").mkdir(parents=True, exist_ok=True) (root / "web" / "support.html").write_text("ready") (root / "sync").mkdir(exist_ok=True) (root / "sync" / "daily.html").write_text("sync") (root / "out").mkdir(exist_ok=True) (root / "out" / "arguments.json").write_text( __import__("json").dumps(vars(args), sort_keys=True) ) (root / "manifest.json").write_text( __import__("json").dumps( { "channel_id": args.channel_id, "support_path": f"/channel/{args.channel_id}/web/support.html", "daily_path": f"/channel/{args.channel_id}/sync/daily.html", }, sort_keys=True, ) ) """ ), encoding="utf-8", ) self.builder.chmod(self.builder.stat().st_mode | stat.S_IXUSR) self.service = BuildService( Settings( artifact_root=self.base / "artifacts", bearer_token=TOKEN, project_script=self.builder, python=os.environ.get("PYTHON", os.sys.executable), build_timeout=10, ) ) self.client = TestClient( create_application(settings=self.service.settings, service=self.service) ) def tearDown(self) -> None: self.temporary.cleanup() def request( self, method: str, path: str, body: dict[str, Any] | bytes | None = None, *, authenticated: bool = True, content_type: str = "application/json", ) -> tuple[int, dict[str, Any] | None, dict[str, str]]: headers: dict[str, str] = {} if authenticated: headers["Authorization"] = f"Bearer {TOKEN}" if body is not None or method in {"POST", "PUT", "PATCH"}: headers["Content-Type"] = content_type if isinstance(body, dict): response = self.client.request(method, path, headers=headers, json=body) elif isinstance(body, bytes): headers.setdefault("Content-Type", content_type) response = self.client.request(method, path, headers=headers, content=body) else: response = self.client.request(method, path, headers=headers) payload = response.json() if response.content else None return response.status_code, payload, { key: value for key, value in response.headers.items() } @staticmethod def valid_body(**overrides: Any) -> dict[str, Any]: body: dict[str, Any] = { "deployment_domains": ["deploy.example"], "reporting_domains": ["report.example"], } body.update(overrides) return body def build_count(self) -> int: path = self.base / "build-count" return int(path.read_text()) if path.exists() else 0 def test_health_is_public_and_auth_is_required_elsewhere(self) -> None: status, payload, _ = self.request("GET", "/health", authenticated=False) self.assertEqual(200, status) self.assertEqual({"status": "ok"}, payload) status, payload, _ = self.request( "GET", f"/v1/channels/{CHANNEL}", authenticated=False ) self.assertEqual(401, status) self.assertEqual("unauthorized", payload["error"]["code"]) def test_build_status_and_builder_contract(self) -> None: status, payload, headers = self.request( "POST", f"/v1/channels/{CHANNEL}/build", self.valid_body() ) self.assertEqual(201, status) self.assertEqual(f"/v1/channels/{CHANNEL}", headers["location"]) self.assertEqual("built", payload["status"]) self.assertEqual(64, len(payload["manifest"]["input_digest"])) release = self.service.channels / CHANNEL self.assertTrue((release / "web" / "support.html").is_file()) self.assertTrue((release / "sync" / "daily.html").is_file()) self.assertEqual(payload["support_path"], f"/channel/{CHANNEL}/web/support.html") self.assertEqual(payload["daily_path"], f"/channel/{CHANNEL}/sync/daily.html") arguments = json.loads((release / "out" / "arguments.json").read_text()) self.assertEqual( str(release.parent.parent / "staging" / payload["manifest"]["request_id"]), arguments["root"], ) self.assertEqual(CHANNEL, arguments["channel_id"]) self.assertTrue(arguments["force"]) self.assertEqual("test", arguments["support_template"]) self.assertEqual("test", payload["manifest"]["input"]["support_template"]) status, current, _ = self.request("GET", f"/v1/channels/{CHANNEL}") self.assertEqual(200, status) self.assertEqual(payload["manifest"], current["manifest"]) def test_support_template_is_passed_to_builder(self) -> None: status, payload, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", self.valid_body(support_template="blank"), ) self.assertEqual(201, status) self.assertEqual("blank", payload["manifest"]["input"]["support_template"]) arguments = json.loads( (self.service.channels / CHANNEL / "out" / "arguments.json").read_text() ) self.assertEqual("blank", arguments["support_template"]) def test_identical_build_is_idempotent(self) -> None: first, _, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", self.valid_body() ) second, payload, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", self.valid_body() ) self.assertEqual(201, first) self.assertEqual(200, second) self.assertEqual("built", payload["status"]) self.assertEqual(1, self.build_count()) def test_conflict_and_force_rebuild(self) -> None: self.request("POST", f"/v1/channels/{CHANNEL}/build", self.valid_body()) changed = self.valid_body(deployment_domains=["changed.example"]) status, payload, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", changed ) self.assertEqual(409, status) self.assertEqual("build_conflict", payload["error"]["code"]) changed["force"] = True status, payload, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", changed ) self.assertEqual(201, status) self.assertEqual(["changed.example"], payload["manifest"]["input"]["deployment_domains"]) self.assertEqual(2, self.build_count()) def test_failed_force_build_keeps_old_release(self) -> None: _, original, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", self.valid_body() ) status, payload, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", self.valid_body(deployment_domains=["fail.example"], force=True), ) self.assertEqual(500, status) self.assertEqual("build_failed", payload["error"]["code"]) status, current, _ = self.request("GET", f"/v1/channels/{CHANNEL}") self.assertEqual(200, status) self.assertEqual(original["manifest"], current["manifest"]) self.assertEqual([], list(self.service.staging.iterdir())) def test_nonblocking_channel_lock_returns_conflict(self) -> None: with self.service._channel_lock(CHANNEL): status, payload, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", self.valid_body() ) self.assertEqual(409, status) self.assertEqual("channel_locked", payload["error"]["code"]) def test_delete_is_authenticated_and_idempotent(self) -> None: self.request("POST", f"/v1/channels/{CHANNEL}/build", self.valid_body()) first, payload, _ = self.request("DELETE", f"/v1/channels/{CHANNEL}") second, _, _ = self.request("DELETE", f"/v1/channels/{CHANNEL}") self.assertEqual((204, None), (first, payload)) self.assertEqual(204, second) self.assertFalse((self.service.channels / CHANNEL).exists()) def test_delete_rejects_symlink_release(self) -> None: outside = self.base / "outside" outside.mkdir() (self.service.channels / CHANNEL).symlink_to(outside, target_is_directory=True) status, payload, _ = self.request("DELETE", f"/v1/channels/{CHANNEL}") self.assertEqual(409, status) self.assertEqual("unsafe_release", payload["error"]["code"]) self.assertTrue(outside.exists()) def test_validation_errors_are_422(self) -> None: cases = [ (f"/v1/channels/{'A' * 32}/build", self.valid_body()), (f"/v1/channels/{CHANNEL}/build", {}), ( f"/v1/channels/{CHANNEL}/build", self.valid_body(deployment_domains=["https://bad.example/path"]), ), ( f"/v1/channels/{CHANNEL}/build", self.valid_body(force="yes"), ), ( f"/v1/channels/{CHANNEL}/build", self.valid_body(extra=True), ), ( f"/v1/channels/{CHANNEL}/build", self.valid_body(support_template="hud"), ), ] for path, body in cases: with self.subTest(path=path, body=body): status, payload, _ = self.request("POST", path, body) self.assertEqual(422, status) self.assertEqual("validation_error", payload["error"]["code"]) def test_malformed_json_is_422(self) -> None: status, payload, _ = self.request( "POST", f"/v1/channels/{CHANNEL}/build", b"{" ) self.assertEqual(422, status) self.assertEqual("validation_error", payload["error"]["code"]) if __name__ == "__main__": unittest.main()