arena/tests/test_probe.py

from __future__ import annotations

import re
import subprocess
import sys
import tomllib
import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
ROW = re.compile(r"^(?P<name>\S(?:.*\S)?)\s{2,}(?:-?\d+\.\d+\s*){4}\s+(?P<verdict>.+?)$")

# The environments that are allowed to carry no probe yet, because their taskset layer is not
# written. Everything discovered and NOT in here MUST be gated.
#
# This list is the floor, and it is why it exists rather than a bare count: without it, both
# loops below are vacuous when nothing is gated, and CI goes green while gating zero
# environments. Commenting out a single `@probes(...)` decorator used to pass — the degated
# name simply moved into the warning line, which already legitimately carries five names.
#
# It only ever shrinks. Delete a name here in the same commit that lands its probe.
PENDING_PROBES = frozenset({
    "grand-exchange-live",
    "tera-crow-nav",
    "tera-drive-101",
    "tera-office-nav",
    "tera-california-flight",
})


def discovered_tasksets() -> set[str]:
    """The same denominator `probe.py:discover()` uses, derived independently of it.

    Deriving it a second time here rather than importing `probe` is the point: if discovery
    and the manifests ever disagree, this test is what says so. A directory with no
    `pyproject.toml` is not an environment to either of us.
    """
    names: set[str] = set()
    for manifest in sorted((ROOT / "environments").glob("*/pyproject.toml")):
        data = tomllib.loads(manifest.read_text(encoding="utf-8"))
        declared = data.get("tool", {}).get("arena", {}).get("tasksets")
        names.update(declared or [data["project"]["name"]])
    return names


class RootProbeIntegrationTests(unittest.TestCase):
    def test_public_root_probe_gates_every_environment(self) -> None:
        completed = subprocess.run(
            [sys.executable, "probe.py"],
            cwd=ROOT,
            text=True,
            capture_output=True,
            # The probe sweeps constants and runs multi-block ladders, and the interactive
            # environments will make it dearer still. A tight timeout here only ever fails a
            # slow machine, never a wrong reward.
            timeout=180,
            check=False,
        )
        detail = f"\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
        self.assertEqual(completed.returncode, 0, msg=f"root probe failed:{detail}")

        discovered = discovered_tasksets()
        self.assertTrue(discovered, msg="no environment manifests found — the scan is empty")

        rows = {}
        for line in completed.stdout.splitlines()[1:]:
            match = ROW.match(line.rstrip())
            if match:
                rows[match["name"]] = match["verdict"].strip()

        for name, verdict in rows.items():
            with self.subTest(environment=name):
                self.assertIn(name, discovered, msg=f"probe gated {name}, no manifest declares it")
                self.assertEqual(verdict, "ok", msg=f"{name} is not ok:{detail}")

        # Anything discovered and not gated has to be named in the warning. That warning is
        # temporary — it becomes a hard exit 1 in probe.py once the last unregistered
        # environment lands its probe — and this assertion holds either way.
        ungated = discovered - set(rows)
        for name in sorted(ungated):
            with self.subTest(environment=name):
                self.assertIn(name, completed.stderr,
                              msg=f"{name} is neither gated nor reported as ungated:{detail}")

        # The floor. Being named in the warning is not a licence to stop being gated: an
        # environment may only be ungated if it is on the pending list. Without this, removing
        # a `@probes(...)` decorator moves the name from the table into the warning and every
        # assertion above stays vacuously true.
        self.assertLessEqual(
            ungated, PENDING_PROBES,
            msg=f"these were gated and no longer are: {sorted(ungated - PENDING_PROBES)}{detail}",
        )
        self.assertTrue(rows, msg=f"the probe gated nothing at all:{detail}")


if __name__ == "__main__":
    unittest.main()