"""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))