NexusOS/tests/test_snippet_probes.py

"""Automatic code-snippet probe suite.

Walks the catalog in tests/snippet_probes/catalog.py and, for every probe:

  * screen — asserts critique rejects it (never executed)
  * run    — asserts critique is clean, then runs via synapse.code_run
             (skipped cleanly when the host has no toolchain)
  * tool   — same as run, plus tools.dispatch("run_snippet") envelope checks

Adding a language to RUN_LANGS without a smoke probe fails
test_every_run_lang_has_a_smoke_probe. That is the point of the catalog:
coverage is automatic and visible.
"""
from __future__ import annotations

import asyncio
import json

import pytest

from synapse import code_run, tools
from tests.snippet_probes.catalog import PROBES, Probe


def _has_toolchain(lang: str) -> bool:
    key = code_run.resolve_lang(lang)
    entry = code_run.RUN_LANGS.get(key)
    return bool(entry and entry["tool"]())


def _ids(probes=PROBES):
    return [p.id for p in probes]


# ---------------------------------------------------------------------------
# Catalog integrity (runs even when every compiled toolchain is missing)
# ---------------------------------------------------------------------------

def test_probe_ids_are_unique():
    ids = [p.id for p in PROBES]
    assert len(ids) == len(set(ids)), "duplicate probe ids in catalog"


def test_every_run_lang_has_a_smoke_probe():
    """New RUN_LANGS entry without a smoke probe = silent blind spot."""
    smoke = {
        code_run.resolve_lang(p.lang)
        for p in PROBES
        if "smoke" in p.tags and p.kind in ("run", "tool")
    }
    missing = set(code_run.RUN_LANGS) - smoke
    assert not missing, (
        f"RUN_LANGS without a smoke probe: {sorted(missing)}. "
        "Add a Probe(..., tags=('smoke', ...)) to tests/snippet_probes/catalog.py."
    )


def test_every_run_lang_has_a_screen_probe():
    screened = {
        code_run.resolve_lang(p.lang)
        for p in PROBES
        if p.kind == "screen"
    }
    missing = set(code_run.RUN_LANGS) - screened
    assert not missing, (
        f"RUN_LANGS without a screening probe: {sorted(missing)}."
    )


def test_catalog_langs_resolve_into_run_langs_or_aliases():
    for probe in PROBES:
        key = code_run.resolve_lang(probe.lang)
        assert key in code_run.RUN_LANGS, (
            f"probe {probe.id!r} lang={probe.lang!r} resolves to unknown {key!r}"
        )


# ---------------------------------------------------------------------------
# Per-probe execution
# ---------------------------------------------------------------------------

@pytest.mark.parametrize("probe", PROBES, ids=_ids())
def test_snippet_probe(probe: Probe):
    if probe.kind == "screen":
        _assert_screen(probe)
        return

    issues = code_run.critique(probe.lang, probe.source)
    assert issues == [], f"{probe.id}: unexpected critique issues: {issues}"

    if not _has_toolchain(probe.lang):
        pytest.skip(f"no {code_run.resolve_lang(probe.lang)} toolchain on this machine")

    result = code_run.run(probe.lang, probe.source, stdin=probe.stdin)
    _assert_run_result(probe, result)

    if probe.kind == "tool":
        _assert_tool_envelope(probe, result)


def _assert_screen(probe: Probe) -> None:
    assert probe.screen_needle, f"{probe.id}: screen probe needs screen_needle"
    issues = code_run.critique(probe.lang, probe.source)
    assert issues, f"{probe.id}: expected screening to reject the snippet"
    assert any(probe.screen_needle in i for i in issues), (
        f"{probe.id}: needle {probe.screen_needle!r} not in {issues}"
    )
    # Screening is the whole point — do not execute a rejected snippet.
    # (A future change that runs despite issues would be a security regression.)


def _assert_run_result(probe: Probe, result: dict) -> None:
    assert result.get("ok") is probe.expect_ok, (
        f"{probe.id}: ok={result.get('ok')} expected {probe.expect_ok}; full={result}"
    )
    if probe.expect_stage is not None:
        assert result.get("stage") == probe.expect_stage, result

    if not probe.expect_ok:
        # Failure path: still check optional stderr/stdout breadcrumbs.
        for needle in probe.expect_stderr_contains:
            assert needle in (result.get("stderr") or ""), result
        for needle in probe.expect_stdout_contains:
            assert needle in (result.get("stdout") or ""), result
        return

    stdout = (result.get("stdout") or "").strip()
    stderr = (result.get("stderr") or "")
    if probe.expect_stdout is not None:
        assert stdout == probe.expect_stdout, (
            f"{probe.id}: stdout {stdout!r} != {probe.expect_stdout!r}"
        )
    for needle in probe.expect_stdout_contains:
        assert needle in stdout, result
    for needle in probe.expect_stderr_contains:
        assert needle in stderr, result
    if probe.expect_exit is not None:
        assert result.get("exit_code") == probe.expect_exit, result
    else:
        # Explicit "don't care" still requires that a run happened.
        assert "exit_code" in result, result


def _assert_tool_envelope(probe: Probe, direct: dict) -> None:
    out = json.loads(asyncio.run(tools.dispatch("run_snippet", {
        "lang": probe.lang,
        "source": probe.source,
        "stdin": probe.stdin,
    })))
    assert out.get("ok") is probe.expect_ok, out
    assert "fence" in out and out["fence"].startswith("```nexus-run\n"), out
    body = out["fence"].split("\n", 1)[1].rsplit("\n", 1)[0]
    envelope = json.loads(body)
    assert envelope.get("lang") == code_run.resolve_lang(probe.lang)
    if probe.expect_stdout is not None:
        assert (envelope.get("stdout") or "").strip() == probe.expect_stdout
    # Direct driver and tool path must agree on exit for successful runs.
    if probe.expect_ok and probe.expect_exit is not None:
        assert out.get("exit_code") == direct.get("exit_code") == probe.expect_exit


# ---------------------------------------------------------------------------
# One-shot inventory (useful when running the file directly)
# ---------------------------------------------------------------------------

def test_probe_inventory_lists_toolchain_readiness():
    """Not an assertion about readiness — just fails if the inventory shape
    breaks, so `pytest -k inventory -s` is a quick host capability dump."""
    rows = []
    for name, entry in code_run.RUN_LANGS.items():
        tool = entry["tool"]()
        n = sum(1 for p in PROBES if code_run.resolve_lang(p.lang) == name)
        rows.append({"lang": name, "ready": bool(tool), "probes": n, "tool": tool})
    assert rows and all(r["probes"] >= 1 for r in rows)
    # Printed only under -s; kept as a structured object for debuggability.
    print("snippet-probe inventory:", json.dumps(rows, indent=2))