NexusOS/tests/test_code_run.py

"""The execution track: synapse/code_run.py.

Python is the only toolchain guaranteed present (it is the interpreter running
these tests), so it carries the behavioural coverage — limits, streams, exit
codes, isolation. The compiled languages are covered for the parts that hold
without their toolchain installed (screening, argv shape, the missing-toolchain
message) and skipped where they need it, so this file passes on a machine with
no clang and no rustc as well as on one with both.
"""
import shutil
import sys

import pytest

from synapse import code_run


def _requires(lang):
    entry = code_run.RUN_LANGS[lang]
    return pytest.mark.skipif(
        not entry["tool"](), reason=f"no {lang} toolchain on this machine"
    )


# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------

def test_every_lang_has_a_complete_driver():
    for name, entry in code_run.RUN_LANGS.items():
        assert entry["summary"], name
        assert callable(entry["tool"]), name
        assert callable(entry["source_name"]), name
        assert callable(entry["run"]), name
        assert callable(entry["critique"]), name
        assert entry["install"], name


def test_aliases_resolve_to_real_langs():
    for alias, target in code_run.ALIASES.items():
        assert target in code_run.RUN_LANGS, alias
    assert code_run.resolve_lang("C++") == "cpp"
    assert code_run.resolve_lang(" Py ") == "python"
    assert code_run.resolve_lang("javascript") == "javascript"  # unknown passes through


def test_erlang_source_name_follows_the_dialect():
    """escript reads the same bytes two ways depending on the extension. Writing
    a bare main/1 script to main.erl makes it a module with no exports, which
    fails at load with an error that says nothing about the real problem."""
    assert code_run._erlang_source_name("main(_) -> ok.") == "main.escript"
    assert code_run._erlang_source_name("-module(main).\nmain(_) -> ok.") == "main.erl"


def test_erlang_escript_gets_a_shebang_when_missing():
    """OTP 28+ rejects shebang-less .escript with 'Premature end of file'."""
    bare = "main(_) -> ok.\n"
    out = code_run._erlang_write_source(bare)
    assert out.startswith("#!/usr/bin/env escript\n")
    assert out.endswith(bare)
    already = "#!/usr/bin/env escript\nmain(_) -> ok.\n"
    assert code_run._erlang_write_source(already) == already
    module = "-module(main).\n-export([main/1]).\nmain(_) -> ok.\n"
    assert code_run._erlang_write_source(module) == module


# ---------------------------------------------------------------------------
# Screening
# ---------------------------------------------------------------------------

def test_clean_snippets_pass_screening():
    assert code_run.critique("python", "print(sum(range(100)))") == []
    assert code_run.critique("c", '#include <stdio.h>\nint main(void){puts("x");}') == []
    assert code_run.critique("rust", 'fn main(){ println!("x"); }') == []
    assert code_run.critique("erlang", 'main(_) -> io:format("x~n").') == []


@pytest.mark.parametrize("lang,source,needle", [
    ("python", "import socket\nprint(socket)", "network"),
    ("python", "import requests\nprint(requests)", "network"),
    ("python", "import subprocess\nsubprocess.run(['ls'])", "processes"),
    ("c", "#include <sys/socket.h>\nint main(void){return 0;}", "network"),
    ("c", '#include <stdlib.h>\nint main(void){system("ls");}', "processes"),
    ("rust", "fn main(){ std::process::Command::new(\"ls\"); }", "processes"),
    ("erlang", 'main(_) -> os:cmd("ls").', "processes"),
])
def test_screening_catches_the_obvious_reaches(lang, source, needle):
    issues = code_run.critique(lang, source)
    assert issues, f"{lang} snippet passed screening"
    assert any(needle in i for i in issues), issues


def test_screening_catches_absolute_paths_and_urls():
    assert any("absolute" in i for i in code_run.critique(
        "python", 'data = open("/etc/passwd").read()\nprint(data)'))
    # Escaped and raw Windows paths both count.
    assert any("absolute" in i for i in code_run.critique(
        "python", 'data = open("C:\\\\Users\\\\me").read()'))
    assert any("absolute" in i for i in code_run.critique(
        "python", r'data = open(r"C:\Users\me").read()'))
    assert any("http" in i for i in code_run.critique(
        "python", 'print("see https://example.com/data.csv")'))


