Something went wrong. Try again.
Something went wrong. Try again.
40 kB · 778 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779#!/usr/bin/env python3"""Test battery for the bundled-catalog build assets (catalog spec §5 gates).
Offline: the fixture mode must produce assets that pass the FULL validationgates, the packer must hard-fail on missing/invalid assets (negative tests),network mode is verified against a local HTTP server (byte parity, authheader, redirect refusal, size cap), and the assets must land in the packagelayout the installer expects (assemble + verify against a synthetic repotree, including the commit-stamped continuous flow).
Run: python3 scripts/release/test_catalog_assets.py (or: make catalog-assets-gates)"""
from __future__ import annotations
import contextlibimport http.serverimport ioimport jsonimport osimport reimport shutilimport subprocessimport sysimport tarfileimport tempfileimport threadingimport timeimport unittestfrom pathlib import Path
SCRIPTS_DIR = Path(__file__).resolve().parentREPO = SCRIPTS_DIR.parent.parentsys.path.insert(0, str(SCRIPTS_DIR))
import bundle_catalog # noqa: E402 (same release-scripts directory)from assemble_artifacts import TARGET_ALIASES # noqa: E402 (same directory)
BUNDLER = SCRIPTS_DIR / "bundle_catalog.py"ASSEMBLER = SCRIPTS_DIR / "assemble_artifacts.py"VERIFIER = SCRIPTS_DIR / "verify_release.py"PACKER = REPO / "scripts" / "package_release.py"
HOST_TARGET = "x86_64-unknown-linux-gnu"# Archives carry the TS platform alias (the name the update flow's channel# manifest requires): prime-agent-<version>-<platform>.tar.gz.HOST_ARCHIVE_PLATFORM = TARGET_ALIASES[HOST_TARGET]
def run_cli(script, args, env_extra=None, cwd=None): env = dict(os.environ) env.pop("GITHUB_TOKEN", None) env.pop("PRIME_CATALOG_REPO_TOKEN", None) if env_extra: env.update(env_extra) return subprocess.run( [sys.executable, str(script), *args], capture_output=True, text=True, env=env, cwd=cwd or REPO, check=False, )
class RecordingServer: """A local HTTP stand-in for the catalog repo (never touches the network)."""
def __init__(self, routes): self.seen_headers = [] handler = self._make_handler(routes) self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) self.port = self.server.server_address[1] self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start()
def _make_handler(self, routes): seen = self.seen_headers outer = self
class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): seen.append(dict(self.headers)) route = routes.get(self.path) if route is None: self.send_response(404) self.send_header("content-length", "0") self.end_headers() return body, headers, status = route self.send_response(status) for name, value in headers.items(): self.send_header(name, value) self.send_header("content-length", str(len(body))) self.end_headers() self.wfile.write(body)
def log_message(self, *args): pass
return Handler
@property def base(self): return f"http://127.0.0.1:{self.port}"
def stop(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5)
class DripServer: """A response that always has another chunk on the way (the slow-drip shape): individual reads never hit the socket timeout, so only a TOTAL deadline can stop the fetch."""
def __init__(self, interval=0.3, chunk=b"x" * 1024): class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): self.send_response(200) # Under the 20 MiB cap so the size pre-check passes and the # read loop actually runs: only the deadline can end it (the # drip never reaches the advertised length in time). self.send_header("content-length", str(4 * 1024 * 1024)) self.end_headers() try: while True: self.wfile.write(chunk) self.wfile.flush() time.sleep(interval) except (BrokenPipeError, ConnectionResetError): pass
def log_message(self, *args): pass
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.port = self.server.server_address[1] self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start()
@property def base(self): return f"http://127.0.0.1:{self.port}"
def stop(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5)
class SyntheticRepo: """A minimal repo tree the assembler/verifier/packer accept."""
def __init__(self, root: Path, version: str = "9.9.9", sha: str | None = None): self.root = root self.version = version (root / "skills").mkdir(parents=True) (root / "skills" / "skill.md").write_text("# skill\n") (root / "LICENSE").write_text("license\n") (root / "README.md").write_text("readme\n") runtime = root / "runtime" runtime.mkdir() (runtime / "pyproject.toml").write_text("[project]\nname='rlm'\n") (runtime / "src").mkdir() # The exe-adjacent kernel runtime tree package_release.py stages. kernel = root / "prime-agent-runtime" (kernel / "src" / "rlm").mkdir(parents=True) (kernel / "pyproject.toml").write_text("[project]\nname='rlm'\n") (kernel / "src" / "rlm" / "repl.py").write_text("# repl\n") printed = version if sha is None else f"{version}-continuous.{sha}" (runtime / "src" / "repl.py").write_text("# repl\n") source = root / "prime-agent.c" source.write_text(f'#include <stdio.h>\nint main(void) {{ puts("{printed}"); return 0; }}\n') self.raw_binary = root / "bin" / "cargo-prime-agent" self.raw_binary.parent.mkdir() subprocess.run(["gcc", "-g", "-Wl,--build-id", "-o", str(self.raw_binary), str(source)], check=True) self.binary = root / "bin" / "prime-agent" self.decoder = root / "bin" / f"prime-agent-{version}-{HOST_ARCHIVE_PLATFORM}.debug.gz" split = run_cli(SCRIPTS_DIR / "split_debug.py", [ "--binary", str(self.raw_binary), "--shipped", str(self.binary), "--out", str(self.binary.parent), "--version", version, "--target", HOST_TARGET]) if split.returncode: raise AssertionError(split.stderr)
def assemble(self, out_dir, catalog_assets=None, sha=None, target=HOST_TARGET): args = [ "--repo-root", str(self.root), "--version", self.version, "--target", target, "--binary", str(self.binary), "--decoder", str(self.decoder), "--runtime-dir", str(self.root / "runtime"), "--out-dir", str(out_dir), ] if sha: args += ["--sha", sha] if catalog_assets is not None: args += ["--catalog-assets", str(catalog_assets)] return run_cli(ASSEMBLER, args)
class ValidationGates(unittest.TestCase):
def test_fixture_passes_the_full_gates(self): """Offline build verifier: --fixture produces a VALIDATED snapshot (>= 42 transport tuples, >= 68 services) without any network.""" with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "assets" result = run_cli(BUNDLER, ["generate", "--fixture", "--out", str(out)]) self.assertEqual(result.returncode, 0, result.stderr) facts = json.loads(result.stdout) self.assertGreaterEqual(facts["models"]["transportTuples"], bundle_catalog.MIN_MODEL_TRANSPORT_TUPLES) self.assertGreaterEqual(facts["mcpServices"]["services"], bundle_catalog.MIN_MCP_SERVICES) # The verify command (no small-fixture waiver) agrees. verify = run_cli(BUNDLER, ["verify", "--out", str(out)]) self.assertEqual(verify.returncode, 0, verify.stderr) self.assertGreaterEqual( json.loads(verify.stdout)["mcpServices"]["services"], bundle_catalog.MIN_MCP_SERVICES)
def test_fixture_shapes_match_the_catalog_contract(self): bodies = bundle_catalog.fixture_catalog_bodies() models = json.loads(bodies["models.bundled.json"]) self.assertEqual(set(models), {"schemaVersion", "models"}) self.assertEqual(models["schemaVersion"], 1) seen_tuples = set() for model in models["models"]: # Strict model schema (spec §1.1): no headers, bounded fields. self.assertNotIn("headers", model) for field in ("id", "name"): self.assertTrue(1 <= len(model[field]) <= 1024) self.assertFalse(any(ord(ch) < 0x20 for ch in model[field])) self.assertLessEqual(len(model["provider"]), 128) self.assertLessEqual(len(model["baseUrl"]), 2048) self.assertLessEqual(len(model["input"]), 2) self.assertTrue(set(model["input"]) <= {"text", "image"}) self.assertIn(model["api"], ("anthropic-messages", "openai-completions", "openai-responses", "bedrock-converse-stream", "google-generative-ai", "google-vertex", "mistral-conversations", "openai-codex-responses", "azure-openai-responses")) for cost in model["cost"].values(): self.assertTrue(0 <= cost <= 1_000_000) self.assertLessEqual(model["contextWindow"], 100_000_000) self.assertLessEqual(model["maxTokens"], 100_000_000) seen_tuples.add((model["provider"], model["api"], model["baseUrl"])) self.assertGreaterEqual(len(seen_tuples), 42)
plugins = json.loads(bodies["mcp-services.bundled.json"]) # Envelope: exactly version, counts, entries, in order. self.assertEqual(list(plugins), ["version", "counts", "entries"]) self.assertEqual(plugins["version"], 2) self.assertEqual(set(plugins["counts"]), set(bundle_catalog.COUNTS_KEYS)) transports = {"http", "http-template", "sse", "stdio"} auths = {"oauth", "api_key", "none", "unknown"} readiness = {"oauth-ready", "user-setup", "prime-restricted", "unknown"} setups, verifications = set(), set() for entry in plugins["entries"]: self.assertRegex(entry["server"], r"^[a-z0-9][a-z0-9-]{0,63}$") self.assertIn(entry["transport"]["type"], transports) self.assertIn(entry["auth"]["strategy"], auths) self.assertIn(entry["setup"]["readiness"], readiness) setups.add(entry["setup"]["status"]) verifications.add(entry["verification"]["status"]) self.assertEqual(entry["aliases"], sorted(a.lower() for a in entry["aliases"])) self.assertNotIn(entry["server"], entry["aliases"]) self.assertNotIn("clientId", json.dumps(entry)) self.assertNotIn("clientSecret", json.dumps(entry)) # Every shape branch is exercised. self.assertEqual(setups, {"ready", "requires-setup"}) self.assertEqual(verifications, {"metadata-reviewed", "unverified"}) self.assertEqual(plugins["counts"]["total"], len(plugins["entries"])) self.assertGreaterEqual(len(plugins["entries"]), 68)
def test_validation_rejects_bad_envelopes_and_minimums(self): with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) good = json.loads(bundle_catalog.fixture_catalog_bodies() ["mcp-services.bundled.json"]) (out / "models.bundled.json").write_text( json.dumps({"schemaVersion": 1, "models": []})) (out / "mcp-services.bundled.json").write_text(json.dumps(good)) with self.assertRaises(SystemExit): bundle_catalog.validate_bundled_model_catalog( out / "models.bundled.json") # 0 tuples < 42 small = {"schemaVersion": 1, "models": [json.loads( bundle_catalog.fixture_catalog_bodies()["models.bundled.json"] )["models"][0]]} (out / "models.bundled.json").write_text(json.dumps(small)) # The small-fixture waiver skips the minimum-count gate only. bundle_catalog.validate_bundled_model_catalog( out / "models.bundled.json", allow_small_fixture=True) bad_version = dict(good, version=3) (out / "mcp-services.bundled.json").write_text(json.dumps(bad_version)) with self.assertRaises(SystemExit): bundle_catalog.validate_bundled_mcp_catalog( out / "mcp-services.bundled.json", allow_small_fixture=True) (out / "mcp-services.bundled.json").write_text("{not json") with self.assertRaises(SystemExit): bundle_catalog.validate_bundled_mcp_catalog( out / "mcp-services.bundled.json", allow_small_fixture=True) # A missing asset fails the dir gate with an actionable message. with contextlib.redirect_stderr(io.StringIO()) as captured, \ self.assertRaises(SystemExit): bundle_catalog.validate_bundled_catalog_dir(out / "nowhere") self.assertIn("Missing bundled catalog asset", captured.getvalue())
def test_fixture_covers_paste_flow_and_builtin_shapes(self): """The special entries the runtime MCP lane consumes: one shared credentialSet alias pair (pasteable), two distinct credential sets (never pasteable), an api-key setup field, and the legacy builtins.""" plugins = json.loads( bundle_catalog.fixture_catalog_bodies()["mcp-services.bundled.json"]) by_server = {entry["server"]: entry for entry in plugins["entries"]} pasteable = by_server["fixture-paste-single"] sets = {field["credentialSet"] for field in pasteable["setup"]["fields"]} self.assertEqual(len(pasteable["setup"]["fields"]), 2) self.assertEqual(sets, {"fixture-pat"}) # alias names, one credential multi = by_server["fixture-paste-multi-cred"] sets = {field["credentialSet"] for field in multi["setup"]["fields"]} self.assertEqual(len(sets), 2) # fail-closed: NOT pasteable api_key = by_server["fixture-api-key"] self.assertEqual(api_key["setup"]["fields"][0]["kind"], "api-key") for builtin in ("linear", "notion"): self.assertTrue(by_server[builtin]["legacyBuiltin"]) # Counts stay derived: total counts every entry including specials. self.assertEqual(plugins["counts"]["total"], len(plugins["entries"])) self.assertEqual( plugins["counts"]["total"], bundle_catalog.MIN_MCP_SERVICES + 5)
def test_small_fixture_verify_waiver(self): with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "assets" out.mkdir() small_models = {"schemaVersion": 1, "models": [ {"id": "x", "name": "X", "api": "openai-responses", "provider": "openai", "baseUrl": "https://api.openai.com/v1", "reasoning": False, "input": ["text"], "cost": {"input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0}, "contextWindow": 128000, "maxTokens": 4096}]} small_plugins = {"version": 2, "counts": {}, "entries": []} (out / "models.bundled.json").write_text(json.dumps(small_models)) (out / "mcp-services.bundled.json").write_text(json.dumps(small_plugins)) strict = run_cli(BUNDLER, ["verify", "--out", str(out)]) self.assertNotEqual(strict.returncode, 0) self.assertIn("transport tuples", strict.stderr) waived = run_cli(BUNDLER, ["verify", "--out", str(out), "--allow-small-fixture"]) self.assertEqual(waived.returncode, 0, waived.stderr)
def test_fixture_http_template_entries_match_the_runtime_schema(self): """Regression (Cursor Bugbot): the fixture's http-template entries must parse under the runtime's strict catalog_schema — template variables are {name, description} objects and the entry url is empty (a fail-closed runtime parse drops the whole asset otherwise).""" plugins = json.loads(bundle_catalog.fixture_catalog_bodies() ["mcp-services.bundled.json"]) templates = [entry for entry in plugins["entries"] if entry["transport"]["type"] == "http-template"] self.assertTrue(templates, "the fixture must exercise the http-template shape") for entry in templates: transport = entry["transport"] self.assertEqual(set(transport), {"type", "template", "variables"}) self.assertTrue(transport["template"]) self.assertTrue(transport["variables"]) for variable in transport["variables"]: self.assertEqual(set(variable), {"name", "description"}) self.assertTrue(variable["name"]) self.assertTrue(variable["description"]) self.assertEqual(entry["url"], "")
def test_validator_rejects_runtime_invalid_plugin_shapes(self): """The packaging gate must reject shapes the runtime rejects (the pre-fix fixture bug: string template variables + non-empty url).""" with tempfile.TemporaryDirectory() as tmp: catalog = json.loads(bundle_catalog.fixture_catalog_bodies() ["mcp-services.bundled.json"]) entry = next(item for item in catalog["entries"] if item["transport"]["type"] == "http-template") entry["transport"]["variables"] = ["region"] path = Path(tmp) / "mcp-services.bundled.json" path.write_text(json.dumps(catalog)) with self.assertRaises(SystemExit): bundle_catalog.validate_bundled_mcp_catalog(path)
def test_validator_rejects_models_missing_required_fields(self): """The packaging gate must reject model entries the runtime's strict parse would drop (missing cost/contextWindow pass the tuple count but fail the runtime schema).""" with tempfile.TemporaryDirectory() as tmp: models = {"schemaVersion": 1, "models": [json.loads( bundle_catalog.fixture_catalog_bodies()["models.bundled.json"] )["models"][0]]} del models["models"][0]["cost"] del models["models"][0]["contextWindow"] path = Path(tmp) / "models.bundled.json" path.write_text(json.dumps(models)) with self.assertRaises(SystemExit): bundle_catalog.validate_bundled_model_catalog( path, allow_small_fixture=True)
class CatalogDirMode(unittest.TestCase):
def test_catalog_dir_copy_is_byte_identical(self): bodies = bundle_catalog.fixture_catalog_bodies() with tempfile.TemporaryDirectory() as tmp: tmp = Path(tmp) checkout = tmp / "prime-agent-catalog" (checkout / "models").mkdir(parents=True) (checkout / "plugins").mkdir() (checkout / "models" / "catalog.v1.json").write_text( bodies["models.bundled.json"]) (checkout / "plugins" / "catalog.v2.json").write_text( bodies["mcp-services.bundled.json"]) out = tmp / "assets" result = run_cli( BUNDLER, ["generate", "--catalog-dir", str(checkout), "--out", str(out)]) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual( (out / "models.bundled.json").read_bytes(), (checkout / "models" / "catalog.v1.json").read_bytes()) self.assertEqual( (out / "mcp-services.bundled.json").read_bytes(), (checkout / "plugins" / "catalog.v2.json").read_bytes()) # A checkout missing a file fails loudly. (checkout / "plugins" / "catalog.v2.json").unlink() broken = run_cli(BUNDLER, ["generate", "--catalog-dir", str(checkout), "--out", str(tmp / "assets2")]) self.assertNotEqual(broken.returncode, 0) self.assertIn("missing", broken.stderr)
class NetworkMode(unittest.TestCase):
def setUp(self): bodies = bundle_catalog.fixture_catalog_bodies() no_newline = bodies["models.bundled.json"].rstrip("\n") self.server = RecordingServer({ "/models/catalog.v1.json": (no_newline.encode(), {}, 200), "/plugins/catalog.v2.json": (bodies["mcp-services.bundled.json"].encode(), {}, 200), "/moved": (b"gone", {"location": "/models/catalog.v1.json"}, 302), "/huge": (b"x" * 1024, {"content-length": str(bundle_catalog.MAX_REMOTE_CATALOG_BYTES + 1)}, 200), })
def tearDown(self): self.server.stop()
def _generate(self, out, env_extra=None, extra=()): return run_cli(BUNDLER, [ "generate", "--network", "--models-url", f"{self.server.base}/models/catalog.v1.json", "--mcp-services-url", f"{self.server.base}/plugins/catalog.v2.json", "--out", str(out), *extra, ], env_extra=env_extra)
def test_network_fetch_is_byte_identical_and_normalizes_newlines(self): with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "assets" result = self._generate(out) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual( (out / "models.bundled.json").read_text(), bundle_catalog.fixture_catalog_bodies()["models.bundled.json"]) self.assertEqual( (out / "mcp-services.bundled.json").read_text(), bundle_catalog.fixture_catalog_bodies()["mcp-services.bundled.json"])
def test_network_sends_bearer_only_when_configured(self): with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "assets" # Without a token: no Authorization header. self.server.seen_headers.clear() result = self._generate(out) self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(self.server.seen_headers) self.assertTrue(all( "Authorization" not in headers for headers in self.server.seen_headers)) # With GITHUB_TOKEN: Bearer rides both fetches. self.server.seen_headers.clear() result = self._generate(Path(tmp) / "assets2", env_extra={"GITHUB_TOKEN": "sekrit"}) self.assertEqual(result.returncode, 0, result.stderr) auth_headers = [h.get("Authorization") for h in self.server.seen_headers] self.assertEqual(auth_headers, ["Bearer sekrit", "Bearer sekrit"])
def test_network_failures_are_hard_failures(self): with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "assets" # 404 -> hard fail with the private-repo hint. result = run_cli(BUNDLER, ["generate", "--network", "--models-url", f"{self.server.base}/missing", "--mcp-services-url", f"{self.server.base}/plugins/catalog.v2.json", "--out", str(out)]) self.assertNotEqual(result.returncode, 0) self.assertIn("HTTP 404", result.stderr) # Redirects are refused (a moved catalog is a client change). result = run_cli(BUNDLER, ["generate", "--network", "--models-url", f"{self.server.base}/moved", "--mcp-services-url", f"{self.server.base}/plugins/catalog.v2.json", "--out", str(out)]) self.assertNotEqual(result.returncode, 0) self.assertIn("refused redirect", result.stderr) # Oversized content-length is refused up front. result = run_cli(BUNDLER, ["generate", "--network", "--models-url", f"{self.server.base}/huge", "--mcp-services-url", f"{self.server.base}/plugins/catalog.v2.json", "--out", str(out)]) self.assertNotEqual(result.returncode, 0) self.assertIn("too large", result.stderr) # No partial output may be left behind by a failed fetch. self.assertFalse((out / "models.bundled.json").exists())
def test_conflicting_modes_are_rejected(self): with tempfile.TemporaryDirectory() as tmp: result = run_cli(BUNDLER, ["generate", "--fixture", "--network", "--out", str(Path(tmp) / "assets")]) self.assertNotEqual(result.returncode, 0) self.assertIn("mutually exclusive", result.stderr)
def test_failed_mcp_fetch_leaves_the_previous_snapshot_intact(self): """Both bodies are fetched before either target is replaced: a failed MCP fetch must not leave a mixed snapshot (fresh models file beside the previous services file).""" with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "assets" first = self._generate(out) self.assertEqual(first.returncode, 0, first.stderr) models_before = (out / "models.bundled.json").read_text() services_before = (out / "mcp-services.bundled.json").read_text() broken = RecordingServer({ "/models/catalog.v1.json": (models_before.encode(), {}, 200), "/plugins/catalog.v2.json": (b"gone", {}, 404), }) try: result = run_cli(BUNDLER, [ "generate", "--network", "--models-url", f"{broken.base}/models/catalog.v1.json", "--mcp-services-url", f"{broken.base}/plugins/catalog.v2.json", "--out", str(out)]) self.assertNotEqual(result.returncode, 0) self.assertIn("HTTP 404", result.stderr) finally: broken.stop() self.assertEqual((out / "models.bundled.json").read_text(), models_before) self.assertEqual((out / "mcp-services.bundled.json").read_text(), services_before)
def test_network_fetch_enforces_a_total_deadline(self): """The 5 s timeout is the TOTAL fetch budget, not per blocking op: a slow-drip server that always delivers another chunk keeps every individual read alive, so only the absolute deadline ends it.""" drip = DripServer() try: with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "assets" started = time.monotonic() result = run_cli(BUNDLER, [ "generate", "--network", "--models-url", f"{drip.base}/models/catalog.v1.json", "--mcp-services-url", f"{drip.base}/plugins/catalog.v2.json", "--out", str(out)]) elapsed = time.monotonic() - started self.assertNotEqual(result.returncode, 0) self.assertIn("hard timeout", result.stderr) self.assertLess(elapsed, 60, "the deadline must bound the total fetch time") finally: drip.stop()
class PackerGates(unittest.TestCase): """The release packer must hard-fail without validated assets, and valid assets must land in the tarball layout the installer expects."""
def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.tmp = Path(self._tmp.name) self.assets = self.tmp / "catalog-assets" result = run_cli(BUNDLER, ["generate", "--fixture", "--out", str(self.assets)]) self.assertEqual(result.returncode, 0, result.stderr) self.repo = SyntheticRepo(self.tmp / "synrepo")
def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) self._tmp.cleanup()
def test_packer_fails_without_catalog_assets(self): out = self.tmp / "dist-missing" result = self.repo.assemble(out, catalog_assets=None) self.assertNotEqual(result.returncode, 0) self.assertIn("missing bundled catalog assets", result.stderr) # Nothing was packed: no archive, no manifest. self.assertFalse( (out / f"prime-agent-9.9.9-{HOST_ARCHIVE_PLATFORM}.tar.gz").exists()) self.assertFalse((out / "manifest.json").exists())
def test_packers_never_ship_generated_skill_caches(self): cache = self.repo.root / "skills" / "__pycache__" cache.mkdir() (cache / "skill.cpython-311.pyc").write_bytes(b"generated bytecode") (self.repo.root / "skills" / "stale.pyc").write_bytes(b"generated bytecode") assembled = self.repo.assemble(self.tmp / "assemble-caches", catalog_assets=self.assets) self.assertEqual(assembled.returncode, 0, assembled.stderr) archive = self.tmp / "assemble-caches" / f"prime-agent-9.9.9-{HOST_ARCHIVE_PLATFORM}.tar.gz" with tarfile.open(archive) as tar: self.assertFalse(any("__pycache__" in name or name.endswith(".pyc") for name in tar.getnames())) poisoned = self.tmp / "poisoned" poisoned.mkdir() shutil.copy2(archive, poisoned / archive.name) with tarfile.open(poisoned / archive.name) as original: items = [(member, original.extractfile(member).read() if member.isfile() else None) for member in original.getmembers()] with tarfile.open(poisoned / archive.name, "w:gz") as bad: for member, payload in items: bad.addfile(member, io.BytesIO(payload) if payload is not None else None) extra = tarfile.TarInfo("skills/__pycache__/skill.cpython-311.pyc") extra.size = 1 bad.addfile(extra, io.BytesIO(b"x")) checked = run_cli(VERIFIER, ["--dist-dir", str(poisoned), "--version", "9.9.9", "--target", HOST_TARGET]) self.assertNotEqual(checked.returncode, 0) self.assertIn("development cache entry", checked.stderr) args = ["--root", str(self.repo.root), "--version", "9.9.9", "--binary", str(self.repo.binary), "--decoder", str(self.repo.decoder), "--skip-build", "--catalog-assets", str(self.assets), "--out-dir", str(self.tmp / "package-caches")] packaged = run_cli(PACKER, args) self.assertEqual(packaged.returncode, 0, packaged.stderr) staged = self.tmp / "package-caches" / "prime-agent-9.9.9-linux-x64" / "skills" self.assertFalse(any("__pycache__" in str(path) or path.suffix == ".pyc" for path in staged.rglob("*")))
def test_packer_fails_on_invalid_assets(self): cases = { "corrupt-json": ("{\n", "{\n"), "wrong-versions": ( json.dumps({"schemaVersion": 2, "models": []}), json.dumps({"version": 3, "entries": []})), } for label, (models_body, plugins_body) in cases.items(): bad = self.tmp / f"assets-{label}" bad.mkdir() (bad / "models.bundled.json").write_text(models_body) (bad / "mcp-services.bundled.json").write_text(plugins_body) result = self.repo.assemble(self.tmp / f"dist-{label}", catalog_assets=bad) self.assertNotEqual(result.returncode, 0, label) self.assertIn("Invalid bundled", result.stderr, label) # Under-minimum counts fail even with a well-formed envelope. small = self.tmp / "assets-small" small.mkdir() (small / "models.bundled.json").write_text( json.dumps({"schemaVersion": 1, "models": []})) good_plugins = json.loads( bundle_catalog.fixture_catalog_bodies()["mcp-services.bundled.json"]) (small / "mcp-services.bundled.json").write_text(json.dumps(good_plugins)) result = self.repo.assemble(self.tmp / "dist-small", catalog_assets=small) self.assertNotEqual(result.returncode, 0) self.assertIn("transport tuples", result.stderr) # Missing one of the two files fails too. half = self.tmp / "assets-half" half.mkdir() shutil.copyfile(self.assets / "models.bundled.json", half / "models.bundled.json") result = self.repo.assemble(self.tmp / "dist-half", catalog_assets=half) self.assertNotEqual(result.returncode, 0) self.assertIn("Missing bundled catalog asset", result.stderr)
def test_assets_land_in_the_tarball_the_installer_expects(self): out = self.tmp / "dist" result = self.repo.assemble(out, catalog_assets=self.assets) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("bundled catalog assets", result.stdout) archive = out / f"prime-agent-9.9.9-{HOST_ARCHIVE_PLATFORM}.tar.gz" self.assertTrue(archive.is_file()) with tarfile.open(archive) as tar: names = tar.getnames() self.assertIn("models.bundled.json", names) self.assertIn("mcp-services.bundled.json", names) # The staged assets are byte-identical to the generated ones. for name in ("models.bundled.json", "mcp-services.bundled.json"): member = tar.extractfile(name) self.assertEqual(member.read(), (self.assets / name).read_bytes(), name) # The full end-to-end verifier (payload shape, checksums, manifest, # extracted catalog assets, livecheck) passes on the synthetic tree. verify = run_cli(VERIFIER, ["--dist-dir", str(out), "--version", "9.9.9", "--target", HOST_TARGET]) self.assertEqual(verify.returncode, 0, verify.stderr) self.assertIn("catalog assets", verify.stdout)
def test_continuous_flow_stamps_and_verifies(self): # The #274 pipeline: commit-stamped continuous builds carry the assets # the same way, and verify enforces the stamp end to end. sha = "0123456789abcdef0123456789abcdef01234567" repo = SyntheticRepo(self.tmp / "synrepo-continuous", version="9.9.9", sha=sha) out = self.tmp / "dist-continuous" result = repo.assemble(out, catalog_assets=self.assets, sha=sha) self.assertEqual(result.returncode, 0, result.stderr) verify = run_cli(VERIFIER, ["--dist-dir", str(out), "--version", "9.9.9", "--target", HOST_TARGET, "--sha", sha]) self.assertEqual(verify.returncode, 0, verify.stderr)
def test_kernel_packer_gates_the_same_assets(self): # package_release.py (the exe-adjacent kernel packaging) fails without # assets and ships them beside the binary when they are valid. args = ["--root", str(self.repo.root), "--version", "9.9.9", "--binary", str(self.repo.binary), "--decoder", str(self.repo.decoder), "--skip-build", "--out-dir", str(self.tmp / "release-package")] without = run_cli(PACKER, args) self.assertNotEqual(without.returncode, 0) self.assertIn("missing bundled catalog assets", without.stderr) with_assets = run_cli(PACKER, [*args, "--catalog-assets", str(self.assets)]) self.assertEqual(with_assets.returncode, 0, with_assets.stderr) staged = self.tmp / "release-package" / "prime-agent-9.9.9-linux-x64" self.assertTrue((staged / "models.bundled.json").is_file()) self.assertTrue((staged / "mcp-services.bundled.json").is_file())
def test_kernel_packer_rejects_decoder_leak(self): leaked = self.repo.root / "prime-agent-runtime" / "leak.debug.gz" leaked.write_bytes(b"decoder-like payload") out = self.tmp / "release-package-leak" result = run_cli(PACKER, ["--root", str(self.repo.root), "--version", "9.9.9", "--binary", str(self.repo.binary), "--decoder", str(self.repo.decoder), "--out-dir", str(out), "--catalog-assets", str(self.assets)]) self.assertNotEqual(result.returncode, 0) self.assertIn("decoder-like sidecar", result.stderr) self.assertFalse(list(out.glob("*.tar.gz")))
def test_kernel_packer_requires_decoder_for_explicit_linux_binary(self): result = run_cli(PACKER, ["--root", str(self.repo.root), "--version", "9.9.9", "--binary", str(self.repo.binary), "--out-dir", str(self.tmp / "release-package-raw"), "--catalog-assets", str(self.assets)]) self.assertNotEqual(result.returncode, 0) self.assertIn("requires --decoder", result.stderr)
def test_linux_assembler_requires_split_binary_and_decoder(self): base = ["--repo-root", str(self.repo.root), "--version", "9.9.9", "--target", HOST_TARGET, "--runtime-dir", str(self.repo.root / "runtime"), "--catalog-assets", str(self.assets), "--out-dir", str(self.tmp / "invalid")] wrong_dir = self.tmp / "wrong-decoder" wrong_dir.mkdir() wrong_source = wrong_dir / "wrong.c" wrong_source.write_text('int main(void) { return 17; }\n') wrong_raw = wrong_dir / "cargo-prime-agent" subprocess.run(["gcc", "-g", "-Wl,--build-id", "-o", str(wrong_raw), str(wrong_source)], check=True) split = run_cli(SCRIPTS_DIR / "split_debug.py", [ "--binary", str(wrong_raw), "--shipped", str(wrong_dir / "prime-agent"), "--out", str(wrong_dir), "--version", "9.9.9", "--target", HOST_TARGET]) self.assertEqual(split.returncode, 0, split.stderr) wrong_decoder = wrong_dir / self.repo.decoder.name for flags, error in (([], "requires explicit --binary"), (["--binary", str(self.repo.binary)], "requires explicit --binary"), (["--binary", str(self.repo.raw_binary), "--decoder", str(self.repo.decoder)], "still has DWARF"), (["--binary", str(self.repo.binary), "--decoder", str(wrong_decoder)], "build ID does not match")): result = run_cli(ASSEMBLER, [*base, *flags]) self.assertNotEqual(result.returncode, 0, flags) self.assertIn(error, result.stderr) self.assertFalse(list((self.tmp / "invalid").glob("*.tar.gz")))
if __name__ == "__main__": unittest.main(verbosity=2)