175 lines
7.3 KiB
Python
175 lines
7.3 KiB
Python
"""Static and safety tests for the guarded live OOS runner."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import run_leo_m3taversal_oos_handler_suite as suite # noqa: E402
|
|
import working_leo_m3taversal_oos_protocol as protocol_lib # noqa: E402
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_protocol_worktree(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(protocol_lib, "harness_worktree_clean", lambda: True)
|
|
|
|
|
|
def protocol() -> dict:
|
|
return protocol_lib.freeze_protocol(
|
|
"runner-test-seed",
|
|
created_at_utc="2026-07-15T00:00:00+00:00",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("grounding_mode", suite.GROUNDING_MODES)
|
|
def test_guarded_remote_script_compiles_and_installs_preexecution_gate(grounding_mode: str) -> None:
|
|
script = suite.build_guarded_remote_script(
|
|
"cafef00d",
|
|
prompts=[{"id": "P1", "dimension": "demo", "message": "Read the database only"}],
|
|
suite_mode="test-mode",
|
|
report_prefix="oos-test-report",
|
|
prompt_note="frozen test",
|
|
grounding_mode=grounding_mode,
|
|
leakage_scan={"pass": True, "exact_prompt_matches": []},
|
|
)
|
|
compile(script, "<guarded-remote>", "exec")
|
|
assert "OOS_PROMPT_LEAKAGE_SCAN = {'pass': True" in script
|
|
assert 'OOS_PROMPT_LEAKAGE_SCAN = {"pass": true' not in script
|
|
assert "install_read_only_tool_surface" in script
|
|
assert "trace_payload_sha256" in script
|
|
assert 'tool_trace_module.__file__ = "<embedded leo_tool_trace.py>"' in script
|
|
assert "LEO_TOOL_TRACE_SOURCE_SHA256" in script
|
|
assert "from leo_tool_trace import extract_kb_tool_trace" not in script
|
|
assert '"tool_trace_source_sha256": LEO_TOOL_TRACE_SOURCE_SHA256' in script
|
|
assert 'report["preexecution_safety_gate"]' in script
|
|
assert 'report["remote_temp_profile_prompt_leakage_scan"]' in script
|
|
assert '"errors": remote_scan_errors' in script
|
|
assert "scan_path.resolve(strict=True)" in script
|
|
assert "2_000_000" not in script
|
|
assert 'raise RuntimeError("OOS pre-execution safety gate failed")' in script
|
|
assert "send_message_tool_enabled" in script
|
|
assert "shell_tool_enabled" in script
|
|
assert 'structured_read_tool_name") == "teleo-kb"' in script
|
|
assert "structured_read_tool_restricted_to_frozen_kb_tool" in script
|
|
assert "expected_kb_tool_sha256=OOS_KB_TOOL_SOURCE_SHA256" in script
|
|
assert "structured_read_tool_definition_sha256" in script
|
|
assert "reassert_read_only_tool_surface" in script
|
|
assert "removed_known_runner_tools" in script
|
|
assert "final_surface_matches_preexecution" in script
|
|
assert "OOS post-runner tool surface changed" in script
|
|
assert '"scope": "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv"' in script
|
|
assert "Telegram transport is denied in the OOS no-post harness" in script
|
|
assert "runner.adapters.clear()" in script
|
|
assert 'report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)' in script
|
|
assert script.index('report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)') < script.index(
|
|
"source = SessionSource("
|
|
)
|
|
assert "timeout=180" in script
|
|
assert "timeout=300" not in script
|
|
if grounding_mode == "db_tool_ablated":
|
|
assert "shutil.rmtree(db_context_dir)" in script
|
|
|
|
|
|
def test_prompt_leakage_scan_uses_partial_markers_and_runtime_roots(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
frozen = protocol()
|
|
marker = frozen["trials"][0]["prompts"][0]["leakage_markers"][1]
|
|
runtime_root = tmp_path / "skills"
|
|
runtime_root.mkdir()
|
|
(runtime_root / "leaked.md").write_text(f"prefix {marker} suffix", encoding="utf-8")
|
|
monkeypatch.setattr(suite, "RUNTIME_PROMPT_SCAN_ROOTS", (runtime_root,))
|
|
scan = suite.prompt_leakage_scan(frozen)
|
|
assert scan["pass"] is False
|
|
assert scan["exact_prompt_matches"][0]["prompt_id"] == frozen["trials"][0]["prompts"][0]["id"]
|
|
assert scan["exact_prompt_matches"][0]["marker_sha256"]
|
|
|
|
|
|
def test_current_repo_runtime_has_no_frozen_prompt_marker_leakage() -> None:
|
|
scan = suite.prompt_leakage_scan(protocol())
|
|
assert scan["pass"] is True, scan["exact_prompt_matches"]
|
|
assert scan["scanned_files"] > 0
|
|
|
|
|
|
@pytest.mark.parametrize("root_state", ("missing", "empty", "symlink"))
|
|
def test_prompt_leakage_scan_fails_closed_without_real_coverage(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, root_state: str
|
|
) -> None:
|
|
runtime_root = tmp_path / "runtime"
|
|
if root_state == "empty":
|
|
runtime_root.mkdir()
|
|
elif root_state == "symlink":
|
|
target = tmp_path / "target"
|
|
target.mkdir()
|
|
(target / "runtime.py").write_text("safe runtime text", encoding="utf-8")
|
|
runtime_root.symlink_to(target, target_is_directory=True)
|
|
monkeypatch.setattr(suite, "RUNTIME_PROMPT_SCAN_ROOTS", (runtime_root,))
|
|
|
|
scan = suite.prompt_leakage_scan(protocol())
|
|
|
|
assert scan["pass"] is False
|
|
assert scan["expected_roots"][0]["exists"] is (root_state != "missing")
|
|
assert scan["scanned_files"] == 0
|
|
|
|
|
|
def test_prompt_leakage_scan_fails_closed_on_nested_symlink(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
runtime_root = tmp_path / "runtime"
|
|
runtime_root.mkdir()
|
|
(runtime_root / "runtime.py").write_text("safe runtime text", encoding="utf-8")
|
|
hidden = tmp_path / "hidden"
|
|
hidden.mkdir()
|
|
(hidden / "prompt.md").write_text("unscanned prompt text", encoding="utf-8")
|
|
(runtime_root / "linked").symlink_to(hidden, target_is_directory=True)
|
|
monkeypatch.setattr(suite, "RUNTIME_PROMPT_SCAN_ROOTS", (runtime_root,))
|
|
|
|
scan = suite.prompt_leakage_scan(protocol())
|
|
|
|
assert scan["pass"] is False
|
|
assert scan["scanned_files"] == 1
|
|
assert len(scan["symlink_entries"]) == 1
|
|
|
|
|
|
def test_restart_probe_requires_full_fingerprint_guard_and_cleanup() -> None:
|
|
counts = {"public.claims": 2}
|
|
fingerprint = {"status": "ok", "fingerprint_sha256": "a" * 64}
|
|
report = {
|
|
"remote_returncode": 0,
|
|
"pass_runtime": True,
|
|
"posted_to_telegram": False,
|
|
"db_counts_changed": False,
|
|
"db_counts_before": counts,
|
|
"db_counts_after": counts,
|
|
"db_fingerprint_unchanged": True,
|
|
"db_fingerprint_before": fingerprint,
|
|
"db_fingerprint_after": fingerprint,
|
|
"preexecution_safety_gate": {"status": "pass"},
|
|
"telegram_transport_deny": {
|
|
"enabled": True,
|
|
"attempt_count": 0,
|
|
"runner_adapters_empty": True,
|
|
},
|
|
"temp_profile_removed": True,
|
|
"post_run_orphan_readback": {"no_matching_processes": True},
|
|
}
|
|
assert suite._restart_probe_passes(report) is True
|
|
for path in (
|
|
("db_fingerprint_unchanged",),
|
|
("preexecution_safety_gate", "status"),
|
|
("post_run_orphan_readback", "no_matching_processes"),
|
|
("telegram_transport_deny", "enabled"),
|
|
):
|
|
broken = {**report}
|
|
if len(path) == 1:
|
|
broken[path[0]] = False
|
|
else:
|
|
broken[path[0]] = {**report[path[0]], path[1]: False}
|
|
assert suite._restart_probe_passes(broken) is False
|
|
|
|
|
|
def test_portable_artifact_paths_are_repo_relative() -> None:
|
|
assert suite._portable_artifact_path(ROOT / "pyproject.toml") == "pyproject.toml"
|