from __future__ import annotations import json import threading from pathlib import Path import pytest from checkin_cli import customer_admin as customer_admin_module from checkin_cli.customer_cleanup import ( CleanupBlockedError, CleanupIntegrityError, archive_customer_cleanup, post_cleanup_authority_inventory, resume_customer_cleanup, ) from checkin_cli.customer_cleanup_inventory import relevant_rows KEY = "client_001" FOREIGN_KEY = "client_002" def _private_write(path: Path, content: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) path.parent.chmod(0o700) path.write_bytes(content) path.chmod(0o600) def _profile(tmp_path: Path) -> Path: root = tmp_path / "profile" registry = root / "customers/registry.json" payload = { "version": 1, "owner": {"user_id": "1", "chat_id": "-100", "topic_id": "10"}, "customers": [{ "customer_key": KEY, "display_name": "Customer", "telegram": {"user_id": "2", "chat_id": "-100", "topic_id": "20"}, "enabled": False, "ai_processing_consent": { "granted": False, "recorded_on": "2026-08-18", "notice_version": "v1" }, }], } _private_write(registry, (json.dumps(payload) + "\n").encode()) _private_write(root / f"data/customers/{KEY}/onboarding/state.json", b'{"answer":1}\n') _private_write(root / f"data/customers/{KEY}/outbox/message.json", b'{"state":"cancelled"}\n') return root def test_cleanup_archives_verifies_prunes_and_reports_terminal(tmp_path: Path) -> None: root = _profile(tmp_path) receipt = archive_customer_cleanup(root, KEY) assert receipt.phase == "committed" assert receipt.archive_verified assert not (root / f"data/customers/{KEY}").exists() manifest = json.loads(receipt.manifest_path.read_text()) assert manifest["operation_id"] == receipt.operation_id assert manifest["customer_key"] == KEY assert [row["relative_path"] for row in manifest["inventory"]] == sorted( row["relative_path"] for row in manifest["inventory"] ) inventory = post_cleanup_authority_inventory(root, KEY) assert inventory.terminal assert inventory.archive_verified and inventory.journal_committed assert inventory.active_count == inventory.pending_count == inventory.unknown_count == 0 def test_duplicate_and_concurrent_callers_observe_one_operation(tmp_path: Path) -> None: root = _profile(tmp_path) barrier = threading.Barrier(2) results = [] def run() -> None: barrier.wait() results.append(archive_customer_cleanup(root, KEY)) workers = [threading.Thread(target=run) for _ in range(2)] for worker in workers: worker.start() for worker in workers: worker.join() assert len(results) == 2 assert len({item.operation_id for item in results}) == 1 assert len(list((root / "data/customer-cleanup/archives").iterdir())) == 1 @pytest.mark.parametrize("phase", ["prepared", "copied_verified", "source_pruned", "committed"]) def test_crash_after_each_phase_resumes_forward_only(tmp_path: Path, phase: str) -> None: root = _profile(tmp_path) def crash(current: str) -> None: if current == phase: raise RuntimeError("injected crash") with pytest.raises(RuntimeError, match="injected crash"): archive_customer_cleanup(root, KEY, fault_injector=crash) receipt = resume_customer_cleanup(root, KEY) assert receipt.phase == "committed" phases = [json.loads(line)["phase"] for line in receipt.journal_path.read_text().splitlines()] assert phases == ["prepared", "copied_verified", "source_pruned", "committed"] def test_archive_mismatch_is_rejected_on_resume(tmp_path: Path) -> None: root = _profile(tmp_path) def crash(phase: str) -> None: if phase == "copied_verified": raise RuntimeError("stop") with pytest.raises(RuntimeError): archive_customer_cleanup(root, KEY, fault_injector=crash) archived = next((root / "data/customer-cleanup/archives").glob("*/files/**/*state.json")) archived.chmod(0o600) archived.write_bytes(b"tampered") with pytest.raises(CleanupIntegrityError, match="archive verification"): resume_customer_cleanup(root, KEY) def test_symlink_and_world_readable_source_are_rejected(tmp_path: Path) -> None: root = _profile(tmp_path) target = root / "outside.json" _private_write(target, b"secret") link = root / f"data/customers/{KEY}/linked.json" link.symlink_to(target) with pytest.raises(CleanupIntegrityError, match="symlink"): archive_customer_cleanup(root, KEY) link.unlink() exposed = root / f"data/customers/{KEY}/exposed.json" exposed.write_text("secret") exposed.chmod(0o644) with pytest.raises(CleanupIntegrityError, match="owner-only"): archive_customer_cleanup(root, KEY) def _activation_journal(customer_key: str, state: str = "committed") -> dict[str, object]: row: dict[str, object] = { "version": 3, "state": state, "recovery_required": False, "transaction_id": "activation-transaction", "customer_id": customer_key, "registry_path": "/private/profile/customers/registry.json", "data_root": f"/private/profile/data/customers/{customer_key}", "checklist_evidence_path": "/private/profile/checklist.json", "audit_path": "/private/profile/data/customer-activation-audit.jsonl", "registry_sha256": "a" * 64, "previous_registry_sha256": "b" * 64, "audit_record_sha256": "c" * 64, "previous_audit_sha256": None, "previous_registry": {"version": 1, "customers": []}, "created_at": "2026-08-18T00:00:00+00:00", "prepared_at": "2026-08-18T00:00:00+00:00", "nutrition_activation_receipt": {"schema_version": "nutrition_activation_v2"}, "staff_membership_evidence_path": "/private/profile/staff-membership.json", "staff_membership_evidence_sha256": "d" * 64, "staff_chat_inventory_sha256": "e" * 64, "membership_subscription_epoch_id": "epoch-1", } if state == "committed": row["committed_at"] = "2026-08-18T00:01:00+00:00" return row def _write_activation_journal(root: Path, row: dict[str, object]) -> Path: path = root / "data/customer-activation-journal.json" customer_admin_module._write_activation_journal(path, row) return path def test_golden_path_multiline_committed_activation_journal_allows_cleanup( tmp_path: Path, ) -> None: root = _profile(tmp_path) journal = _activation_journal(KEY) _write_activation_journal(root, journal) _private_write( root / "data/customer-activation-audit.jsonl", (json.dumps({"customer_id": KEY, "transaction_id": journal["transaction_id"]}) + "\n").encode(), ) receipt = archive_customer_cleanup(root, KEY) assert receipt.phase == "committed" inventory = post_cleanup_authority_inventory(root, KEY) assert inventory.terminal assert inventory.categories["activation"] == 2 assert inventory.unknown_count == 0 def test_activation_journal_committed_prepared_recovery_and_invalid_states( tmp_path: Path, ) -> None: for name, state, recovery_required, expected_unknown in ( ("committed", "committed", False, 0), ("prepared", "prepared", False, 1), ("recovery", "prepared", True, 1), ("invalid", "future_state", False, 1), ): root = _profile(tmp_path / name) journal = _activation_journal(KEY, state) journal["recovery_required"] = recovery_required if state == "future_state": _private_write( root / "data/customer-activation-journal.json", (json.dumps(journal, sort_keys=True, indent=2) + "\n").encode(), ) else: _write_activation_journal(root, journal) inventory = post_cleanup_authority_inventory(root, KEY) assert inventory.categories["activation"] == 1 assert inventory.unknown_count == expected_unknown assert not inventory.terminal def test_foreign_activation_journal_is_valid_but_not_relevant(tmp_path: Path) -> None: root = _profile(tmp_path) _write_activation_journal(root, _activation_journal(FOREIGN_KEY)) inventory = post_cleanup_authority_inventory(root, KEY) assert inventory.categories["activation"] == 0 assert inventory.unknown_count == 0 @pytest.mark.parametrize( "payload", [ "[]\n", "42\n", '{"customer_id":"client_001"}\n', json.dumps(_activation_journal(KEY)) + "\n" + json.dumps(_activation_journal(KEY)), ], ids=["list", "scalar", "missing-contract", "multiple-objects"], ) def test_malformed_activation_journal_fails_closed(tmp_path: Path, payload: str) -> None: root = _profile(tmp_path) _private_write(root / "data/customer-activation-journal.json", payload.encode()) with pytest.raises(CleanupIntegrityError, match="authority file is invalid"): post_cleanup_authority_inventory(root, KEY) def test_activation_journal_ambiguous_customer_binding_fails_closed(tmp_path: Path) -> None: root = _profile(tmp_path) journal = _activation_journal(KEY) journal["customer_key"] = FOREIGN_KEY _write_activation_journal(root, journal) with pytest.raises(CleanupIntegrityError, match="authority file is invalid"): post_cleanup_authority_inventory(root, KEY) def test_unsupported_authority_filename_fails_closed(tmp_path: Path) -> None: path = tmp_path / "arbitrary.json" _private_write(path, json.dumps(_activation_journal(KEY)).encode()) with pytest.raises(CleanupIntegrityError, match="unsupported authority file"): relevant_rows(path, KEY) def test_activation_audit_jsonl_supports_customer_id_and_is_globally_strict( tmp_path: Path, ) -> None: root = _profile(tmp_path) audit = root / "data/customer-activation-audit.jsonl" _private_write( audit, (json.dumps({"customer_id": KEY, "transaction_id": "target"}) + "\n").encode(), ) assert post_cleanup_authority_inventory(root, KEY).categories["activation"] == 1 _private_write( audit, ( json.dumps({"customer_id": FOREIGN_KEY, "transaction_id": "foreign"}) + "\n{corrupt foreign row}\n" ).encode(), ) with pytest.raises(CleanupIntegrityError, match="authority file is invalid"): archive_customer_cleanup(root, KEY) def test_pending_and_unknown_delivery_authority_block_terminal_cleanup(tmp_path: Path) -> None: for state in ("prepared", "unknown"): root = _profile(tmp_path / state) _private_write( root / "data/scheduled-deliveries.jsonl", (json.dumps({"customer_key": KEY, "state": state}) + "\n").encode(), ) with pytest.raises(CleanupBlockedError): archive_customer_cleanup(root, KEY) inventory = post_cleanup_authority_inventory(root, KEY) assert not inventory.terminal assert (inventory.pending_count if state == "prepared" else inventory.unknown_count) == 1 def test_inventory_independently_reports_orphan_customer_authority(tmp_path: Path) -> None: root = _profile(tmp_path) receipt = archive_customer_cleanup(root, KEY) _private_write(root / f"data/customers/{KEY}/service-state/current.json", b"{}") inventory = post_cleanup_authority_inventory(root, KEY) assert not inventory.terminal assert inventory.orphan_count == 1 assert inventory.categories["service_state"] == 1 assert receipt.phase == "committed" def test_cross_customer_candidate_and_hardlink_are_rejected(tmp_path: Path) -> None: root = _profile(tmp_path) other = root / "data/customers/client_002/secret.json" _private_write(other, b"other") with pytest.raises(CleanupIntegrityError, match="cross-customer"): archive_customer_cleanup(root, KEY, candidate_paths=(other,)) source = root / f"data/customers/{KEY}/onboarding/state.json" hardlink = source.with_name("second.json") hardlink.hardlink_to(source) with pytest.raises(CleanupIntegrityError, match="hard-linked"): archive_customer_cleanup(root, KEY)