def test_url_in_a_comment_is_not_a_network_reach():
    """A citation in a comment is not fetch(); bouncing it costs a useless retry."""
    c = (
        "/* spec: https://en.cppreference.com/w/c/string */\n"
        "#include <stdio.h>\n"
        "int main(void){ puts(\"ok\"); return 0; }\n"
    )
    assert code_run.critique("c", c) == []


def test_erlang_nproc_ceiling_is_relative_to_the_user(monkeypatch):
    """RLIMIT_NPROC is UID-scoped; an absolute 512 is not 512 of headroom."""
    monkeypatch.setattr(code_run, "_user_process_count", lambda: 400)
    assert code_run._max_procs_for("erlang") == 400 + code_run._ERLANG_PROC_HEADROOM
    assert code_run._max_procs_for("python") == code_run._MAX_PROCS
    monkeypatch.setattr(code_run, "_user_process_count", lambda: None)
    assert code_run._max_procs_for("erlang") is None


def test_missing_entry_point_is_named_up_front():
    """Without this the user sees a linker error about _main, or an escript load
    failure — neither of which says 'you forgot the entry point'."""
    assert any("main()" in i for i in code_run.critique("c", "int add(int a){return a+1;}"))
    assert any("main()" in i for i in code_run.critique("rust", "fn add(a: i32) -> i32 { a }"))
    assert any("main/1" in i for i in code_run.critique("erlang", "add(A) -> A + 1."))


def test_empty_source_says_so_and_stops():
    issues = code_run.critique("python", "  ")
    assert issues == ["source is empty or too short to run."]


def test_unknown_lang_is_refused_by_both_entry_points():
    assert code_run.critique("brainfuck", "+++.")[0].startswith("cannot run")
    assert code_run.run("brainfuck", "+++.")["ok"] is False


def test_oversized_source_is_refused_without_running():
    issues = code_run.critique("python", "x = 1\n" * 40_000)
    assert issues and "characters" in issues[0]


# ---------------------------------------------------------------------------
# Execution (python: always available)
# ---------------------------------------------------------------------------

def test_stdout_and_exit_code_come_back():
    out = code_run.run("python", "print('hello', 6 * 7)")
    assert out["ok"] is True
    assert out["stdout"].strip() == "hello 42"
    assert out["exit_code"] == 0


def test_streams_are_kept_separate():
    out = code_run.run("python", "import sys\nprint('o')\nprint('e', file=sys.stderr)")
    assert out["stdout"].strip() == "o"
    assert out["stderr"].strip() == "e"


def test_nonzero_exit_is_still_a_successful_run():
    out = code_run.run("python", "raise SystemExit(3)")
    assert out["ok"] is True
    assert out["exit_code"] == 3


def test_a_traceback_is_returned_not_raised():
    out = code_run.run("python", "print(1/0)")
    assert out["ok"] is True
    assert out["exit_code"] != 0
    assert "ZeroDivisionError" in out["stderr"]


def test_stdin_is_piped_in():
    out = code_run.run("python", "import sys\nprint(sys.stdin.read().upper())", stdin="abc")
    assert out["stdout"].strip() == "ABC"


def test_a_program_reading_stdin_with_none_given_does_not_hang():
    """stdin is always closed rather than inherited: a snippet calling input()
    with nothing piped in would otherwise block on the server's own stdin until
    the wall clock killed it, five seconds of 'thinking' for nothing."""
    out = code_run.run("python", "print(input('prompt: '))")
    assert out["ok"] is True
    assert "EOFError" in out["stderr"]


def test_an_infinite_loop_is_killed_and_explained():
    out = code_run.run("python", "while True:\n    pass")
    assert out["ok"] is False
    assert out["stage"] == "run"
    assert "did not finish" in out["error"]


def test_output_is_truncated_not_streamed_whole():
    out = code_run.run("python", f"print('x' * {code_run.MAX_OUTPUT * 3})")
    assert len(out["stdout"]) < code_run.MAX_OUTPUT + 200
    assert "truncated" in out["stdout"]


