275 lines
9.2 KiB
Python
275 lines
9.2 KiB
Python
"""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()
|