Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242#!/usr/bin/env python3"""Offline scheduling, selection and state isolation regressions."""from concurrent.futures import ThreadPoolExecutorimport jsonimport osfrom pathlib import Pathimport socketimport subprocessimport sysimport tempfileimport timeimport unittestfrom unittest.mock import patch
import check_workspace_tests as gate
class WorkspaceTestScheduling(unittest.TestCase): def test_default_inventory_keeps_units_integrations_examples_and_exclusives(self): with tempfile.TemporaryDirectory() as tmp: manifest = Path(tmp) / "Cargo.toml" manifest.write_text('[package]\nname="sample"\nversion="0.1.0"\n') targets = [ {"kind": ["lib"], "name": "sample", "test": True}, {"kind": ["bin"], "name": "app", "test": True}, {"kind": ["test"], "name": "ordinary", "test": True}, {"kind": ["test"], "name": "rlm_child_e2e", "test": True}, {"kind": ["test"], "name": "kernel_dispose_e2e", "test": True}, {"kind": ["example"], "name": "opted-in", "test": True}, {"kind": ["example"], "name": "default-example", "test": False}, {"kind": ["bench"], "name": "bench", "test": False}, ] package = {"id": "sample", "manifest_path": str(manifest), "targets": targets} ordinary, exclusive = gate.target_groups({"workspace_members": ["sample"], "packages": [package, package]}) self.assertEqual(ordinary, [("workspace-units", ["--lib", "--bins"]), ("test-rlm_child_e2e", ["--test", "rlm_child_e2e"]), ("test-ordinary", ["--test", "ordinary"]), ("example-opted-in", ["--example", "opted-in"])]) self.assertEqual(exclusive, [("test-kernel_dispose_e2e", ["--test", "kernel_dispose_e2e"]) ]) manifest.write_text('[[test]]\nname="custom"\nharness=false\n') with self.assertRaisesRegex(ValueError, "custom test harness"): gate.target_groups({"workspace_members": ["sample"], "packages": [package]})
def test_private_home_has_the_qualified_fallback_kernel(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) kernel = root / "qualified/bin/python" kernel.parent.mkdir(parents=True) kernel.write_text("approved fixture") ambient = {"PATH": "/tools", "CARGO_TARGET_DIR": str(root / "target"), "CARGO_HOME": str(root / "cargo"), "PRIME_AGENT_KERNEL_PYTHON": str(kernel), "PA_E2E_KERNEL_PYTHON": str(kernel), "OPENAI_API_KEY": "must-not-leak"} env = gate.target_environment(root, root / "isolated", ambient) self.assertEqual(Path(env["HOME"]).joinpath(".prime/agent/kernel-venv/bin/python").resolve(), kernel.resolve()) self.assertNotIn("OPENAI_API_KEY", env) self.assertEqual(env["RUST_TEST_THREADS"], "4") self.assertEqual(env["CARGO_HOME"], ambient["CARGO_HOME"])
def test_disabled_and_feature_gated_manifests_fail_closed(self): with tempfile.TemporaryDirectory() as tmp: manifest = Path(tmp) / "Cargo.toml" package = {"id": "sample", "manifest_path": str(manifest), "targets": []} for declaration in ('test=false', 'required-features=["optional"]'): manifest.write_text('[[bin]]\nname="app"\n' + declaration + '\n') with self.assertRaisesRegex(ValueError, "disabled or feature-gated"): gate.target_groups({"workspace_members": ["sample"], "packages": [package]})
def test_selector_qualified_state_names_do_not_collide(self): with tempfile.TemporaryDirectory() as tmp: manifest = Path(tmp) / "Cargo.toml" manifest.write_text('[package]\nname="sample"\nversion="0.1.0"\n') targets = [{"kind": [kind], "name": "same", "test": True} for kind in ("test", "example")] ordinary, _ = gate.target_groups({"workspace_members": ["sample"], "packages": [ {"id": "sample", "manifest_path": str(manifest), "targets": targets}]}) self.assertEqual([name for name, _ in ordinary], ["test-same", "example-same"])
def test_cleanup_uses_one_budget_and_signals_exited_leader_groups(self): from unittest.mock import Mock, call children = [Mock(pid=101), Mock(pid=102)] children[0].wait.side_effect = [subprocess.TimeoutExpired("cargo", 2), 0] children[1].wait.side_effect = [subprocess.TimeoutExpired("cargo", 0.5), 0] with patch.object(gate.os, "killpg") as killpg, \ patch.object(gate.time, "monotonic", side_effect=[100, 100, 101.5]): gate.cleanup(children) self.assertEqual(children[0].wait.call_args_list[0], call(timeout=2)) self.assertEqual(children[1].wait.call_args_list[0], call(timeout=0.5)) self.assertEqual(killpg.call_args_list, [ call(101, gate.signal.SIGTERM), call(102, gate.signal.SIGTERM), call(101, gate.signal.SIGKILL), call(102, gate.signal.SIGKILL)])
def test_expired_deadline_does_not_launch_pending_targets(self): with patch.object(gate.subprocess, "Popen") as launch: with self.assertRaisesRegex(TimeoutError, "deadline expired"): gate.run_groups([("pending", ["--test", "pending"])], repo=Path("/fixture"), state=Path("/private"), env={}, deadline=time.monotonic() - 1, jobs=2) launch.assert_not_called()
def test_signal_handler_enters_exception_cleanup_path(self): with self.assertRaises(KeyboardInterrupt): gate.interrupted(gate.signal.SIGTERM, None)
def test_sigterm_cleans_a_term_resistant_cargo_group(self): with tempfile.TemporaryDirectory(prefix="pa-ws-", dir="/tmp") as tmp: root = Path(tmp) cargo = root / "cargo" cargo.write_text(f"#!{sys.executable}\n" + "import os, signal, socket\n" "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" "s=socket.socket(socket.AF_UNIX);s.connect(os.environ['PROBE_SOCKET'])\n" "s.sendall(b'ready');s.recv(1)\n") cargo.chmod(0o700) server = socket.socket(socket.AF_UNIX) server.bind(str(root / "ready.sock")) server.listen(1) server.settimeout(5) code = ( "import os, signal, time; from pathlib import Path; " "import check_workspace_tests as g; " "g.target_environment=lambda *args: dict(os.environ); " "signal.signal(signal.SIGTERM, g.interrupted); " "g.run_groups([('probe', ['--test', 'probe'])], " "repo=Path.cwd(), state=Path(os.environ['PROBE_ROOT']), " "env=dict(os.environ), deadline=time.monotonic()+30, jobs=2)" ) env = dict(os.environ, PATH=str(root), PROBE_SOCKET=str(root / "ready.sock"), PROBE_ROOT=str(root)) driver = subprocess.Popen([sys.executable, "-B", "-c", code], cwd=Path(gate.__file__).parent, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) peer = None try: peer, _ = server.accept() peer.settimeout(5) self.assertEqual(peer.recv(100), b"ready") started = time.monotonic() driver.send_signal(gate.signal.SIGTERM) driver.communicate(timeout=4) self.assertNotEqual(driver.returncode, 0) self.assertLess(time.monotonic() - started, 4) self.assertEqual(peer.recv(1), b"") finally: if driver.poll() is None: driver.kill() driver.communicate() if peer is not None: peer.close() server.close()
def run_probe(self, fail): with tempfile.TemporaryDirectory(prefix="pa-ws-", dir="/tmp") as tmp: root = Path(tmp) cargo = root / "cargo" cargo.write_text(f"#!{sys.executable}\n" + "import os, socket, sys\n" "s=socket.socket(socket.AF_UNIX);s.connect(os.environ['PROBE_SOCKET'])\n" "s.sendall(sys.argv[-1].encode()+b'\\n')\n" "reply=s.recv(1);sys.exit(7 if reply==b'f' else 0)\n") cargo.chmod(0o700) server = socket.socket(socket.AF_UNIX) server.bind(str(root / "ready.sock")) server.listen(3) server.settimeout(5) env = dict(os.environ, PATH=str(root), PROBE_SOCKET=str(root / "ready.sock")) peers = [] try: with patch.object(gate, "target_environment", return_value=env) as private, \ ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(gate.run_groups, [(name, ["--test", name]) for name in ("a", "b", "c")], repo=root, state=root, env=env, deadline=time.monotonic()+10, jobs=2) for _ in range(2): peer, _ = server.accept() peers.append(peer) self.assertIn(peer.recv(100).strip(), (b"a", b"b")) # Both targets reached readiness before either was released. peers[0].sendall(b"f" if fail else b"x") if fail: # A failing target no longer cancels the rest: the pending # target still starts, and the failure is reported at the end. peer, _ = server.accept() peers.append(peer) self.assertEqual(peer.recv(100).strip(), b"c") for peer in peers[1:]: peer.sendall(b"x") _, failures = future.result(timeout=5) self.assertEqual([code for _, code, _ in failures], [7]) self.assertTrue((root / "c.log").exists()) else: peer, _ = server.accept() peers.append(peer) self.assertEqual(peer.recv(100).strip(), b"c") for peer in peers[1:]: peer.sendall(b"x") future.result(timeout=5) homes = [call.args[1] for call in private.call_args_list] self.assertEqual(len(homes), len(set(homes))) # A target's private root carries `TMPDIR`, so it stays short whatever # the state depth is, and it does not outlive its group. self.assertTrue(all(home.parent == Path("/tmp") and home.name.startswith("pa-t-") for home in homes)) self.assertTrue(all(not home.exists() for home in homes)) finally: for peer in peers: peer.close() server.close()
def test_only_two_targets_run_and_a_slot_is_reused_after_completion(self): self.run_probe(False)
def test_a_failure_does_not_cancel_the_remaining_targets(self): self.run_probe(True)
def test_exclusive_phase_and_doctests_follow_parallel_phase(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) with patch.object(sys, "argv", ["gate", "--state-root", str(root / "targets"), "--deadline", str(time.monotonic() + 30)]), \ patch.object(gate, "run", return_value=json.dumps({})), \ patch.object(gate, "target_groups", return_value=([("ordinary", [])], [("exclusive", [])])), \ patch.object(gate, "run_groups", return_value=([], [])) as run_groups, \ patch.object(gate, "target_environment", return_value={}), \ patch.object(gate, "run") as run: run.return_value = json.dumps({}) self.assertEqual(gate.main(), 0) self.assertEqual([call.kwargs["jobs"] for call in run_groups.call_args_list], [gate.PARALLEL_SLOTS, 1]) self.assertEqual([call.args[0] for call in run_groups.call_args_list], [[("ordinary", [])], [("exclusive", [])]]) self.assertEqual(run.call_args.args[0], ["cargo", "test", "--locked", "--offline", "--workspace", "--doc"])
if __name__ == "__main__": unittest.main()