def test_the_working_directory_is_ephemeral():
    """Two runs must not see each other's files. The temp dir is the only thing
    standing between a snippet and the user's cwd, so its lifetime is worth a
    test rather than an assumption."""
    first = code_run.run("python", "open('note.txt', 'w').write('hi')\nprint('wrote')")
    assert first["ok"] is True, first
    second = code_run.run("python", "import os\nprint(os.path.exists('note.txt'))")
    assert second["stdout"].strip() == "False"


def test_the_child_does_not_inherit_the_servers_environment():
    """A snippet has no business reading the process environment it happens to
    be spawned from — that is where an API key would be."""
    import os
    os.environ["NEXUS_RUN_LEAK_PROBE"] = "secret"
    try:
        out = code_run.run(
            "python", "import os\nprint(os.environ.get('NEXUS_RUN_LEAK_PROBE'))"
        )
    finally:
        os.environ.pop("NEXUS_RUN_LEAK_PROBE", None)
    assert out["stdout"].strip() == "None"


def test_home_points_at_the_scratch_dir():
    """HOME is rewritten so a snippet writing a dotfile writes it somewhere that
    gets deleted, rather than into the user's real home."""
    out = code_run.run("python", "import os\nprint(os.path.expanduser('~'))")
    assert "nexus-run-" in out["stdout"]


# ---------------------------------------------------------------------------
# Toolchains
# ---------------------------------------------------------------------------

def test_a_missing_toolchain_explains_itself(monkeypatch):
    """The model has to be able to tell 'you cannot run this here' from 'your
    code is wrong' — otherwise it rewrites a correct program repeatedly."""
    monkeypatch.setitem(
        code_run.RUN_LANGS["rust"], "tool", lambda: None
    )
    out = code_run.run("rust", 'fn main(){ println!("x"); }')
    assert out["ok"] is False
    assert out["stage"] == "toolchain"
    assert "rustup.rs" in out["error"]
    assert "Show the code instead" in out["error"]


@_requires("c")
def test_c_compiles_and_runs():
    out = code_run.run("c", '#include <stdio.h>\nint main(void){printf("%d\\n", 6*7);return 0;}')
    assert out["ok"] is True, out
    assert out["stdout"].strip() == "42"


@_requires("c")
def test_a_compile_error_comes_back_as_a_compile_error():
    """Stage matters: the compiler's diagnostics are the useful payload, and
    labelling this a run failure would hide that nothing ever executed."""
    out = code_run.run("c", "int main(void){ return oops; }")
    assert out["ok"] is False
    assert out["stage"] == "compile"
    assert "oops" in out["stderr"]


@_requires("cpp")
def test_cpp_compiles_and_runs():
    out = code_run.run("cpp", '#include <iostream>\nint main(){std::cout << 6*7 << "\\n";}')
    assert out["ok"] is True, out
    assert out["stdout"].strip() == "42"


@_requires("rust")
def test_rust_compiles_and_runs():
    out = code_run.run("rust", 'fn main(){ println!("{}", (1..=10).sum::<i32>()); }')
    assert out["ok"] is True, out
    assert out["stdout"].strip() == "55"


@_requires("erlang")
def test_erlang_runs_a_bare_escript():
    out = code_run.run("erlang", 'main(_) -> io:format("~p~n", [lists:sum(lists:seq(1,10))]).')
    assert out["ok"] is True, out
    assert out["stdout"].strip() == "55"


@_requires("erlang")
def test_erlang_runs_a_module_form_script():
    out = code_run.run(
        "erlang",
        "-module(main).\n-export([main/1]).\nmain(_) -> io:format(\"~p~n\", [7*6]).",
    )
    assert out["ok"] is True, out
    assert out["stdout"].strip() == "42"


@pytest.mark.skipif(sys.platform != "linux", reason="unshare is Linux-only")
def test_network_isolation_probe_is_honest():
    """Either we got a namespace or we did not — but the probe must never claim
    one it did not verify, because the tool description promises 'no network' on
    the strength of it."""
    prefix = code_run._net_isolation()
    assert prefix in ([], ["unshare", "-rn"])
    if prefix:
        assert shutil.which("unshare")