"""Actual owned subprocess regressions; no devices, network or external tools.""" import contextlib import ctypes import json import os import subprocess import sys import tempfile import time import unittest from pathlib import Path from unittest.mock import AsyncMock, Mock, patch ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "skill/scripts")) import dependency_checks as checks import dependency_runner as dependencies import device_runtime as runtime import tool_runner as runner from test_review_regressions import process_exited def parent_code(exit_code=None, output_bytes=0): return ( "import os,subprocess,sys,time,json;from pathlib import Path;" "p=subprocess.Popen([sys.executable,'-I','-B','-c','import time;time.sleep(60)']," "stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL," "creationflags=subprocess.CREATE_NO_WINDOW);" "Path(sys.argv[1]).write_text(json.dumps({'parent':os.getpid(),'child':p.pid}));" f"sys.stdout.write('x'*{output_bytes});sys.stdout.flush();" + ("time.sleep(60)" if exit_code is None else f"sys.exit({exit_code})") ) @contextlib.contextmanager def denied_cleanup_accounting(): original = ctypes.WinDLL def load(name, *args, **kwargs): library = original(name, *args, **kwargs) if name != "kernel32": return library def deny(*args): ctypes.set_last_error(5) return 0 class Proxy: def __getattr__(self, attr): return ( deny if attr == "QueryInformationJobObject" else getattr(library, attr) ) return Proxy() with patch.object(ctypes, "WinDLL", side_effect=load): yield @unittest.skipUnless(os.name == "nt", "Windows Job ownership") class LifecycleReview(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.marker = self.root / "owned-pids.json" def assert_tree_exited(self): pids = json.loads(self.marker.read_text()) self.assertTrue(process_exited(pids["parent"])) self.assertTrue(process_exited(pids["child"])) def test_doctor_public_entry_reaps_successful_startup_descendants(self): native = dependencies prefix = [sys.executable, "-I", "-B", "-c", parent_code(0), str(self.marker)] def resolve(tool_id, profile, roots): if tool_id == "ffprobe": return profile, prefix, {"source": "synthetic", "status": "found"} raise native.UPSTREAM.ToolMissing(tool_id) # Substitute only discovery; the real doctor/check/Job processes run unchanged. with ( patch.object(native, "resolve_dependency", side_effect=resolve), patch.object(native, "path_directories", return_value=iter(())), ): result = native.doctor({}, [], check_versions=True) self.assert_tree_exited() self.assertEqual(result["execution_status"], "completed") def test_startup_check_reaps_descendant_on_normal_exit(self): result = checks.check_startup( "ffprobe", [sys.executable, "-I", "-B", "-c", parent_code(0), str(self.marker)], self.root, time.monotonic() + 8, ) self.assertEqual(result["status"], "passed") self.assert_tree_exited() def test_check_nonzero_reaps_descendants(self): result, output = checks.run_check( [sys.executable, "-I", "-B", "-c", parent_code(7), str(self.marker)], self.root, time.monotonic() + 8, ) self.assertEqual(result["status"], "failed") self.assertEqual(result["returncode"], 7) self.assertEqual(output, "") self.assert_tree_exited() def test_check_timeout_reaps_descendants(self): result, output = checks.run_check( [sys.executable, "-I", "-B", "-c", parent_code(), str(self.marker)], self.root, time.monotonic() + 2, ) self.assertEqual(result["status"], "timed_out") self.assertEqual(output, "") self.assert_tree_exited() def test_check_output_limit_stops_tree_and_discards_partial_output(self): result, output = checks.run_check( [ sys.executable, "-I", "-B", "-c", parent_code(output_bytes=2 * 1024 * 1024), str(self.marker), ], self.root, time.monotonic() + 8, ) self.assertEqual(result["error_code"], "process_output_limit") self.assertNotEqual(result["status"], "passed") self.assertEqual(output, "") self.assert_tree_exited() def test_check_owner_death_closes_job_and_reaps_descendants(self): code = ( "import sys,time;from pathlib import Path;sys.path.insert(0,sys.argv[1]);" "import dependency_checks as c;" "c.run_check([sys.executable,'-I','-B','-c',sys.argv[2],sys.argv[3]],Path(sys.argv[3]).parent,time.monotonic()+40)" ) worker = subprocess.Popen( [ sys.executable, "-I", "-B", "-c", code, str(ROOT), parent_code(), str(self.marker), ], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW, ) try: deadline = time.monotonic() + 2 while not self.marker.exists() and time.monotonic() < deadline: time.sleep(0.01) self.assertTrue(self.marker.exists()) worker.kill() worker.wait(timeout=5) self.assert_tree_exited() finally: if worker.poll() is None: worker.kill() worker.wait(timeout=5) def test_real_execution_then_cleanup_error_preserves_started(self): output = self.root / "actual-write.txt" plan = runner.Plan( "capture", "dumpcap", [ sys.executable, "-I", "-B", "-c", "import sys;from pathlib import Path;Path(sys.argv[1]).write_bytes(b'written')", str(output), ], 5, ) with denied_cleanup_accounting(): result = runner.execute(plan) self.assertEqual(output.read_bytes(), b"written") self.assertEqual(result["execution_status"], "cleanup_unconfirmed") self.assertEqual(result["error_code"], "process_cleanup_unconfirmed") self.assertTrue(result["may_have_started"]) def test_prelaunch_failure_remains_known_not_started(self): plan = runner.Plan("capture", "dumpcap", [str(self.root / "missing.exe")], 5) result = runner.execute(plan) self.assertEqual(result["execution_status"], "launch_failed") self.assertFalse(result["may_have_started"]) def test_unclassified_io_failure_is_unknown_and_redacted(self): plan = runner.Plan("capture", "dumpcap", ["synthetic"], 5) with patch.object(runner, "run_process", side_effect=OSError("private-value")): result = runner.execute(plan) self.assertEqual(result["execution_status"], "unknown") self.assertTrue(result["may_have_started"]) self.assertNotIn("private-value", json.dumps(result)) @unittest.skipUnless(os.name == "nt", "Windows lifecycle receipts") class RuntimeReceiptReview(unittest.IsolatedAsyncioTestCase): async def test_host_receipt_never_calls_started_cleanup_failure_not_started(self): with ( tempfile.TemporaryDirectory() as folder, patch.dict(os.environ, {"ARTEMIS_ARTIFACTS_DIR": folder}), ): for status, code in ( ("cleanup_unconfirmed", "process_cleanup_unconfirmed"), ("unknown", "process_io_unconfirmed"), ("output_limit", "process_output_limit"), ): payload = { "execution_status": status, "error_code": code, "may_have_started": True, } process = Mock(returncode=1) with ( patch.object( runtime.asyncio, "create_subprocess_exec", new=AsyncMock(return_value=process), ), patch.object( runtime, "read_output", new=AsyncMock(return_value=json.dumps(payload).encode()), ), ): result = await runtime.run_cli( "dependency_runner.py", ["capture"], operation="capture", write=True, ) receipt = result.structuredContent["artemis"] self.assertEqual( receipt["execution"], "partial" if status == "output_limit" else "unknown", ) self.assertEqual(receipt["side_effect"], "possible") self.assertFalse(receipt["retryable"]) self.assertEqual(receipt["error_code"], code) if __name__ == "__main__": unittest.main()