from __future__ import annotations import hashlib import io import json import subprocess import sys import threading import time import uuid from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from email.message import Message from pathlib import Path from types import SimpleNamespace import pytest from test_deploy_broker import BROKER_ROOT, _load_module REPOSITORY = "jens/projectbrain" REVISION = "2713a23f6240bdabc1c6c45805a6264b9aaedad4" PREFIX = "/v1/appops/projectbrain-delivery/" PREFLIGHT_SHA = "a" * 64 ORIGINAL_FILE_SHA = "b" * 64 DELIVERY_SHA = "c" * 64 @pytest.fixture def controller(monkeypatch, tmp_path): module = _load_module( "projectbrain_delivery_revalidation_test", BROKER_ROOT / "status_server.py", ) monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True) monkeypatch.setattr(module, "_self_upgrade_secret", lambda: b"test-key-" + b"x" * 48) monkeypatch.setattr(module, "_require_controller_release", lambda: {"status": "ready"}) monkeypatch.setattr(module, "WEBHOOK_INBOX_ROOT", tmp_path) return module def invoke(module, stage: str, payload: dict[str, object], *, trusted: bool = True): handler = module.Handler.__new__(module.Handler) body = json.dumps(payload, separators=(",", ":")).encode() handler.path = PREFIX + stage handler.headers = Message() handler.headers["Content-Length"] = str(len(body)) handler.headers["Content-Type"] = "application/json" handler.rfile = io.BytesIO(body) handler._trusted_client = lambda: trusted responses = [] handler._send_json = lambda status, data: responses.append((status, json.loads(data))) handler._single_header = lambda name: ( "0" * 32 if name == "X-ITWorx-Deployment-Nonce" else "0" if name in {"X-ITWorx-Deployment-Timestamp", "X-ITWorx-Deployment-Signature"} else handler.headers.get(name) ) handler._handle_projectbrain_delivery() return responses[0] def snapshot(): return { "status": "ready", "repository": REPOSITORY, "revision": REVISION, "originalDelivery": DELIVERY_SHA, "liveRevision": "246c0a7f35e32f54fa1bf8a3f720b01bb00dd9af", "liveReceiptSha256": "d" * 64, "imageId": "sha256:" + "e" * 64, "validationRunId": 22470, "validationRunAttempt": 1, "validationJobCount": 14, "preflightSha256": PREFLIGHT_SHA, "_originalFileSha256": ORIGINAL_FILE_SHA, "_semanticDigest": DELIVERY_SHA, "_stateSha256": "1" * 64, "_receiptSha256": "d" * 64, "_runtimeSha256": "2" * 64, "_profileSha256": "3" * 64, "_targetScriptSha256": "6" * 64, "_controllerArtifactSha256": "7" * 64, "_environmentFingerprint": "4" * 64, "_validationEvidenceSha256": "5" * 64, } def test_preflight_is_signed_read_only_and_returns_only_bounded_snapshot_metadata( controller, monkeypatch ): calls = [] monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda revision: calls.append(revision) or snapshot()) monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context) monkeypatch.setattr( controller, "_enqueue_terminal_revalidation", lambda *_a, **_k: pytest.fail("preflight must not write a revalidation record"), ) code, result = invoke(controller, "preflight", {"revision": REVISION}) assert code == 200 assert result == { "status": "ready", "repository": REPOSITORY, "revision": REVISION, "originalDelivery": DELIVERY_SHA, "liveRevision": "246c0a7f35e32f54fa1bf8a3f720b01bb00dd9af", "liveReceiptSha256": "d" * 64, "imageId": "sha256:" + "e" * 64, "validationRunId": 22470, "validationRunAttempt": 1, "validationJobCount": 14, "preflightSha256": PREFLIGHT_SHA, } assert calls == [REVISION] def test_preflight_authenticates_before_reading_project_state(controller, monkeypatch): monkeypatch.setattr(controller, "_self_upgrade_signature_is_valid", lambda *_a, **_k: False) monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda *_a: pytest.fail("unauthenticated read")) assert invoke(controller, "preflight", {"revision": REVISION})[0] == 403 def test_preflight_reports_busy_without_sampling_during_deployment(controller, monkeypatch): monkeypatch.setattr( controller, "_projectbrain_delivery_mutation_lock", lambda: _raise_busy(controller), ) monkeypatch.setattr( controller, "_projectbrain_delivery_snapshot", lambda *_a: pytest.fail("preflight must not read an incoherent deployment snapshot"), ) code, result = invoke(controller, "preflight", {"revision": REVISION}) assert code == 409 assert result == {"status": "blocked", "reason": "projectbrain_deployment_active"} def test_revalidate_rechecks_fingerprint_and_binds_only_idempotency_digest( controller, monkeypatch ): observed = [] queued = [] monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda revision: observed.append(revision) or snapshot()) monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", lambda: _null_context()) monkeypatch.setattr( controller, "_request_terminal_revalidation", lambda *args, **kwargs: queued.append((args, kwargs)) or { "status": "awaiting_validation", "delivery": DELIVERY_SHA, }, ) key = "pb-revalidate-2026-10-07" code, result = invoke( controller, "revalidate", {"revision": REVISION, "expectedPreflightSha256": PREFLIGHT_SHA, "idempotencyKey": key}, ) assert code == 202 assert result["status"] == "awaiting_validation" assert result["repository"] == REPOSITORY assert result["revision"] == REVISION assert result["delivery"] == DELIVERY_SHA assert observed == [REVISION] assert queued[0][0][0:2] == (REPOSITORY, REVISION) assert queued[0][1]["expected_original_sha256"] == ORIGINAL_FILE_SHA assert queued[0][1]["projectbrain_delivery_binding"]["idempotencyKeySha256"] == hashlib.sha256(key.encode()).hexdigest() assert queued[0][1]["allow_projectbrain_terminal_idempotent"] is True assert key not in json.dumps(queued) def test_revalidate_rejects_stale_fingerprint_without_queueing(controller, monkeypatch): queued = [] monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda _revision: snapshot()) monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", lambda: _null_context()) monkeypatch.setattr(controller, "_request_terminal_revalidation", lambda *a, **k: queued.append((a, k))) code, result = invoke( controller, "revalidate", {"revision": REVISION, "expectedPreflightSha256": "f" * 64, "idempotencyKey": "pb-revalidate-2026-10-07"}, ) assert code == 409 assert result == { "status": "blocked", "reason": "projectbrain_delivery_preflight_changed", "noMutation": True, } assert queued == [] def test_revalidate_rejects_unknown_or_unbounded_fields_before_snapshot(controller, monkeypatch): monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda *_a: pytest.fail("invalid input read state")) assert invoke(controller, "preflight", {"revision": REVISION, "repository": REPOSITORY})[0] == 400 assert invoke( controller, "revalidate", {"revision": REVISION, "expectedPreflightSha256": PREFLIGHT_SHA, "idempotencyKey": "short"}, )[0] == 400 def _null_context(): from contextlib import nullcontext return nullcontext() def _raise_busy(module): raise module.ProjectBrainDeliveryRefused("projectbrain_deployment_active") def _git(cwd: Path, *arguments: str) -> str: result = subprocess.run( ["git", *arguments], cwd=cwd, capture_output=True, text=True, check=True, ) return result.stdout.strip() def _snapshot_fixture(controller, monkeypatch, tmp_path, *, target_root_env=False): repository = tmp_path / "source" repository.mkdir() _git(repository, "init", "-b", "main") _git(repository, "config", "user.email", "projectbrain-test@example.invalid") _git(repository, "config", "user.name", "ProjectBrain test") (repository / "scripts").mkdir() deploy_script = repository / "scripts" / "deploy-unraid-single.sh" deploy_script.write_text("#!/bin/sh\nprintf 'live\\n'\n", encoding="utf-8") (repository / "README.md").write_text("live\n", encoding="utf-8") _git(repository, "add", ".") _git(repository, "commit", "-m", "live") live_revision = _git(repository, "rev-parse", "HEAD") deploy_script.write_text("#!/bin/sh\nprintf 'target\\n'\n", encoding="utf-8") (repository / "README.md").write_text("target\n", encoding="utf-8") if target_root_env: (repository / ".env").write_text( "PROJECTBRAIN_PRIVATE_TEST=fixture-only\n", encoding="utf-8", ) _git(repository, "add", ".") _git(repository, "commit", "-m", "target") revision = _git(repository, "rev-parse", "HEAD") repository_path = repository / ".git" validation = { "validation_gate_version": 1, "required": True, "contexts": ["Managed validation / full (push)"], "timeout_seconds": 1800, "poll_interval_seconds": 5, "max_poll_interval_seconds": 30, } project = { "slug": "projectbrain", "branch": "main", "source_dir": "/srv/example/appdata/brain", "deploy_script": "scripts/deploy-unraid-single.sh", "archive_deploy_script": None, "containers": ["projectbrain"], "compose_project": "projectbrain", "compose_files": ["docker-compose.yml"], "gitea_repository_id": 7, "environment": { "PROJECTBRAIN_DEPLOY_SKIP_PULL": "true", "PROJECTBRAIN_DEPLOY_REF": "", "VITE_PROJECTBRAIN_COMMIT": "{revision}", }, "gitea_webhook_secret_file": "jens__projectbrain", "verify_oci_revision": False, "health_timeout": 180, "validation": validation, } inbox = tmp_path / "inbox" inbox.mkdir(mode=0o700) monkeypatch.setattr(controller, "WEBHOOK_INBOX_ROOT", inbox) monkeypatch.setattr(controller, "_load_projects", lambda: {REPOSITORY: project}) monkeypatch.setattr(controller, "_is_current_branch_head", lambda *_a: True) monkeypatch.setattr(controller, "_repository_path_for_delivery", lambda *_a: repository_path) controller_release = { "artifactSha256": "8" * 64, "sourceRepository": "jens/unraid-appops-gateway", "sourceRevision": "9" * 40, "sourceTree": "a" * 40, "builtAt": "2026-10-07T12:00:00.000Z", "controllerVersion": "test", "apiVersion": "itworx.controller-release/v1", } monkeypatch.setattr(controller, "_require_controller_release", lambda: controller_release) semantic = controller._webhook_semantic_digest(7, "refs/heads/main", revision) now = int(time.time()) original = { "schema": controller._WEBHOOK_SCHEMA, "semantic_sha256": semantic, "gitea_repository_id": 7, "ref": "refs/heads/main", "repository": REPOSITORY, "revision": revision, "source": "gitea_webhook", "status": "validation_timeout", "reason": "validation_timeout", "delivery": str(uuid.uuid4()), "body_sha256": "f" * 64, "validation": controller._validation_snapshot( controller._validation_policy(REPOSITORY, project), now, ), "audit": [{"at": now, "stage": "validation_timeout", "reason": "validation_timeout"}], "received_at": now - 1800, "updated_at": now, } original_path = inbox / f"{semantic}.json" controller._write_new_inbox_record(original_path, original) tree = _git(repository, "rev-parse", f"{live_revision}^{{tree}}") tree_digest = hashlib.sha256(f"git-tree:{tree}".encode()).hexdigest() image_id = "sha256:" + "e" * 64 built_at = "2026-10-07T11:00:00.000Z" receipt = { "schema": "gitea-deploy-receipt.v1", "repository": REPOSITORY, "revision": live_revision, "tree": tree, "treeDigest": tree_digest, "build": { "id": live_revision[:12], "repository": REPOSITORY, "revision": live_revision, "tree": tree, "treeDigest": tree_digest, "builtAt": built_at, }, "containers": {"projectbrain": {"image_id": image_id}}, "live": None, } state = { "schema_version": 3, "repository": REPOSITORY, "live_revision": live_revision, "live_status": "healthy", "live_legacy_provenance": False, "live_receipt": receipt, } monkeypatch.setattr(controller, "_deployment_state", lambda _slug: state) container = { "Id": "d" * 64, "Image": image_id, "Name": "/projectbrain", "State": {"Running": True, "Status": "running", "Health": {"Status": "healthy"}}, "Config": {"Image": "projectbrain-unraid:codex-oidc-20260826"}, "Mounts": [{ "Type": "bind", "Source": "/srv/example/appdata/projectbrain", "Destination": "/var/lib/projectbrain", "RW": True, }], } monkeypatch.setattr(controller, "_container_inventory", lambda: {"projectbrain": container}) run_id = 22470 run = { "id": run_id, "run_number": 381, "run_attempt": 1, "created_at": "2026-10-07T12:00:00Z", "path": "managed-validation.yml@refs/heads/main", "event": "push", "head_branch": "main", "head_sha": revision, "repository": {"full_name": REPOSITORY}, "head_repository": {"full_name": REPOSITORY}, "pull_requests": [], "status": "completed", "conclusion": "success", } jobs = [{ "id": 39165, "name": "full", "run_id": run_id, "run_attempt": 1, "head_sha": revision, "head_branch": "main", "status": "completed", "conclusion": "success", }] def api_json(path, *, query): if path.endswith("/branches/main"): return {"protected": True, "commit": {"id": revision}} if path.endswith("/actions/runs"): return {"total_count": 1, "workflow_runs": [run]} if path.endswith(f"/actions/runs/{run_id}/jobs"): return {"total_count": len(jobs), "jobs": jobs} raise AssertionError(f"unexpected bounded validation API path: {path}; query={query}") adoption = _load_module( "projectbrain_delivery_validation_evidence_test", BROKER_ROOT / "registered_legacy_adoption.py", ) monkeypatch.setitem(sys.modules, "registered_legacy_adoption", adoption) monkeypatch.setattr(controller, "_registered_legacy_api_reader", lambda _repository: api_json) deploy = _load_module("projectbrain_delivery_snapshot_deploy_test", BROKER_ROOT / "deploy.py") def snapshot_binding(_project, observed_container): mounts = observed_container.get("Mounts") or [] mount = mounts[0] if len(mounts) == 1 else {} return { "app_dir": "/srv/example/appdata/brain", "data_path": mount.get("Source", ""), "container_name": "projectbrain", "image_ref": str(dict(observed_container.get("Config") or {}).get("Image") or ""), "image_id": str(observed_container.get("Image") or ""), } monkeypatch.setattr(deploy, "projectbrain_snapshot_binding", snapshot_binding) monkeypatch.setattr(deploy, "projectbrain_rollback_environment", lambda *_a: {}) environment = {"PROJECTBRAIN_OWNER_PASSWORD": "test-only-secret"} monkeypatch.setattr(deploy, "static_projectbrain_environment", lambda _path: dict(environment)) monkeypatch.setitem(sys.modules, "deploy", deploy) run_process = subprocess.run def bounded_run(command, **kwargs): if command[0] == "docker": return subprocess.CompletedProcess(command, 0, stdout=image_id + "\n", stderr="") return run_process(command, **kwargs) monkeypatch.setattr(controller.subprocess, "run", bounded_run) return { "project": project, "state": state, "receipt": receipt, "container": container, "jobs": jobs, "environment": environment, "original_path": original_path, "original": original, "original_bytes": original_path.read_bytes(), "inbox": inbox, "repository_path": repository_path, "live_revision": live_revision, "revision": revision, "semantic": semantic, "image_id": image_id, "controller_release": controller_release, } @pytest.mark.parametrize("exact_source", ["omitted", False]) def test_projectbrain_fixed_profile_accepts_only_the_existing_source_mode( controller, monkeypatch, tmp_path, exact_source, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) project = dict(fixture["project"]) if exact_source == "omitted": project.pop("exact_source", None) else: project["exact_source"] = exact_source accepted = controller._projectbrain_delivery_project(project) assert accepted["_repository_name"] == REPOSITORY assert accepted["source_dir"] == "/srv/example/appdata/brain" if exact_source == "omitted": assert "exact_source" not in accepted else: assert accepted["exact_source"] is False @pytest.mark.parametrize("exact_source", [True, "false", 0, None]) def test_projectbrain_fixed_profile_rejects_other_source_modes( controller, monkeypatch, tmp_path, exact_source, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) project = {**fixture["project"], "exact_source": exact_source} with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_project(project) assert failure.value.code == "projectbrain_target_contract_changed" @pytest.mark.parametrize( ("key", "value"), [ ("env_source", None), ("env_source", False), ("env_source", 0), ("env_source", ""), ("compose_dir", None), ("compose_dir", False), ("compose_dir", 0), ("compose_dir", ""), ], ) def test_projectbrain_profile_requires_unconfigured_external_environment_and_compose_dir( controller, monkeypatch, tmp_path, key, value, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) project = {**fixture["project"], key: value} with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_project(project) assert failure.value.code == "projectbrain_target_contract_changed" @pytest.mark.parametrize( ("key", "value"), [ ("PROJECTBRAIN_DEPLOY_SKIP_PULL", "false"), ("PROJECTBRAIN_DEPLOY_SKIP_PULL", False), ("PROJECTBRAIN_DEPLOY_SKIP_PULL", 1), ("PROJECTBRAIN_DEPLOY_REF", "main"), ("PROJECTBRAIN_DEPLOY_REF", None), ("PROJECTBRAIN_DEPLOY_REF", False), ("PROJECTBRAIN_DEPLOY_REF", 0), ("VITE_PROJECTBRAIN_COMMIT", "a" * 40), ("VITE_PROJECTBRAIN_COMMIT", None), ("VITE_PROJECTBRAIN_COMMIT", False), ], ) def test_projectbrain_profile_rejects_execution_override_drift( controller, monkeypatch, tmp_path, key, value, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) project = { **fixture["project"], "environment": dict(fixture["project"]["environment"]), } project["environment"][key] = value with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_project(project) assert failure.value.code == "projectbrain_target_contract_changed" @pytest.mark.parametrize("deploy_ref", ["omitted", ""]) def test_projectbrain_profile_accepts_only_absent_or_empty_deploy_ref( controller, monkeypatch, tmp_path, deploy_ref, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) project = { **fixture["project"], "environment": dict(fixture["project"]["environment"]), } if deploy_ref == "omitted": project["environment"].pop("PROJECTBRAIN_DEPLOY_REF") else: project["environment"]["PROJECTBRAIN_DEPLOY_REF"] = deploy_ref assert controller._projectbrain_delivery_project(project)["environment"] == ( project["environment"] ) def test_actual_projectbrain_profile_passes_real_deploy_source_validation_and_binding( controller, tmp_path, ): repository = tmp_path / "repository" repository.mkdir() _git(repository, "init", "-b", "main") _git(repository, "config", "user.email", "projectbrain-test@example.invalid") _git(repository, "config", "user.name", "ProjectBrain test") (repository / "scripts").mkdir() (repository / "scripts" / "deploy-unraid-single.sh").write_text("#!/bin/sh\n", encoding="utf-8") (repository / "docker-compose.yml").write_text("services: {}\n", encoding="utf-8") _git(repository, "add", ".") _git(repository, "commit", "-m", "live") revision = _git(repository, "rev-parse", "HEAD") project = { "slug": "projectbrain", "branch": "main", "source_dir": "/srv/example/appdata/brain", "deploy_script": "scripts/deploy-unraid-single.sh", "archive_deploy_script": None, "containers": ["projectbrain"], "compose_project": "projectbrain", "compose_files": ["docker-compose.yml"], "gitea_repository_id": 7, "environment": { "PROJECTBRAIN_DEPLOY_SKIP_PULL": "true", "PROJECTBRAIN_DEPLOY_REF": "", "VITE_PROJECTBRAIN_COMMIT": "{revision}", }, } deploy = _load_module( "projectbrain_delivery_deploy_profile_test", BROKER_ROOT / "deploy.py", ) deploy.validate_project_config(repository / ".git", project, revision) mount_source = str(tmp_path / "projectbrain-data") container = { "Config": {"Image": "projectbrain-unraid:codex-oidc-20260826"}, "Image": "sha256:" + "a" * 64, "Mounts": [{ "Type": "bind", "Source": mount_source, "Destination": "/var/lib/projectbrain", "RW": True, }], } binding = deploy.projectbrain_snapshot_binding( {**project, "_repository_name": REPOSITORY}, container, ) assert binding["app_dir"] == str(Path(project["source_dir"]).resolve()) assert binding["data_path"] == mount_source assert binding["image_id"] == container["Image"] assert controller._projectbrain_delivery_project(project)["source_dir"] == project["source_dir"] exact_source_project = {**project, "exact_source": True} with pytest.raises(deploy.DeployError, match="Exact source directory is outside"): deploy.validate_project_config(repository / ".git", exact_source_project, revision) def test_snapshot_binds_full_profile_target_script_controller_artifact_and_real_push_evidence( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) first = controller._projectbrain_delivery_snapshot(fixture["revision"]) fixture["project"]["health_timeout"] = 181 second = controller._projectbrain_delivery_snapshot(fixture["revision"]) fixture["controller_release"]["artifactSha256"] = "9" * 64 third = controller._projectbrain_delivery_snapshot(fixture["revision"]) assert first["status"] == second["status"] == third["status"] == "ready" assert first["_profileSha256"] != second["_profileSha256"] assert first["preflightSha256"] != second["preflightSha256"] assert second["_controllerArtifactSha256"] != third["_controllerArtifactSha256"] assert second["preflightSha256"] != third["preflightSha256"] assert first["_targetScriptSha256"] == controller._projectbrain_delivery_target_script_sha256( fixture["repository_path"], fixture["revision"], ) assert first["_targetScriptSha256"] != controller._projectbrain_delivery_target_script_sha256( fixture["repository_path"], fixture["live_revision"], ) assert first["_controllerArtifactSha256"] == "8" * 64 assert first["validationRunId"] == 22470 assert first["validationJobCount"] == 1 assert "test-only-secret" not in json.dumps(controller._projectbrain_delivery_public_snapshot(first)) def test_snapshot_rejects_target_revision_tracking_private_root_environment_file( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture( controller, monkeypatch, tmp_path, target_root_env=True, ) with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_snapshot(fixture["revision"]) assert failure.value.code == "projectbrain_source_tracks_private_environment" def test_snapshot_rejects_incomplete_full_job_identity_and_extra_required_context( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) adoption = sys.modules["registered_legacy_adoption"] monkeypatch.setattr(adoption, "_validation_evidence", lambda *_a: { "revision": fixture["revision"], "branch": "main", "event": "push", "runId": 22470, "runNumber": 381, "runAttempt": 1, "workflowPath": "managed-validation.yml@refs/heads/main", "validationContext": "Managed validation / full (push)", "jobCount": 1, "jobIds": [], }) with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_snapshot(fixture["revision"]) assert failure.value.code == "projectbrain_full_push_validation_invalid" monkeypatch.setattr(adoption, "_validation_evidence", lambda *_args: { "revision": fixture["revision"], "branch": "main", "event": "push", "runId": 22470, "runNumber": 381, "runAttempt": 1, "workflowPath": "managed-validation.yml@refs/heads/main", "validationContext": "Managed validation / full (push)", "jobCount": 1, "jobIds": [39165], }) fixture["project"]["validation"]["contexts"].append("Extra required status (push)") with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_snapshot(fixture["revision"]) assert failure.value.code == "projectbrain_validation_policy_unavailable" @pytest.mark.parametrize( ("change", "expected_code"), [ ("unhealthy", "projectbrain_runtime_unhealthy"), ("mount", "projectbrain_runtime_binding_changed"), ("receipt_image", "projectbrain_live_image_changed"), ], ) def test_snapshot_rejects_unhealthy_runtime_mount_or_unreceipted_image( controller, monkeypatch, tmp_path, change, expected_code, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) if change == "unhealthy": fixture["container"]["State"]["Health"]["Status"] = "unhealthy" elif change == "mount": fixture["container"]["Mounts"][0]["Source"] = "/srv/example/appdata/other" else: fixture["receipt"]["containers"]["projectbrain"]["image_id"] = "sha256:" + "f" * 64 with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_snapshot(fixture["revision"]) assert failure.value.code == expected_code def test_snapshot_binds_private_environment_and_rejects_deployment_history_in_original( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) first = controller._projectbrain_delivery_snapshot(fixture["revision"]) fixture["environment"]["PROJECTBRAIN_OWNER_PASSWORD"] = "changed-test-only-secret" second = controller._projectbrain_delivery_snapshot(fixture["revision"]) assert first["preflightSha256"] != second["preflightSha256"] assert "changed-test-only-secret" not in json.dumps(controller._projectbrain_delivery_public_snapshot(second)) changed_original = dict(fixture["original"]) changed_original["audit"] = [ *changed_original["audit"], {"at": int(time.time()), "stage": "deployment_started"}, ] fixture["original_path"].write_bytes( controller._projectbrain_delivery_canonical(changed_original) + b"\n", ) with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_snapshot(fixture["revision"]) assert failure.value.code == "projectbrain_original_delivery_not_revalidatable" def test_snapshot_rejects_private_vite_revision_override_without_exposing_value( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) fixture["environment"]["VITE_PROJECTBRAIN_COMMIT"] = "test-only-private-value" with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_delivery_snapshot(fixture["revision"]) assert failure.value.code == "projectbrain_private_environment_overrides_revision" assert "test-only-private-value" not in str(failure.value) def test_deploy_gate_binds_original_bytes_and_signed_sibling_without_rewriting_history( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) initial = controller._projectbrain_delivery_snapshot(fixture["revision"]) idempotency_hash = hashlib.sha256(b"pb-revalidate-2026-10-07").hexdigest() binding = controller._projectbrain_delivery_signed_binding(initial, idempotency_hash) attempt = { "schema": controller._WEBHOOK_REVALIDATION_SCHEMA, "semantic_sha256": fixture["semantic"], "revalidation_of": fixture["semantic"], "gitea_repository_id": 7, "ref": "refs/heads/main", "repository": REPOSITORY, "revision": fixture["revision"], "status": "deploying", "projectbrain_delivery_revalidation": binding, } marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" controller._write_new_inbox_record(marker, attempt) assert controller._projectbrain_delivery_deploy_gate( REPOSITORY, fixture["revision"], initial["preflightSha256"], ) assert fixture["original_path"].read_bytes() == fixture["original_bytes"] changed_original = dict(fixture["original"]) changed_original["operator_note"] = "whole-record drift" fixture["original_path"].write_bytes( controller._projectbrain_delivery_canonical(changed_original) + b"\n", ) assert not controller._projectbrain_delivery_deploy_gate( REPOSITORY, fixture["revision"], initial["preflightSha256"], ) fixture["original_path"].write_bytes(fixture["original_bytes"]) fixture["original_path"].write_bytes(fixture["original_bytes"] + b" ") with pytest.raises(controller.ProjectBrainDeliveryRefused): controller._projectbrain_read_original_delivery(fixture["semantic"]) fixture["original_path"].write_bytes(fixture["original_bytes"]) attempt["projectbrain_delivery_revalidation"] = {**binding, "hmac": "0" * 64} marker.write_bytes(controller._projectbrain_delivery_canonical(attempt) + b"\n") assert not controller._projectbrain_delivery_deploy_gate( REPOSITORY, fixture["revision"], initial["preflightSha256"], ) def test_same_idempotency_key_replays_progressed_marker_and_changed_key_conflicts( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) preflight = snapshot() preflight["_semanticDigest"] = fixture["semantic"] preflight["revision"] = fixture["revision"] key = "pb-revalidate-2026-10-07" key_hash = hashlib.sha256(key.encode()).hexdigest() binding = controller._projectbrain_delivery_signed_binding(preflight, key_hash) record = { "schema": controller._WEBHOOK_REVALIDATION_SCHEMA, "semantic_sha256": fixture["semantic"], "revalidation_of": fixture["semantic"], "gitea_repository_id": 7, "ref": "refs/heads/main", "repository": REPOSITORY, "revision": fixture["revision"], "status": "deployed", "projectbrain_delivery_revalidation": binding, } marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" controller._write_new_inbox_record(marker, record) assert controller._projectbrain_existing_revalidation( fixture["revision"], key_hash, PREFLIGHT_SHA, ) == {"status": "deployed", "delivery": fixture["semantic"]} with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: controller._projectbrain_existing_revalidation( fixture["revision"], hashlib.sha256(b"another-key-2026").hexdigest(), PREFLIGHT_SHA, ) assert failure.value.code == "projectbrain_idempotency_conflict" def test_same_key_replay_returns_the_existing_deploying_phase(controller, monkeypatch, tmp_path): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) preflight = snapshot() preflight["_semanticDigest"] = fixture["semantic"] preflight["revision"] = fixture["revision"] key = "pb-revalidate-active-2026" key_hash = hashlib.sha256(key.encode()).hexdigest() binding = controller._projectbrain_delivery_signed_binding(preflight, key_hash) record = { "schema": controller._WEBHOOK_REVALIDATION_SCHEMA, "semantic_sha256": fixture["semantic"], "revalidation_of": fixture["semantic"], "gitea_repository_id": 7, "ref": "refs/heads/main", "repository": REPOSITORY, "revision": fixture["revision"], "status": "deploying", "projectbrain_delivery_revalidation": binding, } controller._write_new_inbox_record( fixture["inbox"] / f"{fixture['semantic']}.revalidation.json", record, ) assert controller._projectbrain_existing_revalidation( fixture["revision"], key_hash, PREFLIGHT_SHA, ) == {"status": "deploying", "delivery": fixture["semantic"]} def test_concurrent_enqueue_creates_one_additive_attempt_and_preserves_original( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) preflight = controller._projectbrain_delivery_snapshot(fixture["revision"]) binding = controller._projectbrain_delivery_signed_binding( preflight, hashlib.sha256(b"pb-revalidate-concurrent-2026").hexdigest(), ) project = fixture["project"] marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" writer_opened = threading.Event() allow_writer_to_continue = threading.Event() gate_lock = threading.Lock() gate_used = False real_open = controller.os.open def gated_open(path, flags, mode=0o777): nonlocal gate_used descriptor = real_open(path, flags, mode) name = Path(path).name if name == marker.name or name.startswith(f".{marker.name}."): with gate_lock: pause = not gate_used gate_used = True if pause: writer_opened.set() if not allow_writer_to_continue.wait(timeout=5): controller.os.close(descriptor) raise TimeoutError("test writer gate timed out") return descriptor monkeypatch.setattr(controller.os, "open", gated_open) def enqueue(): return controller._enqueue_terminal_revalidation( REPOSITORY, fixture["revision"], project, expected_original_sha256=preflight["_originalFileSha256"], projectbrain_delivery_binding=binding, ) with ThreadPoolExecutor(max_workers=2) as workers: first = workers.submit(enqueue) assert writer_opened.wait(timeout=5) try: assert not marker.exists(), "an incomplete revalidation marker became visible" second = workers.submit(enqueue) assert second.result(timeout=5) == ("awaiting_validation", fixture["semantic"]) finally: allow_writer_to_continue.set() assert first.result(timeout=5) == ("awaiting_validation", fixture["semantic"]) assert fixture["original_path"].read_bytes() == fixture["original_bytes"] attempt_files = list(fixture["inbox"].glob("*.revalidation.json")) assert len(attempt_files) == 1 record = controller._read_inbox_record(attempt_files[0]) assert record["projectbrain_delivery_revalidation"] == binding assert not list(fixture["inbox"].glob(".*.tmp")) def test_new_inbox_record_ignores_stale_temporary_file(controller, tmp_path): marker = tmp_path / "stale.json" record = {"record": "complete"} stale = marker.with_name( f".{marker.name}.{controller.os.getpid()}.{controller.threading.get_ident()}.tmp" ) stale_bytes = b"stale temporary file from an earlier process\n" stale.write_bytes(stale_bytes) controller._write_new_inbox_record(marker, record) assert json.loads(marker.read_text(encoding="utf-8")) == record if controller.os.name != "nt": assert marker.stat().st_mode & 0o777 == 0o600 assert stale.read_bytes() == stale_bytes assert set(tmp_path.iterdir()) == {marker, stale} def test_new_inbox_record_collision_preserves_existing_bytes_and_cleans_temp( controller, tmp_path, ): marker = tmp_path / "existing.json" original = b'{"existing":true}\n' marker.write_bytes(original) with pytest.raises(FileExistsError): controller._write_new_inbox_record(marker, {"replacement": True}) assert marker.read_bytes() == original assert not list(tmp_path.glob(".*.tmp")) def test_new_inbox_record_write_failure_cleans_temporary_file( controller, monkeypatch, tmp_path, ): marker = tmp_path / "failed.json" real_fdopen = controller.os.fdopen class FailingWriter: def __init__(self, handle): self.handle = handle def __enter__(self): self.handle.__enter__() return self def __exit__(self, exc_type, exc, traceback): return self.handle.__exit__(exc_type, exc, traceback) def write(self, _payload): raise OSError("simulated inbox write failure") monkeypatch.setattr( controller.os, "fdopen", lambda descriptor, mode: FailingWriter(real_fdopen(descriptor, mode)), ) with pytest.raises(OSError, match="simulated inbox write failure"): controller._write_new_inbox_record(marker, {"record": True}) assert not marker.exists() assert not list(tmp_path.glob(".*.tmp")) @pytest.mark.parametrize("failure_point", ["directory_fsync", "wakeup"]) def test_publish_followup_failure_reports_unknown_and_same_key_replay_resolves( controller, monkeypatch, tmp_path, failure_point, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context) preflight = controller._projectbrain_delivery_snapshot(fixture["revision"]) idempotency_key = "pb-revalidate-publish-failure-2026" payload = { "revision": fixture["revision"], "expectedPreflightSha256": preflight["preflightSha256"], "idempotencyKey": idempotency_key, } real_fsync = controller._fsync_inbox_root real_wakeup = controller._webhook_wakeup if failure_point == "directory_fsync": def fail_fsync(): raise OSError("simulated directory fsync failure") monkeypatch.setattr(controller, "_fsync_inbox_root", fail_fsync) else: def fail_wakeup(): raise RuntimeError("simulated worker wakeup failure") monkeypatch.setattr(controller, "_webhook_wakeup", SimpleNamespace(set=fail_wakeup)) code, result = invoke(controller, "revalidate", payload) assert code == 503 assert result == { "status": "unknown", "reason": "projectbrain_delivery_enqueue_outcome_unknown", "noMutation": False, } assert fixture["original_path"].read_bytes() == fixture["original_bytes"] marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" record = controller._read_inbox_record(marker) assert record is not None assert record["schema"] == controller._WEBHOOK_REVALIDATION_SCHEMA assert record["status"] == "awaiting_validation" assert record["projectbrain_delivery_revalidation"]["idempotencyKeySha256"] == hashlib.sha256( idempotency_key.encode("ascii"), ).hexdigest() assert not list(fixture["inbox"].glob(".*.tmp")) if failure_point == "directory_fsync": monkeypatch.setattr(controller, "_fsync_inbox_root", real_fsync) else: monkeypatch.setattr(controller, "_webhook_wakeup", real_wakeup) replay_code, replay = invoke(controller, "revalidate", payload) assert replay_code == 202 assert replay["status"] == "awaiting_validation" assert replay["delivery"] == fixture["semantic"] assert fixture["original_path"].read_bytes() == fixture["original_bytes"] def test_revalidation_preflight_refusal_reports_no_mutation( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context) def refuse_snapshot(_revision): raise controller.ProjectBrainDeliveryRefused("projectbrain_delivery_preflight_changed") monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", refuse_snapshot) payload = { "revision": fixture["revision"], "expectedPreflightSha256": "a" * 64, "idempotencyKey": "pb-revalidate-preflight-refusal-2026", } code, result = invoke(controller, "revalidate", payload) assert code == 409 assert result == { "status": "blocked", "reason": "projectbrain_delivery_preflight_changed", "noMutation": True, } assert fixture["original_path"].read_bytes() == fixture["original_bytes"] assert not (fixture["inbox"] / f"{fixture['semantic']}.revalidation.json").exists() def test_generic_terminal_revalidation_cannot_occupy_projectbrain_sibling( controller, monkeypatch, tmp_path, ): fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) result = controller._enqueue_terminal_revalidation( REPOSITORY, fixture["revision"], fixture["project"], ) assert result == ("invalid", "") assert fixture["original_path"].read_bytes() == fixture["original_bytes"] assert not (fixture["inbox"] / f"{fixture['semantic']}.revalidation.json").exists() def test_deploy_rejection_precedes_source_sync_under_project_and_global_locks( monkeypatch, tmp_path, ): deploy = _load_module("projectbrain_delivery_deploy_gate_test", BROKER_ROOT / "deploy.py") revision = "a" * 40 project = { "slug": "projectbrain", "branch": "main", "source_dir": str(tmp_path / "source"), "exact_source": True, "deploy_script": "scripts/deploy-unraid-single.sh", "containers": ["projectbrain"], "compose_project": "projectbrain", "compose_files": ["docker-compose.yml"], } active_locks = [] project_lock = deploy.STATE_ROOT / "locks" / "projectbrain.lock" @contextmanager def lock(path, _description): active_locks.append(path) try: yield finally: active_locks.pop() monkeypatch.setattr(deploy, "exclusive_lock", lock) monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: lock( deploy.GLOBAL_DEPLOY_LOCK, "global", )) monkeypatch.setattr(deploy, "load_projects", lambda: {REPOSITORY: dict(project)}) monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, config: config) monkeypatch.setattr(deploy, "repository_path", lambda *_a: tmp_path) monkeypatch.setattr(deploy, "bind_appops_self_upgrade_recovery", lambda *_a: None) monkeypatch.setattr(deploy, "bind_registered_legacy_request", lambda *_a: None) monkeypatch.setattr(deploy, "validate_commit", lambda *_a: revision) monkeypatch.setattr(deploy, "deployment_metadata", lambda *_a: {"revision": revision}) monkeypatch.setattr(deploy, "validate_project_config", lambda *_a: None) monkeypatch.setattr( deploy, "sync_source", lambda *_a, **_k: pytest.fail("source sync must not run after a stale preflight"), ) monkeypatch.setattr( deploy, "copy_environment", lambda *_a, **_k: pytest.fail("private environment copy must not run after a stale preflight"), ) gate_calls = [] def gate(_repository, _revision, _fingerprint): gate_calls.append(list(active_locks)) return False monkeypatch.setitem(sys.modules, "status_server", SimpleNamespace( _projectbrain_delivery_deploy_gate=gate, )) monkeypatch.setenv(deploy.PROJECTBRAIN_DELIVERY_PREFLIGHT_ENV, "b" * 64) monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", REPOSITORY, revision]) with pytest.raises(deploy.DeployError, match="preflight changed before deployment"): deploy.main() assert gate_calls == [[project_lock, deploy.GLOBAL_DEPLOY_LOCK]] @pytest.mark.parametrize("restore_fails", [False, True]) def test_second_projectbrain_preflight_failure_restores_tracked_source_and_records_live_state( monkeypatch, tmp_path, restore_fails, ): deploy = _load_module("projectbrain_delivery_second_gate_test", BROKER_ROOT / "deploy.py") repository_path = tmp_path / "repository" repository_path.mkdir() _git(repository_path, "init", "-b", "main") _git(repository_path, "config", "user.email", "projectbrain-test@example.invalid") _git(repository_path, "config", "user.name", "ProjectBrain test") source_path = repository_path / "README.md" source_path.write_text("prior live source\n", encoding="utf-8") _git(repository_path, "add", ".") _git(repository_path, "commit", "-m", "prior") prior_revision = _git(repository_path, "rev-parse", "HEAD") source_path.write_text("candidate source\n", encoding="utf-8") _git(repository_path, "add", ".") _git(repository_path, "commit", "-m", "candidate") target_revision = _git(repository_path, "rev-parse", "HEAD") repository_git_dir = repository_path / ".git" source_dir = tmp_path / "source" deploy.sync_source(repository_git_dir, prior_revision, source_dir, "projectbrain") private_env_path = source_dir / ".env" private_env_path.write_text("PROJECTBRAIN_PRIVATE_TEST=fixture-only\n", encoding="utf-8") private_env_before = private_env_path.read_bytes() project = { "slug": "projectbrain", "branch": "main", "source_dir": str(source_dir), "deploy_script": "scripts/deploy-unraid-single.sh", "containers": ["projectbrain"], "compose_project": "projectbrain", "compose_files": ["docker-compose.yml"], } previous_state = { "schema_version": 3, "repository": REPOSITORY, "live_revision": prior_revision, "live_status": "healthy", "live_legacy_provenance": False, "live_artifacts": {"containers": {"projectbrain": {"image_id": "sha256:" + "a" * 64}}}, "live_receipt": {"revision": prior_revision, "testOnly": True}, } active_locks = [] project_lock = deploy.STATE_ROOT / "locks" / "projectbrain.lock" @contextmanager def lock(path, _description): active_locks.append(path) try: yield finally: active_locks.pop() monkeypatch.setattr(deploy, "exclusive_lock", lock) monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: lock( deploy.GLOBAL_DEPLOY_LOCK, "global", )) monkeypatch.setattr(deploy, "load_projects", lambda: {REPOSITORY: dict(project)}) monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, config: config) monkeypatch.setattr(deploy, "repository_path", lambda *_a: repository_git_dir) monkeypatch.setattr(deploy, "repository_path_from_project", lambda _project: repository_git_dir) monkeypatch.setattr(deploy, "bind_appops_self_upgrade_recovery", lambda *_a: None) monkeypatch.setattr(deploy, "bind_registered_legacy_request", lambda *_a: None) monkeypatch.setattr(deploy, "mcp_hub_prepare_current_receipt_model", lambda *_a: None) monkeypatch.setattr(deploy, "validate_commit", lambda *_a: target_revision) monkeypatch.setattr(deploy, "deployment_metadata", lambda *_a: { "revision": target_revision, "tree": "f" * 40, }) monkeypatch.setattr(deploy, "validate_project_config", lambda *_a: None) monkeypatch.setattr(deploy, "read_state", lambda _project: dict(previous_state)) monkeypatch.setattr(deploy, "verified_current_receipt", lambda *_a: None) monkeypatch.setattr(deploy, "validate_forward_revision", lambda *_a: None) monkeypatch.setattr(deploy, "_mcp_hub_assert_no_unresolved_migration_intent", lambda *_a: None) source_syncs = [] actual_sync_source = deploy.sync_source def record_sync(repo, revision, target, slug, clean_untracked_paths=None, *, exact=False): source_syncs.append((revision, tuple(active_locks))) if restore_fails and revision == prior_revision: raise deploy.DeployError("fixture source restore failure") return actual_sync_source( repo, revision, target, slug, clean_untracked_paths, exact=exact, ) monkeypatch.setattr(deploy, "sync_source", record_sync) environment_copies = [] actual_copy_environment = deploy.copy_environment def record_copy_environment(config, target): environment_copies.append((tuple(active_locks), config.get("env_source"))) return actual_copy_environment(config, target) monkeypatch.setattr(deploy, "copy_environment", record_copy_environment) gate_calls = [] gate_results = iter((True, False)) def gate(_repository, _revision, _fingerprint): gate_calls.append(tuple(active_locks)) return next(gate_results) monkeypatch.setitem(sys.modules, "status_server", SimpleNamespace( _projectbrain_delivery_deploy_gate=gate, )) state_updates = [] monkeypatch.setattr( deploy, "write_state", lambda repository, config, revision, status, message, **kwargs: state_updates.append({ "repository": repository, "project": config, "revision": revision, "status": status, "message": message, **kwargs, }), ) monkeypatch.setattr( deploy, "capture_compose_rollback", lambda *_a: pytest.fail("production snapshot must not run after a stale second gate"), ) monkeypatch.setattr( deploy, "deploy_project", lambda *_a: pytest.fail("production mutation must not run after a stale second gate"), ) monkeypatch.setenv(deploy.PROJECTBRAIN_DELIVERY_PREFLIGHT_ENV, "b" * 64) monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", REPOSITORY, target_revision]) with pytest.raises(deploy.DeployError, match="preflight changed before deployment"): deploy.main() assert len(gate_calls) == 2 assert gate_calls == [ (project_lock, deploy.GLOBAL_DEPLOY_LOCK), (project_lock, deploy.GLOBAL_DEPLOY_LOCK), ] assert source_syncs == [ (target_revision, (project_lock,)), (prior_revision, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)), ] assert environment_copies == [((project_lock,), None)] + ( [] if restore_fails else [((project_lock, deploy.GLOBAL_DEPLOY_LOCK), None)] ) assert source_dir.joinpath("README.md").read_text(encoding="utf-8") == ( "candidate source\n" if restore_fails else "prior live source\n" ) assert private_env_path.read_bytes() == private_env_before assert len(state_updates) == 1 state = state_updates[0] assert state["status"] == "failed" assert state["live_revision"] == prior_revision assert state["live_artifacts"] == previous_state["live_artifacts"] assert state["live_receipt"] == previous_state["live_receipt"] assert state["live_status"] == ("unknown" if restore_fails else "healthy") if restore_fails: assert "source rollback error: fixture source restore failure" in state["message"] def test_snapshot_lock_is_nonblocking_and_uses_project_then_global_flock( controller, monkeypatch, tmp_path, ): project_lock = tmp_path / "project.lock" global_lock = tmp_path / "global.lock" controller.PROJECTBRAIN_DELIVERY_PROJECT_LOCK_PATH = project_lock controller.PROJECTBRAIN_DELIVERY_LOCK_PATH = global_lock descriptors = {11: project_lock, 12: global_lock} opened = [] acquired = [] released = [] closed = [] busy = False fake_os = SimpleNamespace( name="posix", O_RDWR=1, O_NOFOLLOW=2, open=lambda path, _flags: _open_fake(path, descriptors, opened), fstat=lambda _descriptor: SimpleNamespace( st_mode=0o100600, st_nlink=1, st_uid=0, ), geteuid=lambda: 0, close=lambda descriptor: closed.append(descriptor), ) def flock(descriptor, flags): if flags == 6 and busy and descriptors[descriptor] == global_lock: raise BlockingIOError if flags == 6: acquired.append(descriptors[descriptor]) elif flags == 8: released.append(descriptors[descriptor]) monkeypatch.setattr(controller, "os", fake_os) monkeypatch.setitem(sys.modules, "fcntl", SimpleNamespace( LOCK_EX=2, LOCK_NB=4, LOCK_UN=8, flock=flock, )) with controller._projectbrain_delivery_mutation_lock(), pytest.raises( controller.ProjectBrainDeliveryRefused, ) as thread_busy: controller._projectbrain_delivery_mutation_lock().__enter__() assert thread_busy.value.code == "projectbrain_delivery_operation_busy" assert opened == [project_lock, global_lock] assert acquired == [project_lock, global_lock] assert released == [global_lock, project_lock] assert closed == [12, 11] busy = True opened.clear() acquired.clear() released.clear() closed.clear() with pytest.raises(controller.ProjectBrainDeliveryRefused) as process_busy: controller._projectbrain_delivery_mutation_lock().__enter__() assert process_busy.value.code == "projectbrain_deployment_active" assert opened == [project_lock, global_lock] assert acquired == [project_lock] assert released == [project_lock] assert closed == [12, 11] def test_deployment_broker_client_and_mcp_tool_keep_fixed_projectbrain_contract(monkeypatch): from backend.app import mcp_server from backend.app.services.deployment_broker import DeploymentBrokerClient, settings monkeypatch.setattr(settings, "mcp_token", "t" * 48) client = DeploymentBrokerClient(base_url="https://controller.example", verify_tls=True) requests = [] monkeypatch.setattr(client, "_request", lambda *args, **kwargs: requests.append((args, kwargs)) or {"status": "ready"}) assert client.projectbrain_delivery_operation("preflight", {"revision": REVISION}) == {"status": "ready"} assert requests[-1][0][:2] == ("POST", PREFIX + "preflight") assert requests[-1][1]["payload"] == {"revision": REVISION} revalidate_payload = { "revision": REVISION, "expectedPreflightSha256": PREFLIGHT_SHA, "idempotencyKey": "pb-revalidate-2026-10-07", } assert client.projectbrain_delivery_operation("revalidate", revalidate_payload) == {"status": "ready"} assert requests[-1][0][:2] == ("POST", PREFIX + "revalidate") assert requests[-1][1]["payload"] == revalidate_payload with pytest.raises(ValueError): client.projectbrain_delivery_operation("preflight", {"revision": REVISION, "repository": REPOSITORY}) tool_calls = [] monkeypatch.setattr( mcp_server, "DeploymentBrokerClient", lambda: SimpleNamespace(projectbrain_delivery_operation=lambda stage, payload: tool_calls.append((stage, payload)) or {"status": "ready"}), ) assert mcp_server.appops_projectbrain_delivery_preflight(REVISION) == {"status": "ready"} with pytest.raises(ValueError): mcp_server.appops_projectbrain_delivery_revalidate( REVISION, PREFLIGHT_SHA, "short", ) assert mcp_server.appops_projectbrain_delivery_revalidate( REVISION, PREFLIGHT_SHA, "pb-revalidate-2026-10-07", ) == {"status": "ready"} assert tool_calls == [ ("preflight", {"revision": REVISION}), ("revalidate", { "revision": REVISION, "expectedPreflightSha256": PREFLIGHT_SHA, "idempotencyKey": "pb-revalidate-2026-10-07", }), ] def _open_fake(path, descriptor_paths, opened): descriptor = 11 if not opened else 12 opened.append(Path(path)) descriptor_paths[descriptor] = Path(path) return descriptor