Agentswarm/orchestrator/sandbox_runner.py

"""In-sandbox test harness — runs INSIDE the isolated workdir as a child process.

Prefers **pytest** (native support for pytest-style classes, fixtures, parametrize, marks AND
unittest.TestCase) — the test styles agents actually produce. Falls back to a stdlib-only
collector (unittest.TestCase + bare module-level ``test_*`` functions) when pytest is absent, so
the harness still works without the dependency. Writes a machine-readable ``_result.json``
({total, passed, failed, errored, details}); the parent (orchestrator/sandbox.py) reads that
file and never trusts stdout for counts.

Copied into the ephemeral sandbox workdir at run time and executed there with the workdir as CWD.
"""
import json
import os
import sys

RESULT_FILE = "_result.json"


def _run_with_pytest(workdir: str) -> dict:
    """Run every test under workdir with pytest; collect pass/fail/error via an inline plugin.
    Native support for pytest classes/fixtures/parametrize/marks + unittest.TestCase."""
    import pytest

    class _Collector:
        def __init__(self):
            self.total = 0
            self.passed = 0
            self.failed = 0
            self.errored = 0
            self.details = []

        def pytest_runtest_logreport(self, report):
            text = (getattr(report, "longreprtext", "") or "")[:500]
            if report.when == "call":
                self.total += 1
                if report.outcome == "passed":
                    self.passed += 1
                    self.details.append({"test": report.nodeid, "status": "passed"})
                else:
                    self.failed += 1
                    self.details.append({"test": report.nodeid, "status": "failed", "error": text})
            elif report.when in ("setup", "teardown") and report.outcome == "failed":
                # setup/teardown failure (e.g. fixture error) counts as one errored test
                self.total += 1
                self.failed += 1
                self.errored += 1
                self.details.append({"test": report.nodeid, "status": "error", "error": text})

        def pytest_collectreport(self, report):
            # import/collection failure (e.g. missing dependency) counts as one errored test
            if report.failed:
                text = (getattr(report, "longreprtext", "") or "")[:500]
                self.total += 1
                self.failed += 1
                self.errored += 1
                self.details.append({"test": report.nodeid or "collection",
                                     "status": "error", "error": text})

    collector = _Collector()
    # -q quiet, disable cache writes, ignore any repo pytest config so the sandbox is hermetic
    pytest.main(["-q", "-p", "no:cacheprovider", "--no-header", "-o", "addopts=", workdir],
                plugins=[collector])
    return {"total": collector.total, "passed": collector.passed,
            "failed": collector.failed, "errored": collector.errored, "details": collector.details}


def _run_with_stdlib(workdir: str) -> dict:
    """Fallback when pytest is unavailable: unittest.TestCase + bare module-level test_* functions."""
    import importlib.util
    import unittest

    def _load_module(path):
        name = "sbx_" + os.path.splitext(os.path.basename(path))[0]
        spec = importlib.util.spec_from_file_location(name, path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module

    test_files = sorted(f for f in os.listdir(workdir)
                        if f.startswith("test_") and f.endswith(".py"))
    total = passed = failed = errored = 0
    details = []
    suite = unittest.TestSuite()
    bare_funcs = []
    for tf in test_files:
        try:
            module = _load_module(os.path.join(workdir, tf))
        except Exception as exc:
            total += 1; errored += 1; failed += 1
            details.append({"test": tf, "status": "error", "error": repr(exc)})
            continue
        suite.addTests(unittest.defaultTestLoader.loadTestsFromModule(module))
        for attr in dir(module):
            if not attr.startswith("test_"):
                continue
            obj = getattr(module, attr)
            if callable(obj) and not isinstance(obj, type) and getattr(obj, "__module__", None) == module.__name__:
                bare_funcs.append((f"{tf}::{attr}", obj))
    ut_result = unittest.TestResult()
    suite.run(ut_result)
    total += ut_result.testsRun
    ut_failed = len(ut_result.failures) + len(ut_result.errors)
    failed += ut_failed
    errored += len(ut_result.errors)
    passed += ut_result.testsRun - ut_failed
    for label, fn in bare_funcs:
        total += 1
        try:
            fn(); passed += 1
            details.append({"test": label, "status": "passed"})
        except AssertionError as exc:
            failed += 1
            details.append({"test": label, "status": "failed", "error": str(exc)})
        except Exception as exc:
            failed += 1; errored += 1
            details.append({"test": label, "status": "error", "error": repr(exc)})
    return {"total": total, "passed": passed, "failed": failed, "errored": errored, "details": details}


def main() -> None:
    workdir = os.getcwd()
    sys.path.insert(0, workdir)
    try:
        import pytest  # noqa: F401
        result = _run_with_pytest(workdir)
    except ImportError:
        result = _run_with_stdlib(workdir)
    with open(os.path.join(workdir, RESULT_FILE), "w", encoding="utf-8") as fh:
        json.dump(result, fh)


if __name__ == "__main__":
    try:
        main()
    except Exception as exc:  # never leave the parent without a result file
        with open(RESULT_FILE, "w", encoding="utf-8") as fh:
            json.dump({"total": 0, "passed": 0, "failed": 0, "errored": 1, "fatal": repr(exc)}, fh)