diff --git a/harness/tests/README.md b/harness/tests/README.md index 88a6a2c..57daa2e 100644 --- a/harness/tests/README.md +++ b/harness/tests/README.md @@ -21,6 +21,14 @@ trees under `/opt/data/gitea-agent/harness/tests/{common,spec,impl,review}`. This document is owned by the canonical harness tests and must be updated in place when their supported discovery contract or command paths change. +Broker fixture contract: in-process broker fixtures bind and connect through +`/proc/self/fd/{parent_fd}/{socket_name}`, matching the production broker. Long +UTF-8 roots and socket paths are supported; the fixture must not shorten them or +reject them merely because the conventional Unix-domain path encoding exceeds +107 bytes. Fixture setup failures clean up only resources they own, preserve +foreign/replaced paths, close every client and directory descriptor, and retain +unauthenticated-token rejection behavior. + It validates all `test*.py` files recursively under exactly these required roles: `common`, `spec`, `impl`, and `review`. `harness/tests`, every role, and every nested test package must have an `__init__.py`; every role and every diff --git a/harness/tests/common/test_managed_lock_broker.py b/harness/tests/common/test_managed_lock_broker.py index 23e599b..f4b1b54 100644 --- a/harness/tests/common/test_managed_lock_broker.py +++ b/harness/tests/common/test_managed_lock_broker.py @@ -70,14 +70,19 @@ class ManagedLockBrokerTests(unittest.TestCase): if token is not None: data["token"] = token import socket - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: - s.connect(data["socket"]) - payload = {"action": action, "token": data["token"] if token is None else token} - if ttl is not None: - payload["ttl_seconds"] = ttl - s.sendall((json.dumps(payload) + "\n").encode()) - with s.makefile("rb") as response_file: - return json.loads(response_file.readline()) + parent_fd, name = broker._open_parent(Path(data["socket"])) + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: + s.settimeout(3) + s.connect(f"/proc/self/fd/{parent_fd}/{name}") + payload = {"action": action, "token": data["token"] if token is None else token} + if ttl is not None: + payload["ttl_seconds"] = ttl + s.sendall((json.dumps(payload) + "\n").encode()) + with s.makefile("rb") as response_file: + return json.loads(response_file.readline()) + finally: + os.close(parent_fd) def test_constructor_rejects_lexical_root_and_handle_aliases(self): with tempfile.TemporaryDirectory() as d: @@ -255,7 +260,8 @@ class ManagedLockBrokerTests(unittest.TestCase): proc.terminate(); proc.wait(timeout=3) proc.stdout.close(); proc.stderr.close() - def make_bound_instance(self, root): + def make_bound_instance(self, root, *, fail_stage=None, captured=None, stages=None, failure=None): + """Build a real fixture, cleaning up and re-raising setup failures.""" import socket ensure_cache(root) locks = root / "repositories/alice/ledger/locks" @@ -265,16 +271,158 @@ class ManagedLockBrokerTests(unittest.TestCase): instance = broker.Broker(Namespace(root=str(root), owner="alice", repository="ledger", issue=7, role="gitea-impl-agent", run_id="cleanup-test", handle=str(runtime / "broker.handle"), socket=str(runtime / "broker.sock"), ttl_seconds=5.0, acquire_timeout=0.0)) - instance.set.issue.acquire() - broker._write_handle(instance.handle, instance.state, exclusive=True, - on_created=instance._record_handle_identity) - instance.handle_created = True; instance.started = True - instance.socket_parent_fd = broker._open_directory(runtime) - instance.server = socket.socket(socket.AF_UNIX) - instance.server.bind(str(instance.sock_path)) - st = instance.sock_path.lstat() - instance.socket_identity = (st.st_dev, st.st_ino) - return instance + if captured is not None: + captured.append(instance) + observed = stages if stages is not None else {} + try: + instance.set.issue.acquire() + observed["issue_acquired"] = instance.set.issue.held + if fail_stage == "after-issue": + raise RuntimeError("sentinel after issue acquire") + broker._write_handle(instance.handle, instance.state, exclusive=True, + on_created=instance._record_handle_identity) + instance.handle_created = True; instance.started = True + observed["handle_created"] = instance.handle_created and instance.handle_identity is not None + if fail_stage == "after-handle": + raise RuntimeError("sentinel after handle creation") + instance.socket_parent_fd = broker._open_directory(runtime) + observed["parent_fd_assigned"] = instance.socket_parent_fd is not None + if fail_stage == "after-parent-fd": + raise RuntimeError("sentinel after parent fd ownership") + instance.server = socket.socket(socket.AF_UNIX) + if fail_stage == "socket-constructor": + raise RuntimeError("sentinel socket constructor failure") + instance.server.bind(f"/proc/self/fd/{instance.socket_parent_fd}/{instance.sock_path.name}") + st = os.stat(instance.sock_path.name, dir_fd=instance.socket_parent_fd, follow_symlinks=False) + instance.socket_identity = (st.st_dev, st.st_ino) + observed["socket_bound"] = instance.socket_identity is not None + if fail_stage == "post-bind": + raise RuntimeError("sentinel after socket identity") + return instance + except BaseException as original: + if failure is not None: + failure.append(original) + try: + instance.cleanup("released") + except BaseException as cleanup_error: + original.add_note(f"fixture cleanup failed: {type(cleanup_error).__name__}") + raise + + def assert_failed_fixture_cleanup(self, instance, stages, failure, original): + self.assertIsNotNone(instance) + self.assertIs(failure[0], original) + self.assertTrue(stages["issue_acquired"]) + self.assertTrue(probe_free(instance.set.issue.path)) + self.assertTrue(probe_free(instance.set.repository.path)) + self.assertIsNone(instance.socket_parent_fd) + self.assertIsNone(instance.server) + self.assertTrue(instance.handle_created) + terminal = json.loads(instance.handle.read_text()) + self.assertNotIn("token", terminal) + self.assertIn(terminal["status"], ("released", "cleanup-incomplete")) + + def test_fixture_partial_handle_write_failure_releases_owned_lease(self): + captured = [] + writes = [] + original_write = broker._write_all + sentinel = OSError("sentinel partial fixture handle write") + + def partial_write(fd, data): + if not writes: + writes.append(os.write(fd, data[:1])) + raise sentinel + return original_write(fd, data) + + with tempfile.TemporaryDirectory() as d: + try: + with patch.object(broker, "_write_all", side_effect=partial_write): + with self.assertRaises(OSError) as raised: + self.make_bound_instance(Path(d), captured=captured) + self.assertIs(raised.exception, sentinel) + self.assertEqual(writes, [1], "partial write was not exercised") + instance = captured[0] + self.assertTrue(instance.handle_created) + self.assertTrue(probe_free(instance.set.issue.path), "fixture leaked its acquired lease") + self.assertIsNone(instance.socket_parent_fd) + self.assertIsNone(instance.server) + terminal = json.loads(instance.handle.read_text()) + self.assertNotIn("token", terminal) + self.assertEqual(terminal["status"], "released") + finally: + if captured: + captured[0].cleanup("released") + + def test_fixture_failure_after_handle_creation_preserves_original_and_cleans(self): + captured, stages, failure = [], {}, [] + instance = None + with tempfile.TemporaryDirectory() as d: + try: + with self.assertRaisesRegex(RuntimeError, "sentinel after handle creation") as raised: + self.make_bound_instance(Path(d), fail_stage="after-handle", + captured=captured, stages=stages, failure=failure) + instance = captured[0] + self.assert_failed_fixture_cleanup(instance, stages, failure, raised.exception) + self.assertTrue(stages["handle_created"]) + self.assertFalse(stages.get("parent_fd_assigned", False)) + finally: + if instance is not None: + instance.cleanup("released") + + def test_fixture_socket_constructor_failure_closes_owned_parent_fd(self): + captured, stages, failure = [], {}, [] + instance = None + with tempfile.TemporaryDirectory() as d: + try: + with patch("socket.socket", side_effect=RuntimeError("sentinel socket constructor failure")): + with self.assertRaisesRegex(RuntimeError, "sentinel socket constructor failure") as raised: + self.make_bound_instance(Path(d), captured=captured, stages=stages, failure=failure) + instance = captured[0] + self.assert_failed_fixture_cleanup(instance, stages, failure, raised.exception) + self.assertTrue(stages["parent_fd_assigned"]) + self.assertFalse(stages.get("socket_bound", False)) + finally: + if instance is not None: + instance.cleanup("released") + + def test_fixture_post_bind_failure_removes_only_captured_socket(self): + captured, stages, failure = [], {}, [] + instance = None + with tempfile.TemporaryDirectory() as d: + try: + with self.assertRaisesRegex(RuntimeError, "sentinel after socket identity") as raised: + self.make_bound_instance(Path(d), fail_stage="post-bind", + captured=captured, stages=stages, failure=failure) + instance = captured[0] + self.assert_failed_fixture_cleanup(instance, stages, failure, raised.exception) + self.assertTrue(stages["parent_fd_assigned"]) + self.assertTrue(stages["socket_bound"]) + self.assertFalse(instance.sock_path.exists()) + finally: + if instance is not None: + instance.cleanup("released") + + def test_fixture_cleanup_diagnostic_does_not_mask_setup_failure(self): + captured, stages, failure = [], {}, [] + instance = None + real_cleanup = broker.Broker.cleanup + + def cleanup_then_raise(status): + real_cleanup(captured[0], status) + raise OSError("sentinel cleanup diagnostic") + + with tempfile.TemporaryDirectory() as d: + try: + with patch.object(broker.Broker, "cleanup", side_effect=cleanup_then_raise): + with self.assertRaisesRegex(RuntimeError, "sentinel after handle creation") as raised: + self.make_bound_instance(Path(d), fail_stage="after-handle", + captured=captured, stages=stages, failure=failure) + instance = captured[0] + self.assertIs(failure[0], raised.exception) + self.assertIn("fixture cleanup failed: OSError", getattr(raised.exception, "__notes__", [])) + self.assertTrue(probe_free(instance.set.issue.path)) + finally: + if instance is not None: + real_cleanup(instance, "released") def test_release_reports_socket_failure_separately_and_stops(self): with tempfile.TemporaryDirectory() as d: @@ -326,12 +474,17 @@ class ManagedLockBrokerTests(unittest.TestCase): target = root / "replacement-target"; target.write_text("untouched") try: data = json.loads(handle.read_text()) - with socket.socket(socket.AF_UNIX) as client: - client.settimeout(3); client.connect(str(sock_path)) - sock_path.unlink(); sock_path.symlink_to(target) - client.sendall((json.dumps({"action": "release", "token": data["token"]}) + "\n").encode()) - with client.makefile("rb") as stream: - result = json.loads(stream.readline()) + parent_fd, socket_name = broker._open_parent(sock_path) + try: + with socket.socket(socket.AF_UNIX) as client: + client.settimeout(3) + client.connect(f"/proc/self/fd/{parent_fd}/{socket_name}") + sock_path.unlink(); sock_path.symlink_to(target) + client.sendall((json.dumps({"action": "release", "token": data["token"]}) + "\n").encode()) + with client.makefile("rb") as stream: + result = json.loads(stream.readline()) + finally: + os.close(parent_fd) self.assertEqual(result["error"], "cleanup-incomplete") self.assertTrue(result["cleanup"]["lease_released"]) self.assertEqual(proc.wait(timeout=3), 2) @@ -580,6 +733,105 @@ class ManagedLockBrokerTests(unittest.TestCase): self.assertNotEqual(p.returncode, 0) self.assertNotIn("token", p.stderr.lower()) + def test_fixture_binds_and_cleans_long_utf8_socket_path(self): + """The in-process fixture must exercise the broker's FD-relative path.""" + with tempfile.TemporaryDirectory() as d: + root = Path(d) / ("é" * 28) + root.mkdir() + instance = None + try: + instance = self.make_bound_instance(root) + self.assertGreater(len(os.fsencode(instance.sock_path)), 107) + self.assertTrue(instance.server) + self.assertIsNotNone(instance.socket_parent_fd) + finally: + if instance is not None: + instance.cleanup("released") + self.assertIsNone(instance.socket_parent_fd) + self.assertIsNone(instance.server) + self.assertTrue(probe_free(root / "repositories/alice/ledger/locks/issue-7.lock")) + self.assertFalse((runtime_dir(root, run_id="cleanup-test") / "broker.sock").exists()) + + def test_long_utf8_helper_request_preserves_unauthorized_semantics(self): + with tempfile.TemporaryDirectory() as d: + root = Path(d) / ("é" * 28) + root.mkdir() + proc, handle, _ = self.start(root) + try: + self.assertEqual(self.request(handle, "status")["status"], "held") + self.assertEqual( + self.request(handle, "status", token=secrets.token_hex(32))["error"], + "unauthorized", + ) + self.assertEqual(self.request(handle, "release")["status"], "released") + finally: + if proc.poll() is None: + proc.kill(); proc.wait(timeout=3) + proc.stdout.close(); proc.stderr.close() + + def test_startup_failures_release_owned_resources_without_masking_original(self): + """Each setup boundary must leave no owned FD, socket, lock, or token.""" + import contextlib + import io + + failures = ("metadata", "runtime-fd", "handle-write", "socket-bind") + for failure in failures: + with self.subTest(failure=failure), tempfile.TemporaryDirectory() as d: + root = Path(d) + ensure_cache(root) + locks = root / "repositories/alice/ledger/locks" + locks.mkdir(parents=True); locks.chmod(0o700) + (locks / "repository.lock").touch(mode=0o600) + (locks / "issue-7.lock").touch(mode=0o600) + runtime = runtime_dir(root, run_id=f"failure-{failure}") + args = Namespace(root=str(root), owner="alice", repository="ledger", issue=7, + role="gitea-impl-agent", run_id=f"failure-{failure}", + handle=str(runtime / "broker.handle"), + socket=str(runtime / "broker.sock"), ttl_seconds=5.0, + acquire_timeout=0.0) + instance = broker.Broker(args) + original_write_all = broker._write_all + original_chmod = broker.os.chmod + calls = {"write": 0} + + def partial_write(fd, data): + calls["write"] += 1 + if calls["write"] == 1: + os.write(fd, data[:1]) + raise OSError("injected handle write") + return original_write_all(fd, data) + + def late_chmod(path, mode, **kwargs): + original_chmod(path, mode, **kwargs) + if path == "broker.sock": + raise OSError("injected socket setup") + + with contextlib.ExitStack() as stack: + if failure == "metadata": + stack.enter_context(patch.object(instance.set.issue, "write_metadata", + side_effect=OSError("injected metadata"))) + elif failure == "runtime-fd": + stack.enter_context(patch.object(broker.os, "dup", + side_effect=OSError("injected dup"))) + elif failure == "handle-write": + stack.enter_context(patch.object(broker, "_write_all", + side_effect=partial_write)) + else: + stack.enter_context(patch.object(broker.os, "chmod", + side_effect=late_chmod)) + with contextlib.redirect_stdout(io.StringIO()): + with self.assertRaises(OSError): + instance.run() + self.assertTrue(probe_free(locks / "issue-7.lock")) + self.assertTrue(probe_free(locks / "repository.lock")) + self.assertIsNone(instance.socket_parent_fd) + self.assertIsNone(instance.server) + if failure in ("handle-write", "socket-bind"): + terminal = json.loads(instance.handle.read_text()) + self.assertNotIn("token", terminal) + self.assertIn(terminal["status"], ("released", "cleanup-incomplete")) + self.assertFalse((runtime / "broker.sock").exists()) + if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main()