test_terminal_verify.py

"""Real-process integration tests for terminal_verify; no mocked subprocess calls."""
import json
import os
import pathlib
import sys
import tempfile
import time
from terminal_verify import run_command, run_python_file


def main():
    checks = []
    with tempfile.TemporaryDirectory(prefix="eim_terminal_test_") as d:
        root = pathlib.Path(d)
        script = root / "sample_program.py"
        script.write_text("import sys\nprint('hello from child')\nprint('diagnostic', file=sys.stderr)\n", encoding="utf-8")
        r = run_python_file("sample_program.py", workspace=root, expected_stdout="hello from child\n",
                            expected_stderr="diagnostic\n", expected_exit_code=0)
        checks.append(("actual Python child stdout/stderr/exit captured", r["status"] == "PASS" and
                       r["actual"]["stdout"] == "hello from child\n" and
                       r["actual"]["stderr"] == "diagnostic\n" and r["actual"]["exit_code"] == 0))
        bad = run_command([sys.executable, "-c", "import sys; print('wrong'); sys.exit(3)"], workspace=root,
                          expected_stdout="right\n", expected_exit_code=0)
        checks.append(("wrong output and nonzero exit independently fail", bad["status"] == "FAIL" and
                       bad["actual"]["exit_code"] == 3 and len(bad["mismatches"]) == 2))
        crash = run_command([sys.executable, "-c", "raise RuntimeError('real boom')"], workspace=root)
        checks.append(("real traceback captured from crashing process", crash["status"] == "FAIL" and
                       "RuntimeError: real boom" in crash["actual"]["stderr"] and crash["actual"]["exit_code"] != 0))
        timed = run_command([sys.executable, "-c", "import time; time.sleep(2)"], workspace=root, timeout=0.15)
        checks.append(("timeout kills a real sleeping process", timed["status"] == "TIMEOUT" and timed["timed_out"] is True))
        missing = run_python_file("missing.py", workspace=root)
        checks.append(("missing target is not reported as passed", missing["status"] == "INVALID_SCRIPT"))
        escaped = run_python_file("../outside.py", workspace=root)
        checks.append(("script path outside workspace rejected", escaped["status"] == "INVALID_SCRIPT"))
        blocked_network = run_command([sys.executable, "-c", "import socket; print('should not run')"], workspace=root)
        checks.append(("network import requires explicit authorization", blocked_network["status"] == "POLICY_BLOCKED"
                       and "--allow-network" in blocked_network["error"]))
        outside = run_command([sys.executable, "-c", "open('/etc/eim-outside-test', 'w').write('x')"], workspace=root)
        checks.append(("obvious write outside workspace is blocked before execution",
                       outside["status"] == "POLICY_BLOCKED" and "escapes the workspace" in outside["error"]))
        risky = run_command([sys.executable, "-c", "import subprocess; print('authorized')"], workspace=root,
                            allow_risky=True)
        checks.append(("explicit risky-command authorization is honoured",
                       risky["status"] == "PASS" and risky["actual"]["stdout"] == "authorized\n"))
        outside_cmd = run_command(["rm", str(root.parent / "not-to-delete")], workspace=root, allow_risky=True)
        checks.append(("outside-workspace destructive target needs separate authorization",
                       outside_cmd["status"] == "POLICY_BLOCKED" and "allow-outside-workspace" in outside_cmd["error"]))
        marker = root / "child_survived.txt"
        child_code = "import time; time.sleep(0.8); open(" + repr(str(marker)) + ", 'w').write('survived')"
        parent_code = "import subprocess,sys,time; subprocess.Popen([sys.executable,'-c'," + repr(child_code) + "]); time.sleep(20)"
        tree = run_command([sys.executable, "-c", parent_code], workspace=root, timeout=0.2, allow_risky=True)
        time.sleep(1.0)
        checks.append(("timeout kills descendants, not only the parent",
                       tree["status"] == "TIMEOUT" and not marker.exists()))
        record = root / "report.json"
        record.write_text(json.dumps(r), encoding="utf-8")
        reread = json.loads(record.read_text(encoding="utf-8"))
        checks.append(("evidence record is valid JSON and includes duration", isinstance(reread["duration_seconds"], float)))
    for name, ok in checks:
        print(("PASS " if ok else "FAIL ") + name)
    print(f"Result: {sum(ok for _, ok in checks)}/{len(checks)} passed; real subprocesses were executed.")
    return 0 if all(ok for _, ok in checks) else 1

if __name__ == "__main__":
    raise SystemExit(main())