"""One store composition has one lock identity (R4 wave 3, item A). Ancestor containment is not authority: a lock on a parent directory is an independent OS lock from the composition's own lock, so both can be held simultaneously. Guards require the held lock's frozen identity to equal the composition's canonical lock identity (:func:`canonical_lock_root`), and the acquired identity is frozen so a later root mutation cannot redirect a held handle. """ from __future__ import annotations import os import subprocess import sys from datetime import UTC, datetime from pathlib import Path import pytest from wf_artifacts.runs.store import FileRunStore from wf_scheduling import recovery as sched_recovery from wf_scheduling.ownership import ( SchedulerOwnership, SecondOwnerError, canonical_lock_root, canonical_store_path, ) from wf_scheduling.store import FileScheduleStore def _stores(root: Path) -> tuple[FileScheduleStore, FileRunStore]: return FileScheduleStore(root / "sched"), FileRunStore(root / "runs") def test_canonical_lock_identity_is_deterministic(tmp_path: Path) -> None: comp = tmp_path / "composition" sched_store, run_store = _stores(comp) assert canonical_lock_root( sched_store.root, run_store.root ) == canonical_store_path(comp) # Aliased spellings of the same composition share the identity. dotted = tmp_path / "sub" / ".." / "composition" assert canonical_lock_root( dotted / "sched", dotted / "runs" ) == canonical_store_path(comp) # A different composition has a different identity; missing roots have none. assert canonical_lock_root(tmp_path / "other") != canonical_store_path(comp) assert canonical_lock_root(None, run_store.root) is None # type: ignore[arg-type] assert canonical_lock_root() is None def test_parent_and_child_locks_cannot_both_authorize(tmp_path: Path) -> None: comp = tmp_path / "composition" sched_store, run_store = _stores(comp) parent_lock = SchedulerOwnership(tmp_path, owner="parent").acquire() try: # A different lock file: simultaneous hold proves independence. child_lock = SchedulerOwnership(comp, owner="child").acquire() try: assert parent_lock.held and child_lock.held assert child_lock.covers(sched_store.root, run_store.root) is True assert parent_lock.covers(sched_store.root, run_store.root) is False now = datetime.now(UTC) sched_recovery.recover( schedule_store=sched_store, run_store=run_store, now=now, ownership=child_lock, ) before_runs = run_store.list_runs() before_admissions = run_store.list_admissions() with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=FileScheduleStore(comp / "sched"), run_store=FileRunStore(comp / "runs"), now=now, ownership=parent_lock, ) # Wrong-root rejection happens before any write. assert FileRunStore(comp / "runs").list_runs() == before_runs assert FileRunStore(comp / "runs").list_admissions() == before_admissions finally: child_lock.release() finally: parent_lock.release() def test_strict_ancestor_lock_rejected_for_nested_composition(tmp_path: Path) -> None: comp = tmp_path / "composition" sched_store, run_store = _stores(comp) grandparent = SchedulerOwnership(tmp_path.parent, owner="grandparent").acquire() try: assert grandparent.covers(sched_store.root, run_store.root) is False finally: grandparent.release() def test_competing_processes_share_the_canonical_lock(tmp_path: Path) -> None: comp = tmp_path / "composition" comp.mkdir(parents=True) sched_store, run_store = _stores(comp) identity = Path(canonical_lock_root(sched_store.root, run_store.root) or "") holder = SchedulerOwnership(identity, owner="parent").acquire() try: script_path = tmp_path / "compete_canonical.py" script_path.write_text( "import sys\n" "sys.path.insert(0, '.')\n" "from pathlib import Path\n" "from wf_scheduling.ownership import canonical_lock_root, SchedulerOwnership, SecondOwnerError\n" f"sched_root = Path({str(sched_store.root)!r})\n" f"runs_root = Path({str(run_store.root)!r})\n" "identity = Path(canonical_lock_root(sched_root, runs_root))\n" "try:\n" " SchedulerOwnership(identity, owner='child').acquire()\n" "except SecondOwnerError:\n" " print('second-owner-rejected')\n" " raise SystemExit(0)\n" "print('unexpectedly-acquired')\n" "raise SystemExit(1)\n", encoding="utf-8", ) proc = subprocess.run( [sys.executable, str(script_path)], capture_output=True, text=True, timeout=60, ) assert proc.returncode == 0, proc.stderr assert "second-owner-rejected" in proc.stdout # A lock on the parent is an independent file: it acquires, but it # cannot authorize the composition while the canonical lock is held. outsider = SchedulerOwnership(tmp_path, owner="outsider").acquire() try: assert outsider.held assert outsider.covers(sched_store.root, run_store.root) is False with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=FileScheduleStore(comp / "sched"), run_store=FileRunStore(comp / "runs"), now=datetime.now(UTC), ownership=outsider, ) finally: outsider.release() finally: holder.release() def test_released_or_missing_ownership_rejected(tmp_path: Path) -> None: comp = tmp_path / "composition" sched_store, run_store = _stores(comp) never = SchedulerOwnership(comp, owner="never-acquired") assert never.covers(sched_store.root, run_store.root) is False with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=sched_store, run_store=run_store, now=datetime.now(UTC), ownership=never, ) held = SchedulerOwnership(comp, owner="test").acquire() assert held.covers(sched_store.root, run_store.root) is True held.release() assert held.covers(sched_store.root, run_store.root) is False with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=sched_store, run_store=run_store, now=datetime.now(UTC), ownership=held, ) def test_root_mutation_cannot_redirect_held_authority(tmp_path: Path) -> None: comp = tmp_path / "composition" sched_store, run_store = _stores(comp) other_sched, other_runs = _stores(tmp_path / "elsewhere") ownership = SchedulerOwnership(comp, owner="test").acquire() try: assert ownership.covers(sched_store.root, run_store.root) is True ownership.root = tmp_path / "elsewhere" # Frozen identity: still proves the locked composition, never the # mutated one. assert ownership.canonical_root() == canonical_store_path(comp) assert ownership.covers(sched_store.root, run_store.root) is True assert ownership.covers(other_sched.root, other_runs.root) is False with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=other_sched, run_store=other_runs, now=datetime.now(UTC), ownership=ownership, ) finally: ownership.release() def test_canonical_windows_path_handling_retained(tmp_path: Path) -> None: base = tmp_path / "sched" dotted = tmp_path / "sub" / ".." / "sched" assert canonical_store_path(dotted) == canonical_store_path(base) if os.name == "nt": assert canonical_store_path("C:\\Temp\\SCHED") == canonical_store_path( "c:/temp/sched" ) def test_non_sibling_store_pairs_have_no_lock_identity(tmp_path: Path) -> None: """Only distinct siblings under one composition root share a lock file. Cross pairs that reuse one protected store (shared schedule store or shared run store), nested pairs, and same-directory dual use would map to different lock files while covering the same store files, so they have no lock identity at all: guards must reject them outright. """ overlap = tmp_path / "overlap" sched_store, run_store = _stores(overlap) other_sched, other_runs = _stores(tmp_path / "other") # Shared schedule store, different run store. assert canonical_lock_root(sched_store.root, other_runs.root) is None # Shared run store, different schedule store. assert canonical_lock_root(other_sched.root, run_store.root) is None # Nested pairs: one store inside the other. assert canonical_lock_root(overlap, sched_store.root) is None assert canonical_lock_root(sched_store.root, overlap) is None # Same directory serving as both stores. assert canonical_lock_root(sched_store.root, sched_store.root) is None def test_shared_schedule_store_cannot_gain_independent_authority( tmp_path: Path, ) -> None: overlap = tmp_path / "overlap" sched_store, run_store = _stores(overlap) _, other_runs = _stores(tmp_path / "other") owner = SchedulerOwnership(overlap, owner="owner").acquire() intruder = SchedulerOwnership(tmp_path, owner="intruder").acquire() try: assert owner.covers(sched_store.root, run_store.root) is True # The cross pair reuses the protected schedule store but maps to # no lock identity, so no held lock can authorize it. assert intruder.covers(sched_store.root, other_runs.root) is False assert owner.covers(sched_store.root, other_runs.root) is False now = datetime.now(UTC) sched_recovery.recover( schedule_store=sched_store, run_store=run_store, now=now, ownership=owner, ) before_runs = FileRunStore(overlap / "runs").list_runs() with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=FileScheduleStore(overlap / "sched"), run_store=FileRunStore(tmp_path / "other" / "runs"), now=now, ownership=intruder, ) # Rejection happens before any write to either store. assert FileRunStore(overlap / "runs").list_runs() == before_runs assert FileRunStore(tmp_path / "other" / "runs").list_runs() == [] assert FileRunStore(tmp_path / "other" / "runs").list_admissions() == [] finally: owner.release() intruder.release() def test_shared_run_store_cannot_gain_independent_authority( tmp_path: Path, ) -> None: overlap = tmp_path / "overlap" sched_store, run_store = _stores(overlap) other_sched, _ = _stores(tmp_path / "other") owner = SchedulerOwnership(overlap, owner="owner").acquire() intruder = SchedulerOwnership(tmp_path, owner="intruder").acquire() try: assert owner.covers(sched_store.root, run_store.root) is True assert intruder.covers(other_sched.root, run_store.root) is False assert owner.covers(other_sched.root, run_store.root) is False now = datetime.now(UTC) with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=FileScheduleStore(tmp_path / "other" / "sched"), run_store=FileRunStore(overlap / "runs"), now=now, ownership=intruder, ) assert FileRunStore(overlap / "runs").list_runs() == [] assert FileRunStore(overlap / "runs").list_admissions() == [] finally: owner.release() intruder.release()