"""Real local process fault regressions. No devices, network or user processes.""" import asyncio import hashlib import importlib.util import os from pathlib import Path import sys import subprocess from unittest.mock import AsyncMock import pytest import psutil from artemis.plugins import sdk from artemis.plugins.sdk import ( FileVerificationError, ManagedProcessError, VerifiedFile, snapshot_file, ) from artemis.plugins.sdk_checks import run_checks @pytest.mark.asyncio async def test_standard_sdk_fault_suite(): result = await run_checks() assert result["status"] == "passed" assert "descendants_reclaimed_on_cancel" in result["checks"] assert "descendants_reclaimed_on_timeout" in result["checks"] assert result["real_device_actions"] is False @pytest.mark.parametrize("max_bytes", [0, -1, True]) def test_snapshot_rejects_invalid_limits_before_open(tmp_path, max_bytes): with pytest.raises(ValueError, match="正整数"): snapshot_file(tmp_path / "missing", "0" * 64, max_bytes=max_bytes) def test_snapshot_rejects_wrong_digest_oversized_and_special_input(tmp_path): path = tmp_path / "firmware.bin" path.write_bytes(b"approved") digest = hashlib.sha256(path.read_bytes()).hexdigest() assert snapshot_file(path, digest.upper()).sha256 == digest with pytest.raises(FileVerificationError, match="批准"): snapshot_file(path, "f" * 64) with pytest.raises(FileVerificationError, match="SHA256"): snapshot_file(path, "not-a-sha256") with pytest.raises(FileVerificationError, match="大小"): snapshot_file(path, digest, max_bytes=3) with pytest.raises(FileVerificationError, match="普通文件"): snapshot_file(tmp_path, digest) with pytest.raises(TypeError): VerifiedFile(bytearray(b"mutable")) def test_snapshot_survives_path_replacement(tmp_path): path = tmp_path / "source" path.write_bytes(b"old") snapshot = snapshot_file(path, hashlib.sha256(b"old").hexdigest()) replacement = tmp_path / "new" replacement.write_bytes(b"new") replacement.replace(path) assert snapshot.data == b"old" assert b"".join(snapshot.chunks(1)) == b"old" with pytest.raises(TypeError): next(snapshot.chunks())[0] = 9 with pytest.raises(ValueError): list(snapshot.chunks(0)) @pytest.mark.asyncio @pytest.mark.parametrize( "updates", [ {"argv": "echo hello"}, {"argv": []}, {"argv": ["bad\x00program"]}, {"timeout": 0}, {"timeout": float("inf")}, {"timeout": True}, {"input_data": "not-bytes"}, {"max_output_bytes": 0}, ], ) async def test_invalid_process_inputs_do_not_launch(monkeypatch, tmp_path, updates): launch = AsyncMock() monkeypatch.setattr(asyncio, "create_subprocess_exec", launch) arguments = {"argv": [sys.executable], "cwd": tmp_path, "timeout": 5, **updates} with pytest.raises(ValueError): await sdk.run_process(**arguments) launch.assert_not_called() @pytest.mark.asyncio async def test_tool_launch_error_does_not_authorize_a_retry(tmp_path): with pytest.raises(ManagedProcessError) as raised: await sdk.run_process([str(tmp_path / "nonexistent-program")], cwd=tmp_path, timeout=5) assert raised.value.error_code == "process_launch_failed" assert raised.value.may_have_started is False assert "nonexistent-program" not in str(raised.value) @pytest.mark.asyncio async def test_combined_stderr_limit_is_enforced(tmp_path): with pytest.raises(ManagedProcessError, match="process_output_limit"): await sdk.run_process( [ sys.executable, "-I", "-B", "-c", "import os; os.write(1,b'a'*800); os.write(2,b'b'*800)", ], cwd=tmp_path, timeout=5, max_output_bytes=1000, ) @pytest.mark.asyncio async def test_process_fixed_argv_does_not_invoke_shell_and_has_explicit_environment(tmp_path): result = await sdk.run_process( [ sys.executable, "-I", "-B", "-c", "import os,sys; print(sys.argv[1]); print(os.environ.get('ARTEMIS_REVIEW_SYNTHETIC'))", "literal & | ; $(no-command)", ], cwd=tmp_path, timeout=5, env={**os.environ, "ARTEMIS_REVIEW_SYNTHETIC": "explicit"}, ) assert result.returncode == 0 assert b"literal & | ; $(no-command)" in result.stdout assert b"explicit" in result.stdout @pytest.mark.asyncio async def test_cleanup_continues_through_repeated_cancellation(): entered = asyncio.Event() release = asyncio.Event() async def cleanup(): entered.set() await release.wait() worker = asyncio.create_task(cleanup()) owner = asyncio.create_task(sdk._finish_cleanup(worker)) await entered.wait() owner.cancel() await asyncio.sleep(0) owner.cancel() await asyncio.sleep(0) assert not worker.done() release.set() with pytest.raises(asyncio.CancelledError): await owner assert worker.done() and not worker.cancelled() @pytest.mark.asyncio @pytest.mark.skipif(os.name != "nt", reason="Windows Job Object ownership barrier") async def test_ownership_failure_never_starts_tool(monkeypatch, tmp_path): marker = tmp_path / "must-not-exist" def deny_assignment(self, pid): raise OSError("synthetic assignment failure") monkeypatch.setattr(sdk.ProcessJob, "assign", deny_assignment) with pytest.raises(ManagedProcessError, match="process_ownership_failed") as raised: await sdk.run_process( [ sys.executable, "-I", "-B", "-c", "from pathlib import Path; import sys; Path(sys.argv[1]).touch()", str(marker), ], cwd=tmp_path, timeout=5, ) assert raised.value.may_have_started is False assert not marker.exists() @pytest.mark.asyncio @pytest.mark.skipif(os.name != "nt", reason="Windows owner-death Job Object guarantee") async def test_forcibly_killing_sdk_owner_reclaims_its_descendants(tmp_path): from artemis.plugins.sdk_checks import PROBE marker = tmp_path / "owned-child.json" owner_script = tmp_path / "owned-sdk-host.py" owner_script.write_text( "import asyncio, sys\nfrom pathlib import Path\n" "from artemis.plugins.sdk import run_process\n" "asyncio.run(run_process([sys.executable, '-I', '-B', sys.argv[1], 'tree', " "sys.argv[2], 'linger'], cwd=Path(sys.argv[2]).parent, timeout=60))\n", encoding="utf-8", ) owner = await asyncio.create_subprocess_exec( sys.executable, "-B", str(owner_script), str(PROBE), str(marker), stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW, ) descendants = [] try: async with asyncio.timeout(10): while not marker.exists(): assert owner.returncode is None await asyncio.sleep(0.02) descendants = psutil.Process(owner.pid).children(recursive=True) assert len(descendants) >= 3 # bootstrap, synthetic tool, synthetic sleeper owner.kill() await owner.wait() async with asyncio.timeout(3): while any(p.is_running() for p in descendants): await asyncio.sleep(0.02) finally: if owner.returncode is None: owner.kill() await owner.wait() for process in descendants: if process.is_running(): process.kill() process.wait(timeout=5) @pytest.mark.asyncio async def test_managed_template_real_mcp_and_host_receipts(): path = ( Path(__file__).resolve().parents[2] / "examples/plugins/managed-tool-demo/test_contract.py" ) spec = importlib.util.spec_from_file_location("managed_template_contract", path.resolve()) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) result = await module.run_contract() assert result["mcp_calls"] == 3 assert result["host_attachments_verified"] == 1 def test_authoring_template_and_checker_copies_stay_identical(): root = Path(__file__).resolve().parents[2] for name in ( "plugin.json", "server.py", "SKILL.md", "README.md", "test_contract.py", "LICENSE", "SOURCE.md", ): assert (root / "examples/plugins/managed-tool-demo" / name).read_bytes() == ( root / "skills/artemis-use/assets/managed-tool-demo" / name ).read_bytes() assert (root / "scripts/check_plugin.py").read_bytes() == ( root / "skills/artemis-use/scripts/check_plugin.py" ).read_bytes() for name in ("plugin.json", "server.py", "SKILL.md", "LICENSE", "SOURCE.md"): assert (root / "examples/plugins/evidence-demo" / name).read_bytes() == ( root / "skills/artemis-use/assets/evidence-demo" / name ).read_bytes() assert (root / "docs/plugin-sdk.md").read_bytes() == ( root / "skills/artemis-use/references/plugin-sdk.md" ).read_bytes()