from harness.tests._support.paths import ROOT as SOURCE_ROOT import fcntl import os import stat import subprocess import sys import tempfile import time import unittest from pathlib import Path from unittest.mock import patch from harness.lib.managed_lock import ( LockOrderError, LockTimeoutError, LockValidationError, ManagedLock, ManagedLockSet, probe_free, validate_held_flock, ) class ManagedLockTests(unittest.TestCase): def test_lock_is_created_securely_and_held_for_context(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "repository.lock" with ManagedLock(path) as lock: self.assertIsNotNone(lock.fd) st = path.stat() self.assertEqual(stat.S_IMODE(st.st_mode), 0o600) self.assertEqual(st.st_nlink, 1) self.assertEqual(st.st_uid, os.getuid()) self.assertTrue(validate_held_flock(lock.fd, path)) self.assertTrue(probe_free(path)) def test_lock_timeout_is_bounded_and_does_not_steal_lock(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "issue.lock" with ManagedLock(path): start = time.monotonic() with self.assertRaises(LockTimeoutError): with ManagedLock(path, timeout=0.05, poll_interval=0.005): pass self.assertLess(time.monotonic() - start, 1) self.assertTrue(probe_free(path)) def test_existing_unsafe_lock_file_is_rejected_without_repair(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "bad.lock" path.write_text("x") os.chmod(path, 0o644) with self.assertRaises(LockValidationError): ManagedLock(path).acquire() self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o644) self.assertEqual(path.read_text(), "x") def test_literal_dotdot_component_is_rejected_before_normalization(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "a" / ".." / "b.lock" with self.assertRaises(LockValidationError): ManagedLock(path) def test_symlink_lock_is_rejected(self): with tempfile.TemporaryDirectory() as d: root = Path(d) target = root / "target" target.write_text("do not touch") path = root / "link.lock" path.symlink_to(target) with self.assertRaises(LockValidationError): ManagedLock(path).acquire() self.assertEqual(target.read_text(), "do not touch") def test_symlink_to_safe_lock_is_rejected(self): with tempfile.TemporaryDirectory() as d: root = Path(d) target = root / "safe-target.lock" target.touch(mode=0o600) path = root / "safe-link.lock" path.symlink_to(target) with self.assertRaises(LockValidationError): ManagedLock(path) def test_lock_path_through_symlinked_parent_is_rejected(self): with tempfile.TemporaryDirectory() as d: root = Path(d) target = root / "real" target.mkdir() parent = root / "linked" parent.symlink_to(target, target_is_directory=True) with self.assertRaises(LockValidationError): ManagedLock(parent / "issue.lock").acquire() self.assertFalse((target / "issue.lock").exists()) def test_ancestor_swap_after_parent_open_cannot_redirect_lock(self): """The final open stays relative to the already pinned parent inode.""" with tempfile.TemporaryDirectory() as d: root = Path(d) parent = root / "locks" parent.mkdir() diverted = root / "diverted" diverted.mkdir() module = __import__("harness.lib.managed_lock", fromlist=["_open_parent"]) original_open_parent = module._open_parent def swap_after_open(path, *, create_parent): fd, name = original_open_parent(path, create_parent=create_parent) parent.rename(root / "original-locks") parent.symlink_to(diverted, target_is_directory=True) return fd, name lock = ManagedLock(parent / "issue.lock") with patch("harness.lib.managed_lock._open_parent", side_effect=swap_after_open): with self.assertRaises(LockValidationError): lock.acquire() self.assertTrue((root / "original-locks" / "issue.lock").exists()) self.assertFalse((diverted / "issue.lock").exists()) self.assertFalse(lock.held) def test_hardlink_lock_is_rejected(self): with tempfile.TemporaryDirectory() as d: root = Path(d) path = root / "lock" alias = root / "alias" path.touch(mode=0o600) os.link(path, alias) with self.assertRaises(LockValidationError): ManagedLock(path).acquire() def test_for_issue_builds_canonical_paths(self): with tempfile.TemporaryDirectory() as d: root = Path(d).absolute() issue, repository = ManagedLockSet.paths_for_issue(root, "alice", "ledger", 7) self.assertEqual(issue, root / "repositories/alice/ledger/locks/issue-7.lock") self.assertEqual(repository, root / "repositories/alice/ledger/locks/repository.lock") locks = ManagedLockSet.for_issue(root, "alice", "ledger", 7) self.assertEqual(locks.issue.path, issue) self.assertEqual(locks.repository.path, repository) def test_for_issue_rejects_unsafe_identity_issue_and_root(self): with tempfile.TemporaryDirectory() as d: root = Path(d).absolute() for owner, repo in [("../alice", "ledger"), ("alice/x", "ledger"), ("alice", ".."), ("alice", "")]: with self.subTest(owner=owner, repo=repo), self.assertRaises(ValueError): ManagedLockSet.for_issue(root, owner, repo, 1) for issue in (0, -1, True, "1"): with self.subTest(issue=issue), self.assertRaises(ValueError): ManagedLockSet.for_issue(root, "alice", "ledger", issue) with self.assertRaises(ValueError): ManagedLockSet.for_issue(Path("relative"), "alice", "ledger", 1) with self.assertRaises(ValueError): ManagedLockSet.for_issue(root / ".." / root.name, "alice", "ledger", 1) def test_managed_lock_set_holds_issue_then_repository_scope_only(self): with tempfile.TemporaryDirectory() as d: root = Path(d) issue, repo = root / "issue", root / "repo" events = [] locks = ManagedLockSet(issue, repo, event_sink=events.append) with locks: self.assertTrue(validate_held_flock(locks.issue.fd, issue)) self.assertFalse(locks.repository.held) with locks.repository_scope(): self.assertTrue(validate_held_flock(locks.repository.fd, repo)) self.assertFalse(locks.repository.held) self.assertEqual(events, ["acquire:issue", "acquire:repository", "release:repository", "release:issue"]) def test_release_issue_rejects_while_repository_held(self): with tempfile.TemporaryDirectory() as d: locks = ManagedLockSet(Path(d) / "issue", Path(d) / "repo") with locks: with locks.repository_scope(): with self.assertRaises(LockOrderError): locks.release_issue() locks.release_issue() self.assertFalse(locks.issue.held) def test_context_exception_cleans_up_issue_and_repository(self): with tempfile.TemporaryDirectory() as d: locks = ManagedLockSet(Path(d) / "issue", Path(d) / "repo") with self.assertRaises(RuntimeError): with locks: with locks.repository_scope(): raise RuntimeError("boom") self.assertFalse(locks.issue.held) self.assertFalse(locks.repository.held) self.assertTrue(probe_free(locks.issue.path)) self.assertTrue(probe_free(locks.repository.path)) def test_repository_scope_requires_issue_and_cleans_up_acquire_exception(self): with tempfile.TemporaryDirectory() as d: locks = ManagedLockSet(Path(d) / "issue", Path(d) / "repo") with self.assertRaises(LockOrderError): with locks.repository_scope(): pass with ManagedLock(locks.repository.path): with locks: with self.assertRaises(LockTimeoutError): with locks.repository_scope(): pass self.assertTrue(locks.issue.held) locks.release_issue() def test_repository_scope_event_exception_releases_repository_and_issue_context(self): with tempfile.TemporaryDirectory() as d: def fail_on_repository(event): if event == "acquire:repository": raise RuntimeError("event failure") locks = ManagedLockSet(Path(d) / "issue", Path(d) / "repo", event_sink=fail_on_repository) locks.issue.acquire() with self.assertRaises(RuntimeError): with locks.repository_scope(): pass self.assertTrue(locks.issue.held) self.assertFalse(locks.repository.held) locks.release_issue() self.assertTrue(probe_free(locks.issue.path)) self.assertTrue(probe_free(locks.repository.path)) def test_write_metadata_replaces_bytes_exactly_on_repeated_writes(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "metadata.lock" with ManagedLock(path) as lock: lock.write_metadata({"long": "x" * 100}) lock.write_metadata({"x": 1}) self.assertEqual(path.read_bytes(), b'{"x":1}\n') def test_metadata_requires_held_lock(self): with tempfile.TemporaryDirectory() as d: lock = ManagedLock(Path(d) / "metadata.lock") with self.assertRaises(RuntimeError): lock.write_metadata({"state": "held"}) def test_probe_free_does_not_delete_or_modify_lock(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "probe.lock" with ManagedLock(path): pass before = path.stat() self.assertTrue(probe_free(path)) after = path.stat() self.assertEqual((before.st_ino, before.st_size, before.st_mode), (after.st_ino, after.st_size, after.st_mode)) def test_lock_contention_from_external_cwd_without_pythonpath(self): root = SOURCE_ROOT script = """ import sys import unittest sys.path.insert(0, sys.argv[1]) suite = unittest.defaultTestLoader.loadTestsFromName( 'harness.tests.common.locking.test_managed_lock.ManagedLockTests.' 'test_other_process_cannot_acquire_while_context_alive') result = unittest.TextTestRunner().run(suite) raise SystemExit(not result.wasSuccessful()) """ env = os.environ.copy() env.pop("PYTHONPATH", None) env["HARNESS_ROOT"] = str(root) with tempfile.TemporaryDirectory() as cwd: result = subprocess.run( [sys.executable, "-I", "-B", "-c", script, str(root)], cwd=cwd, env=env, text=True, capture_output=True, timeout=30) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) def test_invalid_harness_root_stops_before_lock_and_subprocess(self): with tempfile.TemporaryDirectory() as other: file_path = Path(other) / "not-a-directory" file_path.write_text("fixture") for value in (None, "", "relative", other, str(Path(other) / "missing"), str(file_path)): with self.subTest(value=value), patch.dict(os.environ), \ patch(__name__ + ".ManagedLock") as lock, \ patch(__name__ + ".subprocess.run") as run: if value is None: os.environ.pop("HARNESS_ROOT", None) else: os.environ["HARNESS_ROOT"] = value run.return_value.stdout = "blocked" error = None try: self.test_other_process_cannot_acquire_while_context_alive() except Exception as exc: error = exc self.assertIsInstance(error, RuntimeError) self.assertIn("HARNESS_ROOT", str(error)) lock.assert_not_called() run.assert_not_called() def test_other_process_cannot_acquire_while_context_alive(self): value = os.environ.get("HARNESS_ROOT") if not value: raise RuntimeError("HARNESS_ROOT must be configured") root = Path(value) if not root.is_absolute() or not root.is_dir(): raise RuntimeError("HARNESS_ROOT must be an existing absolute directory") root = root.resolve() if root != SOURCE_ROOT: raise RuntimeError("HARNESS_ROOT does not match this test installation") with tempfile.TemporaryDirectory() as d: path = Path(d) / "process.lock" script = """ import sys sys.path.insert(0, sys.argv[2]) from harness.lib.managed_lock import ManagedLock, LockTimeoutError try: with ManagedLock(sys.argv[1], timeout=0.03, poll_interval=0.005): print('acquired') except LockTimeoutError: print('blocked') """ with ManagedLock(path): result = subprocess.run([sys.executable, "-c", script, str(path), str(root)], text=True, capture_output=True, check=True, timeout=30) self.assertEqual(result.stdout.strip(), "blocked") if __name__ == "__main__": unittest.main()