"""Execute the exact Computer helper against disposable filesystem fixtures.""" import base64 import concurrent.futures import hashlib import json from pathlib import Path import subprocess import tempfile import unittest HELPER = Path(__file__).resolve().parents[1] / "crates/supervisor/src/computer_files.py" class ComputerFiles(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) / "home" self.root.mkdir() def call(self, action, name, content=b"", expected=None): return subprocess.run( ["python3", str(HELPER), action, str(self.root), name, json.dumps(expected)], input=base64.b64encode(content), capture_output=True, timeout=10, ) def list_page(self, path="", limit=200, cursor=None): return subprocess.run(["python3", str(HELPER), "list-page", str(self.root), path, "null"], input=json.dumps({"limit": limit, "cursor": cursor}).encode(), capture_output=True, timeout=10) def test_verified_read_returns_only_matching_binary_bytes(self): data = bytes(range(256)) * 4096 (self.root / 'artifact').write_bytes(data) expected = {'sha256': hashlib.sha256(data).hexdigest(), 'size': len(data)} def verified(request, path='artifact'): return subprocess.run(['python3', '-I', str(HELPER), 'read-verified', str(self.root), path], input=json.dumps(request).encode(), capture_output=True, timeout=5) result = verified(expected) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, data) for request in [{**expected, 'sha256': '0' * 64}, {**expected, 'size': len(data) - 1}, {**expected, 'size': True}, {**expected, 'sha256': 'bad'}]: result = verified(request) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, b'') (self.root / 'link').symlink_to('artifact') self.assertNotEqual(verified(expected, 'link').returncode, 0) (self.root / 'artifact').write_bytes(b'later edit') result = verified(expected) self.assertIn(b'CONFLICT:', result.stderr) self.assertEqual(result.stdout, b'') (self.root / 'empty').write_bytes(b'') self.assertEqual(verified({'sha256': hashlib.sha256(b'').hexdigest(), 'size': 0}, 'empty').returncode, 0) def test_paired_verification_is_readonly_and_locks_both_directories(self): (self.root / 'other').mkdir() (self.root / 'other/target').write_bytes(b'target\x00\xff') def verify(): return subprocess.run(['python3', str(HELPER), 'fingerprint-pair-verify', str(self.root), 'source'], input=json.dumps({'destination': 'other/target'}).encode(), capture_output=True, timeout=5) result = verify() self.assertEqual(result.returncode, 0, result.stderr) pair = json.loads(result.stdout) self.assertEqual(pair['source']['kind'], 'missing') self.assertEqual(pair['destination']['sha256'], hashlib.sha256(b'target\x00\xff').hexdigest()) self.assertTrue(pair['paired']) self.assertFalse((self.root / 'source').exists()) self.assertEqual((self.root / 'other/target').read_bytes(), b'target\x00\xff') # A cooperating writer owns the destination lock: even an absent source # must not let paired verification bypass that lock. import fcntl import os directory = os.open(self.root / 'other', os.O_RDONLY | os.O_DIRECTORY) try: fcntl.flock(directory, fcntl.LOCK_EX) with self.assertRaises(subprocess.TimeoutExpired): subprocess.run(['python3', str(HELPER), 'fingerprint-pair-verify', str(self.root), 'source'], input=b'{"destination":"other/target"}', capture_output=True, timeout=0.2) finally: os.close(directory) def test_fingerprint_is_bounded_binary_safe_and_never_follows_symlinks(self): name = "觀察\n'file" content = b"secret-fixture\x00\xff" (self.root / name).write_bytes(content) def observe(path): return subprocess.run(["python3", "-I", str(HELPER), "fingerprint", str(self.root), path], capture_output=True, timeout=5) result = observe(name) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout), {"path": name, "kind": "file", "size": len(content), "sha256": hashlib.sha256(content).hexdigest()}) self.assertNotIn(b"secret-fixture", result.stdout) self.assertEqual((self.root / name).read_bytes(), content) self.assertEqual(json.loads(observe("missing").stdout)["kind"], "missing") (self.root / "link").symlink_to(self.root / name) self.assertNotEqual(observe("link").returncode, 0) with (self.root / "large").open("wb") as stream: stream.truncate(1024 * 1024 * 1024) result = observe("large") self.assertNotEqual(result.returncode, 0) self.assertTrue(result.stderr.startswith(b"SIZE_LIMIT:")) def test_directory_pages_cover_large_directory_without_duplicates(self): names = {str(number) for number in range(4300)} | {"中文\n'檔案"} for name in names: (self.root / name).touch() seen, cursor, version = [], None, None while True: response = self.list_page(cursor=cursor) self.assertEqual(response.returncode, 0, response.stderr) page = json.loads(response.stdout) self.assertLessEqual(len(page["entries"]), 200) seen.extend(entry["path"] for entry in page["entries"]) if version is not None: self.assertEqual(page["version"], version) version = page["version"] cursor = page["nextCursor"] if page["eof"]: self.assertIsNone(cursor) break self.assertTrue(cursor) self.assertEqual(len(seen), len(names)) self.assertEqual(set(seen), names) def test_directory_cursor_replay_change_and_cross_path_guards(self): (self.root / "other").mkdir() for name in ["a", "b", "c"]: (self.root / name).touch() first = json.loads(self.list_page(limit=1).stdout) cursor = first["nextCursor"] second = self.list_page(limit=1, cursor=cursor) self.assertEqual(second.returncode, 0, second.stderr) self.assertEqual(json.loads(second.stdout), json.loads(self.list_page(limit=1, cursor=cursor).stdout)) self.assertIn(b"CONFLICT:", self.list_page("other", cursor=cursor).stderr) (self.root / "new").touch() stale = self.list_page(cursor=cursor) self.assertNotEqual(stale.returncode, 0) self.assertIn(b"CONFLICT:", stale.stderr) self.assertEqual(stale.stdout, b"") for invalid in ["bad-base64", base64.urlsafe_b64encode(b'{"version":"x","cookie":-1}').decode()]: self.assertNotEqual(self.list_page(cursor=invalid).returncode, 0) self.assertNotEqual(self.list_page(limit=True).returncode, 0) self.assertNotEqual(self.list_page("../escape").returncode, 0) empty = json.loads(self.list_page("other").stdout) self.assertEqual(empty["entries"], []) self.assertTrue(empty["eof"]) def test_directory_pages_bound_long_names_and_continue_at_output_limit(self): directory = self.root for number in range(4): directory = directory / (str(number) + "\x01" * 200) directory.mkdir() names = {str(number) + "\x01" * 200 for number in range(220)} for name in names: (directory / name).touch() path = str(directory.relative_to(self.root)) cursor, seen = None, [] while True: response = self.list_page(path, cursor=cursor) self.assertEqual(response.returncode, 0, response.stderr) self.assertLess(len(response.stdout), 600 * 1024) page = json.loads(response.stdout) self.assertLess(len(page["entries"]), 200) seen.extend(Path(entry["path"]).name for entry in page["entries"]) cursor = page["nextCursor"] if page["eof"]: break self.assertEqual(len(seen), len(names)) self.assertEqual(set(seen), names) def range_read(self, path, offset=0, length=65536, version=None): request = {"offset": offset, "length": length, "expectedVersion": version} return subprocess.run(["python3", str(HELPER), "read-range", str(self.root), path, "null"], input=json.dumps(request).encode(), capture_output=True, timeout=10) def test_byte_pages_round_trip_and_refuse_mixed_versions(self): data = bytes(range(256)) * 300 self.call("write", "binary", data) first = json.loads(self.range_read("binary").stdout) self.assertFalse(first["eof"]) self.assertEqual(first["bytesRead"], 65536) self.assertEqual(first["nextOffset"], 65536) second = json.loads(self.range_read("binary", first["nextOffset"], version=first["version"]).stdout) self.assertTrue(second["eof"]) self.assertEqual(base64.b64decode(first["contentBase64"]) + base64.b64decode(second["contentBase64"]), data) self.assertEqual(first["chunkSha256"], hashlib.sha256(data[:65536]).hexdigest()) self.assertEqual(first["version"], second["version"]) final = json.loads(self.range_read("binary", len(data), version=first["version"]).stdout) self.assertEqual(final["bytesRead"], 0) self.assertTrue(final["eof"]) self.call("write", "binary", data) stale = self.range_read("binary", 65536, version=first["version"]) self.assertNotEqual(stale.returncode, 0) self.assertIn(b"CONFLICT:", stale.stderr) self.assertEqual(stale.stdout, b"") for offset, length, version in [(1, 4, None), (-1, 4, None), (True, 4, None), (0, 0, None), (0, 65537, None), (0, 4, "wrong")]: invalid = self.range_read("binary", offset, length, version) self.assertNotEqual(invalid.returncode, 0) self.assertIn(b"INVALID_ARGUMENT:", invalid.stderr) self.assertIn(b"FILE_NOT_FOUND:", self.range_read("missing").stderr) def test_byte_pages_seek_large_sparse_files_without_loading_them(self): size = 1024 ** 4 with (self.root / "large").open("wb") as stream: stream.truncate(size) stream.seek(size - 4) stream.write(b"tail") first = json.loads(self.range_read("large", length=16).stdout) self.assertEqual(first["size"], size) self.assertEqual(base64.b64decode(first["contentBase64"]), b"\x00" * 16) tail = json.loads(self.range_read("large", size - 4, version=first["version"]).stdout) self.assertEqual(base64.b64decode(tail["contentBase64"]), b"tail") self.assertTrue(tail["eof"]) self.assertEqual(tail["version"], first["version"]) self.assertIn(b"INVALID_ARGUMENT:", self.range_read("large", size + 1, version=first["version"]).stderr) (self.root / "link").symlink_to("/etc/passwd") self.assertNotEqual(self.range_read("link").returncode, 0) def search(self, path, query, maximum=20): return subprocess.run(["python3", str(HELPER), "search", str(self.root), path, "null"], input=json.dumps({"query": query, "maxResults": maximum}).encode(), capture_output=True, timeout=10) def test_search_unicode_literal_offsets_and_visible_omissions(self): import os content = "前綴 literal[.*] 中文 end 中文".encode() self.call("write", "nested/text", content) result = json.loads(self.search("nested", "中文").stdout) self.assertTrue(result["complete"]) self.assertEqual([item["byteOffset"] for item in result["matches"]], [content.index("中文".encode()), content.rindex("中文".encode())]) self.assertIn("中文".encode(), base64.b64decode(result["matches"][0]["previewBase64"])) self.assertEqual(len(json.loads(self.search("nested/text", "[.*]").stdout)["matches"]), 1) self.call("write", "binary", b"\x00match") (self.root / "outside").symlink_to("/etc") os.mkfifo(self.root / "fifo") with (self.root / "large").open("wb") as stream: stream.truncate(1024 * 1024 + 1) result = json.loads(self.search("", "absent").stdout) self.assertFalse(result["complete"]) self.assertEqual(result["matches"], []) self.assertEqual(result["skipped"], {"binary": 1, "symlink": 1, "special": 1, "oversize": 1}) limited = json.loads(self.search("nested", "中文", 1).stdout) self.assertTrue(limited["truncated"]) self.assertEqual(limited["limitReason"], "results") self.assertEqual(len(limited["matches"]), 1) self.assertNotEqual(self.search("../escape", "x").returncode, 0) self.assertNotEqual(self.search("nested", "").returncode, 0) self.assertNotEqual(self.search("nested", "x", True).returncode, 0) def test_search_serialized_output_is_bounded(self): directory = self.root for number in range(4): directory = directory / (str(number) + "\x01" * 200) directory.mkdir() (directory / ("file" + "\x01" * 200)).write_text("x" * 101) response = self.search("", "x", 100) self.assertEqual(response.returncode, 0, response.stderr) result = json.loads(response.stdout) self.assertEqual(result["limitReason"], "outputBytes") self.assertLessEqual(result["matchBytes"], 512 * 1024) self.assertLess(len(response.stdout), 513 * 1024) self.assertFalse(result["complete"]) def test_search_entry_and_byte_limits_are_explicit(self): directory = self.root / "many" directory.mkdir() for number in range(4100): (directory / str(number)).touch() result = json.loads(self.search("many", "absent").stdout) self.assertEqual(result["limitReason"], "entries") self.assertEqual(result["entriesScanned"], 4096) self.assertFalse(result["complete"]) directory = self.root / "bytes" directory.mkdir() for number in range(17): (directory / str(number)).write_bytes(b"a" * (1024 * 1024)) result = json.loads(self.search("bytes", "absent").stdout) self.assertEqual(result["limitReason"], "bytes") self.assertEqual(result["bytesRead"], 16 * 1024 * 1024) self.assertFalse(result["complete"]) def move(self, source, destination, original, expected_target=""): request = {"destination": destination, "expectedHash": hashlib.sha256(original).hexdigest(), "expectedDestinationHash": expected_target} return subprocess.run(["python3", str(HELPER), "move", str(self.root), source, "null"], input=json.dumps(request).encode(), capture_output=True, timeout=10) def test_move_preconditions_preserve_files_and_metadata(self): data = b"\x00\xfforiginal" self.call("write", "a/source", data) (self.root / "b").mkdir() (self.root / "a/source").chmod(0o640) result = self.move("a/source", "b/target", data) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((self.root / "a/source").exists()) self.assertEqual((self.root / "b/target").read_bytes(), data) self.assertEqual((self.root / "b/target").stat().st_mode & 0o777, 0o640) self.assertTrue(json.loads(result.stdout)["verified"]) self.call("write", "a/source", b"new") for digest in ["", hashlib.sha256(b"wrong").hexdigest()]: result = self.move("a/source", "b/target", b"new", digest) self.assertNotEqual(result.returncode, 0) self.assertIn(b"CONFLICT", result.stderr) self.assertEqual((self.root / "a/source").read_bytes(), b"new") self.assertEqual((self.root / "b/target").read_bytes(), data) result = self.move("a/source", "b/target", b"new", hashlib.sha256(data).hexdigest()) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.root / "b/target").read_bytes(), b"new") self.assertNotEqual(self.move("b/target", "b/target", b"new").returncode, 0) (self.root / "escape").symlink_to("/tmp") self.assertNotEqual(self.move("b/target", "escape/should-not-exist", b"new").returncode, 0) (self.root / "link").symlink_to("b/target") self.assertNotEqual(self.move("link", "other", b"new").returncode, 0) self.assertNotEqual(self.move("b/target", "../escape", b"new").returncode, 0) def test_moves_lock_directories_in_stable_order(self): self.call("write", "a/source", b"a") self.call("write", "b/source", b"b") with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: first = pool.submit(self.move, "a/source", "b/target", b"a") second = pool.submit(self.move, "b/source", "a/target", b"b") for result in [first.result(), second.result()]: self.assertEqual(result.returncode, 0, result.stderr) self.call("write", "a/one", b"one") self.call("write", "b/two", b"two") with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: first = pool.submit(self.move, "a/one", "winner", b"one") second = pool.submit(self.move, "b/two", "winner", b"two") results = [first.result(), second.result()] self.assertEqual(sum(result.returncode == 0 for result in results), 1) self.assertEqual(sum(b"CONFLICT" in result.stderr for result in results), 1) def patch(self, name, original, offset, delete, replacement, **extra): request = {"offset": offset, "deleteBytes": delete, "contentBase64": base64.b64encode(replacement).decode(), "expectedHash": hashlib.sha256(original).hexdigest(), **extra} return subprocess.run(["python3", str(HELPER), "patch", str(self.root), name, "null"], input=json.dumps(request).encode(), capture_output=True, timeout=10) def test_journal_handoff_rejects_missing_or_truncated_payload_before_directories(self): payload = base64.b64encode(b"original") argv = ["python3", str(HELPER), "write", str(self.root), "nested/file", "null", "journal-v1", str(len(payload)), hashlib.sha256(payload).hexdigest()] for body in [b"", payload, b"LAZYBOY_FILE_COMMIT_V1\n" + payload[:4], b"LAZYBOY_FILE_COMMIT_V1\n" + payload + b"extra"]: result = subprocess.run(argv, input=body, capture_output=True, timeout=5) self.assertNotEqual(result.returncode, 0) self.assertFalse((self.root / "nested").exists()) result = subprocess.run(argv, input=b"LAZYBOY_FILE_COMMIT_V1\n" + payload, capture_output=True, timeout=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.root / "nested/file").read_bytes(), b"original") def test_patch_plan_is_readonly_and_mutation_rechecks_base(self): original = b"before\x00\xffafter" self.call("write", "file", original) request = {"offset": 6, "deleteBytes": 2, "contentBase64": base64.b64encode(b"new").decode(), "expectedHash": hashlib.sha256(original).hexdigest()} plan = subprocess.run(["python3", str(HELPER), "patch-plan", str(self.root), "file", "null"], input=json.dumps(request).encode(), capture_output=True, timeout=10) self.assertEqual(plan.returncode, 0, plan.stderr) desired = b"beforenewafter" result = json.loads(plan.stdout) self.assertTrue(result["planned"]) self.assertEqual(result["sha256"], hashlib.sha256(desired).hexdigest()) self.assertEqual(result["size"], len(desired)) self.assertEqual((self.root / "file").read_bytes(), original) self.assertFalse(list(self.root.glob(".lazyboy-write-*"))) (self.root / "file").write_bytes(b"human edit") self.assertEqual(self.patch("file", original, 6, 2, b"new").returncode, 73) self.assertEqual((self.root / "file").read_bytes(), b"human edit") def test_patch_binary_cas_range_and_atomic_rejection(self): original = b"before\x00\xffafter" self.call("write", "file", original) (self.root / "file").chmod(0o640) result = self.patch("file", original, 6, 2, b"\xfe\x00\x01") self.assertEqual(result.returncode, 0, result.stderr) updated = b"before\xfe\x00\x01after" self.assertEqual((self.root / "file").read_bytes(), updated) self.assertEqual(json.loads(result.stdout)["sha256"], hashlib.sha256(updated).hexdigest()) self.assertTrue(json.loads(result.stdout)["verified"]) self.assertEqual((self.root / "file").stat().st_mode & 0o777, 0o640) self.assertEqual(self.patch("file", original, 0, 1, b"x").returncode, 73) for offset, delete in [(100, 0), (0, 100), (-1, 0), (True, 0)]: self.assertNotEqual(self.patch("file", updated, offset, delete, b"x").returncode, 0) self.assertEqual((self.root / "file").read_bytes(), updated) self.assertNotEqual(self.patch("missing", b"", 0, 0, b"x").returncode, 0) (self.root / "link").symlink_to("file") self.assertNotEqual(self.patch("link", updated, 0, 0, b"x").returncode, 0) self.assertNotEqual(self.patch("file", updated, 0, 0, b"x", extra=True).returncode, 0) self.assertFalse(list(self.root.glob(".lazyboy-write-*"))) def test_concurrent_patches_have_one_winner(self): original = b"original" self.call("write", "file", original) with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda i: self.patch("file", original, 0, 1, str(i).encode()), range(8))) self.assertEqual(sum(r.returncode == 0 for r in results), 1) self.assertEqual(sum(r.returncode == 73 for r in results), 7) def test_stat_is_bounded_and_does_not_follow_links(self): import os large = self.root / "large" with large.open("wb") as stream: stream.truncate(1024 ** 4) large.chmod(0o640) metadata = json.loads(self.call("stat", "large").stdout) self.assertEqual(metadata["size"], 1024 ** 4) self.assertEqual(metadata["kind"], "file") self.assertEqual(metadata["mode"], 0o640) self.assertEqual(metadata["modifiedNs"], str(large.stat().st_mtime_ns)) (self.root / "link").symlink_to("/etc/passwd") self.assertEqual(json.loads(self.call("stat", "link").stdout)["kind"], "symlink") (self.root / "outside").symlink_to("/etc") self.assertNotEqual(self.call("stat", "outside/passwd").returncode, 0) self.assertNotEqual(self.call("stat", "../escape").returncode, 0) self.assertNotEqual(self.call("stat", "missing").returncode, 0) os.mkfifo(self.root / "pipe") self.assertEqual(json.loads(self.call("stat", "pipe").stdout)["kind"], "special") self.assertEqual(json.loads(self.call("stat", "").stdout)["kind"], "dir") def test_binary_and_unusual_names_round_trip(self): data = bytes(range(256)) * 4096 for name in ["shared/中文\n檔案", "$(touch INJECTED)", "`touch INJECTED`", "quote'\"\\name"]: result = self.call("write", name, data) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.call("read", name).stdout, data) self.assertFalse((self.root / "INJECTED").exists()) self.assertEqual(len(json.loads(self.call("list", "").stdout)), 4) def test_concurrent_compare_and_swap_has_one_winner(self): self.assertEqual(self.call("write", "shared/result", b"before").returncode, 0) digest = hashlib.sha256(b"before").hexdigest() with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda i: self.call("write", "shared/result", str(i).encode(), digest), range(8))) self.assertEqual(sum(r.returncode == 0 for r in results), 1) self.assertEqual(sum(r.returncode == 73 for r in results), 7) self.assertEqual(self.call("write", "shared/result", b"overwrite", "").returncode, 73) self.assertEqual(self.call("write", "new", b"create", "").returncode, 0) self.assertFalse(list(self.root.rglob(".lazyboy-write-*"))) def test_symlinks_and_special_files_cannot_escape_or_block(self): outside = Path(self.temp.name) / "sentinel" outside.write_bytes(b"unchanged") (self.root / "link").symlink_to(outside) (self.root / "escape").symlink_to(outside.parent, target_is_directory=True) for name in ["link", "escape/sentinel", "../sentinel", str(outside)]: for action in ["read", "write"]: self.assertNotEqual(self.call(action, name, b"bad").returncode, 0) self.assertEqual(outside.read_bytes(), b"unchanged") import os os.mkfifo(self.root / "fifo") self.assertNotEqual(self.call("read", "fifo").returncode, 0) self.assertNotEqual(self.call("write", "fifo", b"bad").returncode, 0) def test_listing_refuses_oversize_without_partial_success(self): for number in range(4097): (self.root / str(number)).touch() result = self.call("list", "") self.assertNotEqual(result.returncode, 0) self.assertIn(b"DIRECTORY_LIMIT:", result.stderr) self.assertEqual(result.stdout, b"") def test_listing_bounds_serialized_names_not_only_entry_count(self): for number in range(600): (self.root / (str(number) + "\x01" * 200)).touch() result = self.call("list", "") self.assertNotEqual(result.returncode, 0) self.assertIn(b"DIRECTORY_LIMIT:", result.stderr) self.assertEqual(result.stdout, b"") def test_listing_distinguishes_links_and_special_files(self): import os (self.root / "file").write_text("data") (self.root / "link").symlink_to("/etc/passwd") os.mkfifo(self.root / "fifo") entries = json.loads(self.call("list", "").stdout) self.assertEqual({entry["path"]: entry["kind"] for entry in entries}, {"file": "file", "link": "symlink", "fifo": "special"}) def test_read_only_errors_preserve_categories(self): for action in ["read", "stat", "list"]: result = self.call(action, "missing") self.assertNotEqual(result.returncode, 0) self.assertIn(b"FILE_NOT_FOUND:", result.stderr) self.assertEqual(result.stdout, b"") with (self.root / "large").open("wb") as stream: stream.truncate(8 * 1024 * 1024 + 1) self.assertIn(b"SIZE_LIMIT:", self.call("read", "large").stderr) self.assertIn(b"INVALID_ARGUMENT:", self.call("read", "../escape").stderr) def test_read_permission_denial_is_typed(self): import os Path(self.temp.name).chmod(0o755) self.root.chmod(0o755) private = self.root / "private" private.write_bytes(b"must-not-return") private.chmod(0) result = subprocess.run(["python3", "-I", "-c", HELPER.read_text(), "read", str(self.root), "private", "null"], capture_output=True, timeout=10, preexec_fn=(lambda: os.setuid(65534)) if os.geteuid() == 0 else None) self.assertNotEqual(result.returncode, 0) self.assertIn(b"PERMISSION_DENIED:", result.stderr) self.assertEqual(result.stdout, b"") def test_write_sync_failure_before_and_after_replace(self): # Execute the exact helper while injecting ENOSPC at each fsync boundary. import errno for failure in (1, 2): self.call("write", "file", b"original") bootstrap = """ import errno, os, sys source = sys.stdin.read() original_sync = os.fsync failure_point = int(sys.argv[1]) calls = 0 def injected_sync(fd): global calls calls += 1 if calls == failure_point: raise OSError(errno.ENOSPC, 'fixture disk full') return original_sync(fd) os.fsync = injected_sync sys.argv = ['helper', 'write', sys.argv[2], 'file', 'null'] import io sys.stdin = io.TextIOWrapper(io.BytesIO(b'bmV3')) exec(compile(source, 'computer_files.py', 'exec'), {'__name__': '__main__'}) """ result = subprocess.run(["python3", "-I", "-c", bootstrap, str(failure), str(self.root)], input=HELPER.read_bytes(), capture_output=True, timeout=10) self.assertNotEqual(result.returncode, 0) self.assertIn(b"fixture disk full", result.stderr) self.assertEqual((self.root / "file").read_bytes(), b"original" if failure == 1 else b"new") self.assertFalse(list(self.root.glob(".lazyboy-write-*"))) def test_postcondition_verification_refuses_failed_durability(self): (self.root / 'file').write_bytes(b'unchanged') bootstrap = """ import errno, os, sys source = sys.stdin.read() def fail_sync(fd): raise OSError(errno.ENOSPC, 'verification sync failed') os.fsync = fail_sync sys.argv = ['helper', 'fingerprint-verify', sys.argv[1], 'file'] exec(compile(source, 'computer_files.py', 'exec'), {'__name__': '__main__'}) """ result = subprocess.run(['python3', '-I', '-c', bootstrap, str(self.root)], input=HELPER.read_bytes(), capture_output=True, timeout=5) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, b'') self.assertIn(b'verification sync failed', result.stderr) self.assertEqual((self.root / 'file').read_bytes(), b'unchanged') def test_missing_list_and_oversize_are_errors(self): self.assertNotEqual(self.call("list", "absent").returncode, 0) self.assertNotEqual(self.call("write", "large", b"x" * (8 * 1024 * 1024 + 1)).returncode, 0) self.assertFalse((self.root / "large").exists()) def test_atomic_replace_preserves_existing_executable_permissions(self): target = self.root / "script" target.write_bytes(b"before") target.chmod(0o755) self.assertEqual(self.call("write", "script", b"after").returncode, 0) self.assertEqual(target.stat().st_mode & 0o777, 0o755) if __name__ == "__main__": unittest.main()