from __future__ import annotations import hashlib import importlib.util import json import multiprocessing import os import stat import sys import threading from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[2] STORE_PATH = REPO_ROOT / "ops" / "gitea-deploy-broker" / "repopulse_adoption_store.py" RESTORE_PATH = REPO_ROOT / "ops" / "gitea-deploy-broker" / "repopulse_registry_restore.py" def _load_store(): spec = importlib.util.spec_from_file_location("repopulse_adoption_store_test", STORE_PATH) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def _load_registry_restore(store_module, monkeypatch): monkeypatch.setitem(sys.modules, "repopulse_adoption_store", store_module) spec = importlib.util.spec_from_file_location( "repopulse_registry_restore_test", RESTORE_PATH ) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def _sha(value: str) -> str: return hashlib.sha256(value.encode("ascii")).hexdigest() def _second_process_lock(path_text, ready, attempt, attempting, acquired): import fcntl # Separate process startup from the lock attempt. In particular, the # parent must not hold the project lock while waiting for multiprocessing # to import this test module (which can be slow under forkserver on /mnt/c). ready.set() if not attempt.wait(15): return attempting.set() with open(path_text, "a+") as handle: fcntl.flock(handle, fcntl.LOCK_EX) acquired.set() fcntl.flock(handle, fcntl.LOCK_UN) def test_intent_binding_matches_appops_canonical_digest() -> None: module = _load_store() idempotency = _sha("repopulse-adoption-key") intent_id = f"rpa-{idempotency[:20]}" preflight = _sha("preflight-projection") binding = { "intentId": intent_id, "repository": "jens/repopulse", "applicationId": "repopulse", "preflightSha256": preflight, "idempotencySha256": idempotency, } expected = hashlib.sha256( json.dumps(binding, sort_keys=True, separators=(",", ":")).encode("ascii") ).hexdigest() assert module.intent_binding_sha256(intent_id, idempotency, preflight) == expected assert module._document_sha256(binding) != expected @pytest.fixture def store_env(tmp_path, monkeypatch): module = _load_store() controller = tmp_path / "controller" controller.mkdir() controller.chmod(0o755) (controller / "projects.json").write_text( json.dumps( { "jens/repopulse": { "branch": "main", "compose_files": ["docker-compose.yml"], "compose_project": "repopulse", "containers": ["repopulse"], "env_source": "/srv/example/appdata/repopulse/deploy.env", "exact_source": True, "gitea_repository_id": 64, "gitea_webhook_secret_file": "jens__repopulse", "health_timeout": 180, "slug": "repopulse", "source_dir": "/mnt/cache/appdata/gitea-deploy-runners/sources/repopulse", "validation": { "contexts": ["Managed validation / full (push)"], "max_poll_interval_seconds": 30, "poll_interval_seconds": 5, "required": True, "timeout_seconds": 1800, "validation_gate_version": 1, }, "verify_oci_revision": False, }, "jens/other": {"slug": "other"}, }, sort_keys=True, ) + "\n", encoding="utf-8", ) (controller / "projects.json").chmod(0o600) secrets = controller / "appops-deployment-secrets" secrets.mkdir() secrets.chmod(0o700) state = tmp_path / "state" state.mkdir() state.chmod(0o755) locks = state / "locks" locks.mkdir() locks.chmod(0o755) global_lock = tmp_path / "deploy.lock" state_store = state / "repopulse-adoption" monkeypatch.setattr(module, "CONTROLLER_READONLY_ROOT", controller) monkeypatch.setattr(module, "CONTROLLER_RW_ROOT", controller) monkeypatch.setattr(module, "STATE_ROOT", state) monkeypatch.setattr(module, "GLOBAL_DEPLOY_LOCK", global_lock) monkeypatch.setattr(module, "PROJECT_LOCK", locks / "repopulse.lock") monkeypatch.setattr(module, "ADOPTION_STATE_ROOT", state_store) monkeypatch.setattr(module, "EXECUTION_ENABLED", True) monkeypatch.setattr(module, "_IS_POSIX_CONTROLLER", True) test_uid = controller.stat().st_uid monkeypatch.setattr(module, "_ROOT_UID", test_uid) monkeypatch.setattr(module, "_effective_uid", lambda: test_uid) if os.name != "posix": monkeypatch.setattr( module, "_mode_bits", lambda path, info: ( 0o700 if path.name in {"appops-deployment-secrets", "repopulse-adoption"} else 0o755 if stat.S_ISDIR(info.st_mode) else 0o600 ), ) monkeypatch.setattr(module, "_directory_fsync", lambda _path: None) return module, module.RepoPulseAdoptionStore(), controller, state_store def _reserve(store, *, key: str = "adoption-key-2026", preflight: str = "f"): key_sha = hashlib.sha256( f"itworx.repopulse-legacy-adoption/v1:jens/repopulse:{key}".encode("ascii") ).hexdigest() intent_id = f"rpa-{key_sha[:20]}" return store.reserve_intent( intent_id=intent_id, idempotency_sha256=key_sha, preflight_sha256=_sha(preflight), ) def _to_snapshot_verified(store): intent = _reserve(store) intent = store.record_target_gates( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), profile_id="repopulse-managed-v2", plan_sha256=_sha("managed profile"), ) intent = store.record_quiescence( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), evidence_sha256=_sha("maintenance and queue drained"), ) return store.record_snapshot_verified( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), snapshot_file_sha256=_sha("online sqlite backup"), snapshot_logical_sha256=_sha("database logical digest"), snapshot_queue_sha256=_sha("empty queue digest"), ) def test_execution_can_be_hard_disabled_without_effects(monkeypatch): module = _load_store() monkeypatch.setattr(module, "EXECUTION_ENABLED", False) assert module.execution_status()["enabled"] is False assert module.execution_status()["reason"] == "controller_adoption_execution_disabled" assert module.execution_status()["hasRoute"] is True assert module.execution_status()["hasEffectfulRoute"] is False def test_capability_report_is_sanitized_and_does_not_create_state_or_locks( store_env, monkeypatch ): module, _store, controller, state_root = store_env monkeypatch.setattr(module, "EXECUTION_ENABLED", False) credential = b"private-capability-test-secret-value" * 2 credential_path = controller / "appops-deployment-secrets" / "jens__repopulse" credential_path.write_bytes(credential) credential_path.chmod(0o600) report = module.capability_report() assert report == { "repository": "jens/repopulse", "executionEnabled": False, "controllerAliasesReady": True, "registryReady": True, "secretRootReady": True, } assert credential.decode("ascii") not in json.dumps(report) assert not state_root.exists() assert not module.PROJECT_LOCK.exists() assert not module.GLOBAL_DEPLOY_LOCK.exists() def test_capability_report_only_exposes_registry_readiness_on_identity_drift(store_env): module, _store, controller, _state_root = store_env projects_path = controller / "projects.json" projects = json.loads(projects_path.read_text(encoding="utf-8")) projects["jens/repopulse"]["source_dir"] = "/tmp/private-value" projects_path.write_text(json.dumps(projects), encoding="utf-8") projects_path.chmod(0o600) report = module.capability_report() assert report["controllerAliasesReady"] is True assert report["registryReady"] is False assert report["secretRootReady"] is True assert "/tmp/private-value" not in json.dumps(report) def test_read_only_status_when_unreserved_does_not_create_state_or_locks( store_env, monkeypatch ): module, store, _controller, _state = store_env monkeypatch.setattr(module, "EXECUTION_ENABLED", False) assert store.read_only_status() is None assert not module.ADOPTION_STATE_ROOT.exists() assert not module.PROJECT_LOCK.exists() assert not module.GLOBAL_DEPLOY_LOCK.exists() def test_read_only_status_fails_closed_on_receipt_without_intent(store_env, monkeypatch): module, store, _controller, state_root = store_env monkeypatch.setattr(module, "EXECUTION_ENABLED", False) state_root.mkdir(mode=0o700) state_root.chmod(0o700) idempotency_sha256 = _sha("orphan receipt idempotency") intent_id = f"rpa-{idempotency_sha256[:20]}" receipt = { "apiVersion": module.RECEIPT_SCHEMA, "intentId": intent_id, "repository": module.REPOSITORY, "applicationId": module.APPLICATION_ID, "phase": "registry_committed", "intentBindingSha256": module.intent_binding_sha256( intent_id, idempotency_sha256, _sha("orphan receipt preflight") ), "legacyVolumeRetained": True, "intentDurable": True, "managedEnrollmentId": "enr-" + "a" * 20, "projectsSha256": _sha("orphan receipt projects"), "snapshotFileSha256": _sha("orphan receipt snapshot"), } receipt["receiptSha256"] = module._document_sha256(receipt) receipt_path = state_root / "receipt.json" receipt_path.write_bytes(module._canonical_json(receipt)) receipt_path.chmod(0o600) receipt_before = receipt_path.read_bytes() with pytest.raises( module.AdoptionStoreError, match="controller_receipt_without_intent" ): store.read_only_status() assert receipt_path.read_bytes() == receipt_before assert not (state_root / "intent.json").exists() assert {path.name for path in state_root.iterdir()} == {"receipt.json"} assert not module.PROJECT_LOCK.exists() assert not module.GLOBAL_DEPLOY_LOCK.exists() def test_registry_write_guard_does_not_mark_an_execution_transaction( store_env, monkeypatch ): module, store, _controller, _state_root = store_env monkeypatch.setattr(module, "EXECUTION_ENABLED", False) identity = module._lock_identity() with store.registry_write_guard(): assert module._transaction_owner(identity) is None assert not module.ADOPTION_STATE_ROOT.exists() assert module._transaction_owner(identity) is None assert not module.ADOPTION_STATE_ROOT.exists() def test_read_only_status_returns_only_sanitized_hashes_and_phase(store_env): _module, store, controller, _state = store_env credential = b"private-status-test-credential-value" * 2 credential_path = controller / "appops-deployment-secrets" / "jens__repopulse" credential_path.write_bytes(credential) credential_path.chmod(0o600) intent = _reserve(store) status = store.read_only_status() assert status is not None assert status["status"] == "in_progress" assert status["phase"] == "reserved" assert status["hashes"]["intentSha256"] == store.intent_sha256(intent) assert "expectedTargetCredentialSha256" not in json.dumps(status) assert hashlib.sha256(credential).hexdigest() not in json.dumps(status) assert credential.decode("ascii") not in json.dumps(status) def test_read_only_status_fails_closed_on_unsafe_state_root(store_env): module, store, _controller, _state = store_env module.ADOPTION_STATE_ROOT.write_text("unsafe state", encoding="utf-8") with pytest.raises(module.AdoptionStoreError, match="controller_path_unsafe"): store.read_only_status() assert module.ADOPTION_STATE_ROOT.read_text(encoding="utf-8") == "unsafe state" def test_store_reserves_one_fixed_idempotent_intent(store_env): module, store, _controller, _state = store_env first = _reserve(store) replay = _reserve(store) assert replay == first assert first["repository"] == "jens/repopulse" assert first["applicationId"] == "repopulse" assert first["legacyVolume"] == "repopulse_data" assert first["adoptionAuthorized"] is False assert first["phase"] == "reserved" with pytest.raises(module.AdoptionStoreError, match="one_time_adoption_already_reserved"): _reserve(store, key="different-adoption-key-2026") def test_reservation_rejects_legacy_registry_identity_drift(store_env): module, store, controller, _state = store_env projects_path = controller / "projects.json" projects = json.loads(projects_path.read_text(encoding="utf-8")) projects["jens/repopulse"]["source_dir"] = "/tmp/attacker" projects_path.write_text(json.dumps(projects), encoding="utf-8") projects_path.chmod(0o600) with pytest.raises(module.AdoptionStoreError, match="legacy_registry_identity_mismatch"): _reserve(store) def test_reservation_rejects_unrecognized_legacy_registry_fields(store_env): module, store, controller, _state = store_env projects_path = controller / "projects.json" projects = json.loads(projects_path.read_text(encoding="utf-8")) projects["jens/repopulse"]["unexpected_override"] = True projects_path.write_text(json.dumps(projects), encoding="utf-8") projects_path.chmod(0o600) with pytest.raises(module.AdoptionStoreError, match="legacy_registry_identity_mismatch"): _reserve(store) def test_reservation_rejects_boolean_integer_alias_in_legacy_gate(store_env): module, store, controller, _state = store_env projects_path = controller / "projects.json" projects = json.loads(projects_path.read_text(encoding="utf-8")) projects["jens/repopulse"]["validation"]["validation_gate_version"] = True projects_path.write_text(json.dumps(projects), encoding="utf-8") projects_path.chmod(0o600) with pytest.raises(module.AdoptionStoreError, match="legacy_registry_identity_mismatch"): _reserve(store) def test_store_rejects_read_write_controller_alias_mismatch(store_env, monkeypatch, tmp_path): module, store, _controller, _state = store_env other = tmp_path / "other-controller" other.mkdir() other.chmod(0o755) monkeypatch.setattr(module, "CONTROLLER_RW_ROOT", other) with pytest.raises(module.AdoptionStoreError, match="controller_alias_mismatch"): _reserve(store) def test_store_rejects_controller_registry_with_unsafe_mode(store_env, monkeypatch): module, store, _controller, _state = store_env original = module._mode_bits def unsafe_projects_mode(path, info): if path.name == "projects.json": return 0o666 return original(path, info) monkeypatch.setattr(module, "_mode_bits", unsafe_projects_mode) with pytest.raises(module.AdoptionStoreError, match="controller_file_mode_invalid"): _reserve(store) def test_store_rejects_symlinked_controller_root(store_env, monkeypatch, tmp_path): module, store, controller, _state = store_env alias = tmp_path / "controller-alias" try: alias.symlink_to(controller, target_is_directory=True) except OSError as exc: pytest.skip(f"creating a symlink is unavailable on this Windows host: {exc}") monkeypatch.setattr(module, "CONTROLLER_RW_ROOT", alias) with pytest.raises(module.AdoptionStoreError, match="controller_path_unsafe"): _reserve(store) def test_checkpoint_order_and_idempotent_replay_are_digest_bound(store_env): module, store, _controller, _state = store_env intent = _reserve(store) with pytest.raises(module.AdoptionStoreError, match="adoption_step_out_of_order"): store.record_quiescence( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), evidence_sha256=_sha("queue drained"), ) first = store.record_target_gates( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), profile_id="repopulse-managed-v2", plan_sha256=_sha("profile"), ) replay = store.record_target_gates( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(first), profile_id="repopulse-managed-v2", plan_sha256=_sha("profile"), ) assert replay == first with pytest.raises(module.AdoptionStoreError, match="idempotency_conflict"): store.record_target_gates( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(first), profile_id="repopulse-managed-v2", plan_sha256=_sha("changed profile"), ) def test_registry_and_fixed_secret_registration_are_atomic_and_redacted(store_env): module, store, controller, state_root = store_env snapshot = _to_snapshot_verified(store) credential = b"target-token-" + b"x" * 40 receipt = store.register_target( intent_id=snapshot["intentId"], expected_intent_sha256=store.intent_sha256(snapshot), enrollment_id="enr-0123456789abcdef0123", target_credential=credential, ) projects = json.loads((controller / "projects.json").read_text(encoding="utf-8")) target = projects["jens/repopulse"] assert target["managed_profile"] == "repopulse-authenticated" assert target["managed_application_id"] == "repopulse" assert target["managed_runtime_uid"] == target["managed_runtime_gid"] == 10001 assert target["managed_proxy_hostname"] == "repopulse.example.com" assert target["appops_deployment_secret_file"] == "jens__repopulse" assert target["compose_files"] == [".itworx/deployment.yml"] assert target["gitea_repository_id"] == 64 assert target["gitea_webhook_secret_file"] == "jens__repopulse" assert target["health_timeout"] == 180 assert target["validation"] == { "contexts": ["Managed validation / full (push)"], "max_poll_interval_seconds": 30, "poll_interval_seconds": 5, "required": True, "timeout_seconds": 1800, "validation_gate_version": 1, } assert projects["jens/other"] == {"slug": "other"} assert (controller / "appops-deployment-secrets" / "jens__repopulse").read_bytes() == credential assert module._mode_bits( controller / "appops-deployment-secrets", (controller / "appops-deployment-secrets").stat(), ) == 0o700 assert module._mode_bits( controller / "appops-deployment-secrets" / "jens__repopulse", (controller / "appops-deployment-secrets" / "jens__repopulse").stat(), ) == 0o600 assert receipt["phase"] == "registry_committed" assert receipt["legacyVolumeRetained"] is True assert credential.decode("ascii") not in json.dumps(receipt) intent_bytes = (state_root / "intent.json").read_text(encoding="utf-8") receipt_bytes = (state_root / "receipt.json").read_text(encoding="utf-8") assert credential.decode("ascii") not in intent_bytes + receipt_bytes def test_registration_replay_is_idempotent_after_receipt_is_durable(store_env): _module, store, _controller, _state = store_env snapshot = _to_snapshot_verified(store) credential = b"target-token-" + b"x" * 40 first = store.register_target( intent_id=snapshot["intentId"], expected_intent_sha256=store.intent_sha256(snapshot), enrollment_id="enr-0123456789abcdef0123", target_credential=credential, ) current_intent = store.load_intent() assert current_intent is not None replay = store.register_target( intent_id=current_intent["intentId"], expected_intent_sha256=store.intent_sha256(current_intent), enrollment_id="enr-0123456789abcdef0123", target_credential=credential, ) assert replay == first def test_registry_digest_compare_and_swap_rejects_drift_without_mutating_secret(store_env): module, store, controller, _state = store_env snapshot = _to_snapshot_verified(store) path = controller / "projects.json" projects = json.loads(path.read_text(encoding="utf-8")) projects["jens/other"]["changed"] = True path.write_text(json.dumps(projects), encoding="utf-8") path.chmod(0o600) with pytest.raises(module.AdoptionStoreError, match="projects_digest_conflict"): store.register_target( intent_id=snapshot["intentId"], expected_intent_sha256=store.intent_sha256(snapshot), enrollment_id="enr-0123456789abcdef0123", target_credential=b"target-token-" + b"x" * 40, ) assert not (controller / "appops-deployment-secrets" / "jens__repopulse").exists() def test_existing_nonmatching_secret_is_never_overwritten(store_env): module, store, controller, _state = store_env snapshot = _to_snapshot_verified(store) credential_path = controller / "appops-deployment-secrets" / "jens__repopulse" credential_path.write_bytes(b"existing-other-secret-value-not-for-adoption") credential_path.chmod(0o600) with pytest.raises(module.AdoptionStoreError, match="target_credential_conflict"): store.register_target( intent_id=snapshot["intentId"], expected_intent_sha256=store.intent_sha256(snapshot), enrollment_id="enr-0123456789abcdef0123", target_credential=b"target-token-" + b"x" * 40, ) assert credential_path.read_bytes() == b"existing-other-secret-value-not-for-adoption" def test_registration_replay_recovers_after_registry_commit_receipt_failure(store_env, monkeypatch): module, store, controller, _state = store_env snapshot = _to_snapshot_verified(store) credential = b"target-token-" + b"x" * 40 original = store._write_receipt_cas def fail_once(*args, **kwargs): raise OSError("simulated receipt fsync interruption") monkeypatch.setattr(store, "_write_receipt_cas", fail_once) with pytest.raises(module.AdoptionStoreError, match="registration_receipt_write_failed"): store.register_target( intent_id=snapshot["intentId"], expected_intent_sha256=store.intent_sha256(snapshot), enrollment_id="enr-0123456789abcdef0123", target_credential=credential, ) assert json.loads((controller / "projects.json").read_text(encoding="utf-8"))["jens/repopulse"]["managed_application_id"] == "repopulse" assert (controller / "appops-deployment-secrets" / "jens__repopulse").read_bytes() == credential monkeypatch.setattr(store, "_write_receipt_cas", original) current_intent = store.load_intent() assert current_intent is not None receipt = store.register_target( intent_id=current_intent["intentId"], expected_intent_sha256=store.intent_sha256(current_intent), enrollment_id="enr-0123456789abcdef0123", target_credential=credential, ) assert store.load_receipt()["receiptSha256"] == receipt["receiptSha256"] def test_registration_replay_recovers_after_secret_only_partial_commit(store_env, monkeypatch): module, store, controller, _state = store_env snapshot = _to_snapshot_verified(store) credential = b"target-token-" + b"x" * 40 original = module._atomic_replace_cas failed = False def fail_projects_once(path, payload, **kwargs): nonlocal failed if path.name == "projects.json" and not failed: failed = True raise module.AdoptionStoreError("simulated_registry_replace_failure") return original(path, payload, **kwargs) monkeypatch.setattr(module, "_atomic_replace_cas", fail_projects_once) with pytest.raises(module.AdoptionStoreError, match="projects_registry_write_failed"): store.register_target( intent_id=snapshot["intentId"], expected_intent_sha256=store.intent_sha256(snapshot), enrollment_id="enr-0123456789abcdef0123", target_credential=credential, ) assert json.loads((controller / "projects.json").read_text(encoding="utf-8"))["jens/repopulse"]["env_source"] == "/srv/example/appdata/repopulse/deploy.env" assert (controller / "appops-deployment-secrets" / "jens__repopulse").read_bytes() == credential monkeypatch.setattr(module, "_atomic_replace_cas", original) current_intent = store.load_intent() assert current_intent is not None receipt = store.register_target( intent_id=current_intent["intentId"], expected_intent_sha256=store.intent_sha256(current_intent), enrollment_id="enr-0123456789abcdef0123", target_credential=credential, ) assert receipt["phase"] == "registry_committed" def test_recovery_marker_is_durable_and_digest_bound(store_env): module, store, _controller, _state = store_env intent = _to_snapshot_verified(store) marked = store.mark_recovery_required( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), reason="operator_interruption", failure_evidence_sha256=_sha("operator interruption marker"), ) assert marked["status"] == "recovery_required" assert store.load_intent() == marked with pytest.raises(module.AdoptionStoreError, match="idempotency_conflict"): store.mark_recovery_required( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(marked), reason="operator_interruption", failure_evidence_sha256=_sha("different interruption marker"), ) def test_lock_order_is_project_then_global(store_env, monkeypatch): module, store, _controller, _state = store_env observed = [] class FakeLock: def __init__(self, path): self.path = path def __enter__(self): observed.append(self.path) return self def __exit__(self, *_args): return False monkeypatch.setattr(module, "_exclusive_flock", lambda path: FakeLock(path)) _reserve(store) assert observed == [module.PROJECT_LOCK, module.GLOBAL_DEPLOY_LOCK] def test_transaction_holds_each_lock_once_and_rechecks_aliases_under_lock(store_env, monkeypatch): module, store, _controller, _state = store_env another_store = module.RepoPulseAdoptionStore() events = [] original_alias_check = module._check_controller_aliases class FakeLock: def __init__(self, path): self.path = path def __enter__(self): events.append(("lock", self.path)) return self def __exit__(self, *_args): return False def check_aliases(): events.append(("aliases", None)) return original_alias_check() monkeypatch.setattr(module, "_exclusive_flock", lambda path: FakeLock(path)) monkeypatch.setattr(module, "_check_controller_aliases", check_aliases) with store.transaction(): assert store.load_intent() is None assert another_store.load_receipt() is None assert events == [ ("lock", module.PROJECT_LOCK), ("lock", module.GLOBAL_DEPLOY_LOCK), ("aliases", None), ("aliases", None), ("aliases", None), ] def test_deploy_lock_reuse_is_only_visible_after_both_flocks_are_acquired(store_env): module, store, _controller, _state = store_env unrelated = module.STATE_ROOT / "locks" / "other.lock" assert not module.transaction_lock_held(module.PROJECT_LOCK) assert not module.transaction_lock_held(module.GLOBAL_DEPLOY_LOCK) with store.transaction(): assert module.transaction_lock_held(module.PROJECT_LOCK) assert module.transaction_lock_held(module.GLOBAL_DEPLOY_LOCK) assert not module.transaction_lock_held(unrelated) assert not module.transaction_lock_held(module.PROJECT_LOCK) def test_nested_transaction_is_rejected_without_losing_outer_lock(store_env): module, store, _controller, _state = store_env another_store = module.RepoPulseAdoptionStore() with store.transaction(): for nested_store in (store, another_store): with pytest.raises( module.AdoptionStoreError, match="controller_transaction_nested" ), nested_store.transaction(): pytest.fail("nested transaction must not enter") assert store.load_intent() is None def test_transaction_owner_is_released_after_alias_validation_failure(store_env, monkeypatch): module, store, _controller, _state = store_env original_alias_check = module._check_controller_aliases calls = 0 def fail_once(): nonlocal calls calls += 1 if calls == 1: raise module.AdoptionStoreError("controller_alias_mismatch") return original_alias_check() monkeypatch.setattr(module, "_check_controller_aliases", fail_once) with pytest.raises( module.AdoptionStoreError, match="controller_alias_mismatch" ), store.transaction(): pytest.fail("transaction must not yield after alias validation failure") with store.transaction(): assert store.load_intent() is None def test_transaction_rejects_other_thread_store_calls_and_transactions(store_env): module, store, _controller, _state = store_env another_store = module.RepoPulseAdoptionStore() entered = threading.Event() release = threading.Event() failures = [] def hold_transaction(): try: with store.transaction(): try: with store.transaction(): pass except module.AdoptionStoreError as exc: if exc.code != "controller_transaction_nested": failures.append(exc) else: failures.append("nested transaction unexpectedly entered") entered.set() if not release.wait(5): failures.append("transaction release timed out") except (AssertionError, module.AdoptionStoreError) as exc: failures.append(exc) thread = threading.Thread(target=hold_transaction) thread.start() try: assert entered.wait(2) with pytest.raises( module.AdoptionStoreError, match="controller_transaction_thread_mismatch" ): another_store.load_intent() with pytest.raises( module.AdoptionStoreError, match="controller_transaction_thread_mismatch" ), another_store.transaction(): pytest.fail("other thread must not enter transaction") finally: release.set() thread.join(timeout=5) assert not thread.is_alive() assert failures == [] @pytest.mark.skipif(os.name != "posix", reason="cross-process POSIX flock test") def test_project_flock_blocks_a_second_process(store_env): module, store, _controller, _state = store_env ready = multiprocessing.Event() attempt = multiprocessing.Event() attempting = multiprocessing.Event() acquired = multiprocessing.Event() child = multiprocessing.Process( target=_second_process_lock, args=(str(module.PROJECT_LOCK), ready, attempt, attempting, acquired), ) child.start() try: # Give process creation/import a bounded interval while no lock is held. assert ready.wait(15), f"lock contender did not start (exit={child.exitcode})" with store.transaction(): assert store.load_intent() is None attempt.set() assert attempting.wait(5), ( f"lock contender did not attempt acquisition (exit={child.exitcode})" ) assert not acquired.wait(0.1) assert acquired.wait(5) child.join(timeout=5) assert child.exitcode == 0 finally: # Let a child blocked on the handshake exit on assertion failures too. attempt.set() child.join(timeout=5) if child.is_alive(): child.terminate() child.join(timeout=5) def _registered_recovery(store, restore, *, preserve=True, reason="target_health_failed"): intent = _to_snapshot_verified(store) if preserve: with store.transaction(): restore.preserve_before_registry_cas(store) with store.transaction(): store.register_target( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), enrollment_id="enr-0123456789abcdefabcd", target_credential=b"R" * 64, ) intent = store.load_intent() return store.mark_recovery_required( intent_id=intent["intentId"], expected_intent_sha256=store.intent_sha256(intent), reason=reason, failure_evidence_sha256=_sha("managed cutover health failed"), ) def test_registry_snapshot_preserves_exact_old_bytes_only_inside_store_transaction( store_env, monkeypatch ): store_module, store, controller, state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) intent = _to_snapshot_verified(store) original = (controller / "projects.json").read_bytes() with pytest.raises(store_module.AdoptionStoreError, match="restore_transaction_required"): restore.preserve_before_registry_cas(store) with store.transaction(): first = restore.preserve_before_registry_cas(store) replay = restore.preserve_before_registry_cas(store) snapshot_path = state_root / "projects-before.bin" receipt_path = state_root / "registry-snapshot.json" assert snapshot_path.read_bytes() == original assert store_module._mode_bits(snapshot_path, snapshot_path.stat()) == 0o600 assert store_module._mode_bits(receipt_path, receipt_path.stat()) == 0o600 assert first == replay assert first["intentId"] == intent["intentId"] assert first["sourceProjectsSha256"] == hashlib.sha256(original).hexdigest() assert first["snapshotSha256"] == hashlib.sha256(original).hexdigest() assert first["snapshotBytes"] == len(original) assert b"R" * 64 not in receipt_path.read_bytes() def test_registry_snapshot_rejects_source_digest_drift(store_env, monkeypatch): store_module, store, controller, state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) _to_snapshot_verified(store) projects_path = controller / "projects.json" projects = json.loads(projects_path.read_text(encoding="utf-8")) projects["jens/other"]["display"] = "changed while preparing" projects_path.write_text(json.dumps(projects, sort_keys=True) + "\n", encoding="utf-8") projects_path.chmod(0o600) with store.transaction(), pytest.raises( store_module.AdoptionStoreError, match="projects_digest_conflict" ): restore.preserve_before_registry_cas(store) assert not state_root.exists() or not (state_root / "projects-before.bin").exists() def test_registry_restore_reverts_only_exact_managed_target_and_replays_idempotently( store_env, monkeypatch ): store_module, store, controller, _state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) original = (controller / "projects.json").read_bytes() intent = _registered_recovery(store, restore) assert intent["status"] == "recovery_required" with pytest.raises(store_module.AdoptionStoreError, match="restore_transaction_required"): restore.restore_after_failed_managed_cutover(store) with store.transaction(): assert store_module.transaction_lock_held(store_module.PROJECT_LOCK) assert store_module.transaction_lock_held(store_module.GLOBAL_DEPLOY_LOCK) first = restore.restore_after_failed_managed_cutover(store) replay = restore.restore_after_failed_managed_cutover(store) assert (controller / "projects.json").read_bytes() == original assert first == replay assert first["status"] == "restored" assert first["restoredProjectsSha256"] == hashlib.sha256(original).hexdigest() assert first["managedProjectsSha256"] != first["restoredProjectsSha256"] assert first["aliasesVerified"] is True assert "target_credential" not in json.dumps(first) assert b"R" * 64 not in json.dumps(first).encode("utf-8") def test_registry_restore_requires_durable_intent_before_accepting_old_digest( store_env, monkeypatch ): store_module, store, controller, state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) original = (controller / "projects.json").read_bytes() _registered_recovery(store, restore) projects_path = controller / "projects.json" projects_path.write_bytes(original) projects_path.chmod(0o600) with store.transaction(), pytest.raises( store_module.AdoptionStoreError, match="registry_restore_intent_missing" ): restore.restore_after_failed_managed_cutover(store) assert projects_path.read_bytes() == original assert not (state_root / "registry-restore-receipt.json").exists() def test_registry_restore_receipt_prevents_a_second_revert_after_re_registration( store_env, monkeypatch ): store_module, store, controller, _state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) _registered_recovery(store, restore) projects_path = controller / "projects.json" managed = projects_path.read_bytes() with store.transaction(): restore.restore_after_failed_managed_cutover(store) projects_path.write_bytes(managed) projects_path.chmod(0o600) with store.transaction(), pytest.raises( store_module.AdoptionStoreError, match="registry_restore_receipt_conflict" ): restore.restore_after_failed_managed_cutover(store) assert projects_path.read_bytes() == managed def test_registry_restore_compare_and_swap_drift_never_overwrites_new_registry( store_env, monkeypatch ): store_module, store, controller, _state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) _registered_recovery(store, restore) projects_path = controller / "projects.json" drift = b'{"jens/repopulse":{"changed":"after-preflight"}}\n' atomic_replace = store_module._atomic_replace_cas injected = False def change_before_cas(path, payload, **kwargs): nonlocal injected if path == projects_path and not injected: injected = True projects_path.write_bytes(drift) projects_path.chmod(0o600) return atomic_replace(path, payload, **kwargs) monkeypatch.setattr(store_module, "_atomic_replace_cas", change_before_cas) with store.transaction(), pytest.raises( store_module.AdoptionStoreError, match="projects_digest_conflict" ): restore.restore_after_failed_managed_cutover(store) assert injected assert projects_path.read_bytes() == drift def test_registry_restore_requires_a_durable_snapshot(store_env, monkeypatch): store_module, store, controller, _state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) _registered_recovery(store, restore, preserve=False) managed = (controller / "projects.json").read_bytes() with store.transaction(), pytest.raises( store_module.AdoptionStoreError, match="registry_snapshot_missing" ): restore.restore_after_failed_managed_cutover(store) assert (controller / "projects.json").read_bytes() == managed def test_registry_restore_rejects_recovery_reasons_before_cutover(store_env, monkeypatch): store_module, store, controller, _state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) _registered_recovery(store, restore, reason="database_backup_failed") managed = (controller / "projects.json").read_bytes() with store.transaction(), pytest.raises( store_module.AdoptionStoreError, match="registry_restore_not_eligible" ): restore.restore_after_failed_managed_cutover(store) assert (controller / "projects.json").read_bytes() == managed def test_registry_restore_rejects_symlinked_snapshot(store_env, monkeypatch): store_module, store, _controller, state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) _to_snapshot_verified(store) with store.transaction(): store._prepare_journal_root() snapshot_path = state_root / "projects-before.bin" outside = state_root.parent / "outside-snapshot" outside.write_bytes(b"do not read or overwrite") try: os.symlink(outside, snapshot_path) except (OSError, NotImplementedError): pytest.skip("symlink creation is unavailable on this test host") with pytest.raises( store_module.AdoptionStoreError, match="controller_path_unsafe" ): restore.preserve_before_registry_cas(store) assert outside.read_bytes() == b"do not read or overwrite" def test_registry_snapshot_rejects_group_or_world_accessible_snapshot( store_env, monkeypatch ): store_module, store, _controller, state_root = store_env restore = _load_registry_restore(store_module, monkeypatch) _to_snapshot_verified(store) snapshot_path = state_root / "projects-before.bin" with store.transaction(): store._prepare_journal_root() snapshot_path.write_bytes(b"{}\n") snapshot_path.chmod(0o600) original_mode_bits = store_module._mode_bits def unsafe_snapshot_mode(path, info): if path == snapshot_path: return 0o644 return original_mode_bits(path, info) monkeypatch.setattr(store_module, "_mode_bits", unsafe_snapshot_mode) with store.transaction(), pytest.raises( store_module.AdoptionStoreError, match="controller_file_mode_invalid" ): restore.preserve_before_registry_cas(store)