Agentswarm/orchestrator/sandbox_runner.py

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

Stdlib only (no pytest dependency): collects both `unittest.TestCase` tests and bare
`test_*` functions from every `test_*.py` in the working directory, runs them, and writes a
machine-readable `_result.json` ({total, passed, failed, errored, details}). The parent
(orchestrator/sandbox.py) reads that file; it never trusts stdout for counts.

This file is copied into the ephemeral sandbox workdir at run time and executed there with the
workdir as CWD. It must stay self-contained and import nothing outside the stdlib.
"""
import importlib.util
import json
import os
import sys
import unittest

RESULT_FILE = "_result.json"


def _load_module(path: str):
    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)  # may raise on import/collection error
    return module


def main() -> None:
    workdir = os.getcwd()
    sys.path.insert(0, workdir)
    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 = []  # (label, callable)

    for tf in test_files:
        try:
            module = _load_module(os.path.join(workdir, tf))
        except Exception as exc:  # import-time failure counts as one errored test
            total += 1
            errored += 1
            failed += 1
            details.append({"test": tf, "status": "error", "error": repr(exc)})
            continue
        # unittest.TestCase-style tests
        suite.addTests(unittest.defaultTestLoader.loadTestsFromModule(module))
        # bare pytest-style functions defined in this 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))

    # Run the unittest suite (TestCase subclasses).
    ut_result = unittest.TestResult()
    suite.run(ut_result)
    ut_total = ut_result.testsRun
    ut_failed = len(ut_result.failures) + len(ut_result.errors)
    total += ut_total
    failed += ut_failed
    errored += len(ut_result.errors)
    passed += ut_total - ut_failed

    # Run bare test_* functions.
    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)})

    with open(os.path.join(workdir, RESULT_FILE), "w", encoding="utf-8") as fh:
        json.dump(
            {"total": total, "passed": passed, "failed": failed, "errored": errored, "details": details},
            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)