"""Profile-owned weekly-operations storage authority tests.""" from __future__ import annotations import os from concurrent.futures import ThreadPoolExecutor from datetime import date, datetime from pathlib import Path from threading import Barrier from unittest import TestCase from unittest.mock import patch from dataclasses import asdict, replace from checkin_cli.weekly_operations import AppendResult, CanonicalPin, CustomerKey, DayState, WeeklyOperationInput, WeeklyOperationsCorruption, WeeklyOperationsError, WeeklyOperationsInputError, canonical_weekly_row, customer_identity_digest, weekly_row_digest from checkin_cli.weekly_operations_authority import AuthorityId, WeeklyOperationsMigrationRequired, discover_legacy_sidecars, open_authority_root from checkin_cli.weekly_operations_canonical_registry_history import REGISTRY_NAME from checkin_cli.weekly_operations_layout import DATA_SUFFIX, MARKER_NAME, customer_data_name, customer_lock_name from checkin_cli.weekly_operations_parent import acquire_parent_authority, reacquire_parent_authority from checkin_cli.weekly_operations_store import WeeklyOperationsStore from tests._weekly_operations_support import initialize_at, initialize_for_parent CUSTOMER = CustomerKey("client_001") def _operation() -> WeeklyOperationInput: return WeeklyOperationInput.for_customer(CUSTOMER, date(2026, 8, 17), DayState.SUBMITTED, CanonicalPin(1, "a" * 64), datetime.fromisoformat("2026-08-17T20:00:00+09:00")) def _authority_snapshot(root: Path) -> tuple[tuple[str, int, int, bytes], ...]: return tuple(sorted((str(path.relative_to(root)), path.stat().st_mode, path.stat().st_nlink, path.read_bytes() if path.is_file() else b"") for path in root.rglob("*"))) def test_customer_rename_immediately_before_write_uses_profile_authority_only(tmp_path: Path) -> None: # Given: the r3 customer-tree store and an intended dedicated profile authority root. authority = tmp_path / "profile/weekly-operations" authority.mkdir(parents=True, mode=0o700) authority.parent.chmod(0o700) authority_root = initialize_at(authority, AuthorityId("8" * 64)) customer_parent = tmp_path / "customers" customer = customer_parent / "customer" customer_parent.mkdir(mode=0o700) customer.mkdir(mode=0o700) store = WeeklyOperationsStore.for_authority(authority_root, CUSTOMER) ready, renamed = Barrier(2), Barrier(2) first_write = iter((True,)) real_write = os.write def gated_write(descriptor: int, payload: bytes) -> int: if next(first_write, False): _ = ready.wait(timeout=5) _ = renamed.wait(timeout=5) return real_write(descriptor, payload) # When: customer authority-looking paths are detached immediately before write. with patch("os.write", side_effect=gated_write): with ThreadPoolExecutor(max_workers=1) as executor: result = executor.submit(store.append, _operation()) _ = ready.wait(timeout=5) detached = customer_parent.with_name("customers-detached") _ = customer_parent.replace(detached) customer_parent.mkdir(mode=0o700) customer.mkdir(mode=0o700) _ = renamed.wait(timeout=5) try: outcome = result.result(timeout=5) except WeeklyOperationsError as error: outcome = error # Then: authority commits once while both customer namespaces remain untouched. authority_data = authority / customer_data_name(customer_identity_digest(CUSTOMER)) detached_data = detached / "customer/nutrition-plans/weekly-operations.jsonl" replacement_data = customer / "nutrition-plans/weekly-operations.jsonl" authority_rows = authority_data.read_bytes().count(b"\n") if authority_data.exists() else 0 assert (type(outcome), authority_rows, detached_data.exists(), replacement_data.exists()) == (AppendResult, 1, False, False) def test_two_customers_are_isolated_under_opaque_digests(tmp_path: Path) -> None: authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("a" * 64)) other = CustomerKey("client_002") _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).append(_operation()) _ = WeeklyOperationsStore.for_authority(authority, other).append(replace(_operation(), customer_identity_digest=customer_identity_digest(other))) first = authority_path / customer_data_name(customer_identity_digest(CUSTOMER)) second = authority_path / customer_data_name(customer_identity_digest(other)) assert first != second and first.read_bytes().count(b"\n") == second.read_bytes().count(b"\n") == 1 def test_customer_filesystem_path_never_enters_authority_layout_or_bytes(tmp_path: Path) -> None: authority_path = tmp_path / "profile/authority" authority_path.parent.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("b" * 64)) customer_path = tmp_path / "customers/private-name" customer_path.mkdir(parents=True, mode=0o700) _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).append(_operation()) data = authority_path / customer_data_name(customer_identity_digest(CUSTOMER)) assert str(customer_path) not in str(data) and str(customer_path).encode() not in data.read_bytes() def test_malicious_customer_key_cannot_escape_digest_layout(tmp_path: Path) -> None: authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("c" * 64)) malicious = CustomerKey("../../outside") assert WeeklyOperationsStore.for_authority(authority, malicious).read() == () children = tuple(authority_path.iterdir()) assert {child.name for child in children} == {MARKER_NAME, REGISTRY_NAME, customer_lock_name(customer_identity_digest(malicious))} assert all(child.parent == authority_path for child in children) def test_pinned_parent_capability_survives_bootstrap_parent_rename(tmp_path: Path) -> None: bootstrap_parent = tmp_path / "profile" bootstrap_parent.mkdir(mode=0o700) child = bootstrap_parent / "authority" authority = initialize_at(child, AuthorityId("d" * 64)) binding = authority.binding detached_parent = tmp_path / "profile-detached" _ = bootstrap_parent.replace(detached_parent) bootstrap_parent.mkdir(mode=0o700) _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).append(_operation()) reopened = open_authority_root(authority.parent, binding) assert len(WeeklyOperationsStore.for_authority(reopened, CUSTOMER).read()) == 1 assert tuple(bootstrap_parent.iterdir()) == () def test_recreated_parent_path_cannot_reacquire_or_merge_history(tmp_path: Path) -> None: bootstrap_parent = tmp_path / "profile" bootstrap_parent.mkdir(mode=0o700) child = bootstrap_parent / "authority" authority = initialize_at(child, AuthorityId("e" * 64)) parent_binding = authority.parent.binding detached_parent = tmp_path / "profile-detached" _ = bootstrap_parent.replace(detached_parent) bootstrap_parent.mkdir(mode=0o700) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = reacquire_parent_authority(bootstrap_parent / "authority", parent_binding) assert tuple(bootstrap_parent.iterdir()) == () assert WeeklyOperationsStore.for_authority(authority, CUSTOMER).read() == () def test_legacy_discovery_fails_closed_without_moving_data(tmp_path: Path) -> None: clean, legacy = tmp_path / "clean", tmp_path / "legacy" clean.mkdir(mode=0o700) path = legacy / "nutrition-plans/weekly-operations.jsonl" path.parent.mkdir(parents=True, mode=0o700) _ = path.write_bytes(b"legacy") assert discover_legacy_sidecars((clean,)).discovered == () with TestCase().assertRaisesRegex(WeeklyOperationsMigrationRequired, "migration_required"): _ = discover_legacy_sidecars((clean, legacy)) assert path.read_bytes() == b"legacy" def test_authority_bootstrap_intermediate_symlink_is_rejected_without_mutation(tmp_path: Path) -> None: outside = tmp_path / "outside" outside.mkdir(mode=0o700) linked_parent = tmp_path / "profile" linked_parent.symlink_to(outside, target_is_directory=True) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = initialize_for_parent(acquire_parent_authority(linked_parent / "authority"), AuthorityId("f" * 64)) assert tuple(outside.iterdir()) == () def test_authority_marker_mode_and_link_identity_are_enforced(tmp_path: Path) -> None: mode_path, link_path = tmp_path / "mode", tmp_path / "link" mode_path.mkdir(mode=0o700) link_path.mkdir(mode=0o700) mode_authority = initialize_at(mode_path, AuthorityId("1" * 64)) link_authority = initialize_at(link_path, AuthorityId("2" * 64)) (mode_path / MARKER_NAME).chmod(0o644) outside_marker = tmp_path / "marker-link" os.link(link_path / MARKER_NAME, outside_marker) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = open_authority_root(mode_authority.parent, mode_authority.binding) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = open_authority_root(link_authority.parent, link_authority.binding) assert (mode_path / MARKER_NAME).stat().st_mode & 0o777 == 0o644 and outside_marker.stat().st_nlink == 2 def test_r5_red_sentinel_raw_customer_identity_is_absent_from_all_authority_surfaces(tmp_path: Path) -> None: sentinel = CustomerKey("telegram_998877665544_jane_doe") authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("3" * 64)) operation = replace(_operation(), customer_identity_digest=customer_identity_digest(sentinel)) store = WeeklyOperationsStore.for_authority(authority, sentinel) result = store.append(operation) try: _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).append(operation) except WeeklyOperationsInputError as error: error_surface = str(error).encode() else: raise AssertionError paths = tuple(authority_path.rglob("*")) persistent = b"".join(path.read_bytes() for path in paths if path.is_file()) public = repr((operation, result, store.__dict__, asdict(authority.binding))).encode() + result.row.model_dump_json().encode() + error_surface relative_paths = "\n".join(str(path.relative_to(authority_path)) for path in paths) assert sentinel.encode() not in persistent + public and sentinel not in relative_paths def test_r5_red_byte_identical_marker_inode_replacement_fails_fresh_open(tmp_path: Path) -> None: authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("4" * 64)) binding = authority.binding marker = authority_path / MARKER_NAME payload = marker.read_bytes() original = tmp_path / "marker-original" _ = marker.replace(original) _ = marker.write_bytes(payload) marker.chmod(0o600) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = open_authority_root(authority.parent, binding) assert marker.read_bytes() == payload and original.read_bytes() == payload def test_initializer_refuses_an_initialized_root_without_mutation(tmp_path: Path) -> None: authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("7" * 64)) before = _authority_snapshot(authority_path) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = initialize_at(authority_path, authority.binding.authority_id) assert _authority_snapshot(authority_path) == before def test_modified_caller_binding_fails_before_authority_mutation(tmp_path: Path) -> None: authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("5" * 64)) before = _authority_snapshot(authority_path) modified = replace(authority.binding, marker_inode=authority.binding.marker_inode + 1) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = open_authority_root(authority.parent, modified) after = _authority_snapshot(authority_path) assert after == before def test_digest_history_mismatch_fails_closed_without_append(tmp_path: Path) -> None: authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("6" * 64)) store = WeeklyOperationsStore.for_authority(authority, CUSTOMER) row = store.append(_operation()).row path = next((authority_path).glob(f"*{DATA_SUFFIX}")) injected = row.model_copy(update={"customer_identity_digest": "f" * 64, "row_digest": "0" * 64}) injected = injected.model_copy(update={"row_digest": weekly_row_digest(injected)}) payload = canonical_weekly_row(injected, include_digest=True) + b"\n" _ = path.write_bytes(payload) path.chmod(0o600) with TestCase().assertRaises(WeeklyOperationsCorruption): _ = store.append(_operation()) assert path.read_bytes() == payload def test_r6_red_extra_owner_only_root_directory_changes_nlink_and_fails_fresh_open(tmp_path: Path) -> None: authority_path = tmp_path / "authority" authority_path.mkdir(mode=0o700) authority = initialize_at(authority_path, AuthorityId("0" * 64)) binding = authority.binding before_links = authority_path.stat().st_nlink (authority_path / "injected-directory").mkdir(mode=0o700) assert authority_path.stat().st_nlink == before_links + 1 with TestCase().assertRaises(WeeklyOperationsCorruption): _ = open_authority_root(authority.parent, binding)