""" executors/test_runner.py ------------------------ Test Runner for AutoDevAgent. Executes the test suite produced by the TestGeneratorAgent against the generated code. Returns a typed TestResult with pass/fail counts, individual failure messages, and full output. Design: - Python tests: runs the test file in a subprocess using `python -m unittest` so tests are fully isolated from the parent process. Parses the unittest output to extract counts and individual failure messages. - SQL tests: the generated SQL test file is also a Python file (using sqlite3 + unittest), so it runs through the same subprocess path. No special SQL handling needed here. - If tests fail, the debug loop re-engages — the TestResult failure messages are passed back to the DebugAgent as context. - Uses the same timeout as Python execution to prevent runaway tests. Usage: from executors.test_runner import TestRunner from pipeline.state import PipelineState runner = TestRunner() # state.generated_tests must be populated by TestGeneratorAgent updated = runner.run(state) print(updated["test_result"].passed) print(updated["test_result"].failures) """ import logging import os import re import subprocess import sys import tempfile import time from typing import Any from config import settings from pipeline.state import ( PipelineState, PipelineStatus, TestResult, ) logger = logging.getLogger(__name__) class TestRunner: """ Runs generated test suites in a sandboxed subprocess. Writes the test code to a temp file, executes it with `python -m unittest`, parses the output into a typed TestResult, and cleans up. Attributes: timeout: Maximum seconds allowed for the test run. """ def __init__(self) -> None: """Initialise with timeout from central config.""" # Tests get the same generous timeout as code execution self.timeout: int = settings.python_execution_timeout def run(self, state: PipelineState) -> dict[str, Any]: """ Execute the generated tests and return a typed TestResult. Args: state: Current PipelineState. Reads: generated_tests. Returns: Partial state dict with keys: - "test_result": TestResult - "status": PipelineStatus.TESTING """ test_code = state.generated_tests if not test_code or not test_code.strip(): logger.warning("TestRunner received empty test code.") return { "test_result": TestResult( passed=False, output="No test code was provided.", failures=["No test code was provided to run."], ), "status": PipelineStatus.TESTING, } logger.info( "TestRunner executing %d lines of test code", len(test_code.splitlines()), ) tmp_path = None try: # Write test code to a named temp file tmp_file = tempfile.NamedTemporaryFile( mode="w", suffix=".py", delete=False, encoding="utf-8", ) tmp_file.write(test_code) tmp_file.close() tmp_path = tmp_file.name result = self._execute_tests(tmp_path) finally: if tmp_path and os.path.exists(tmp_path): os.unlink(tmp_path) return { "test_result": result, "status": PipelineStatus.TESTING, } # ---------------------------------------------------------------- # # Internal execution # # ---------------------------------------------------------------- # def _execute_tests(self, file_path: str) -> TestResult: """ Run the test file directly with python file.py and parse output. The test file must call unittest.main(verbosity=2) so we get per-test pass/fail status in the output, which we parse into structured counts and failure messages. Args: file_path: Absolute path to the temporary test file. Returns: TestResult with pass/fail counts, output, and failures list. """ start = time.perf_counter() try: proc = subprocess.run( [sys.executable, file_path], capture_output=True, text=True, timeout=self.timeout, ) elapsed = round(time.perf_counter() - start, 3) # unittest writes results to stderr, any extra output to stdout raw_output = (proc.stderr.strip() + "\n" + proc.stdout.strip()).strip() # Clean the temp path from output for cleaner display clean_output = raw_output.replace(file_path, "test_suite.py") logger.info( "TestRunner finished in %.3fs — returncode: %d", elapsed, proc.returncode, ) return _parse_unittest_output(clean_output, proc.returncode) except subprocess.TimeoutExpired: elapsed = round(time.perf_counter() - start, 3) msg = f"Test suite timed out after {self.timeout} seconds." logger.warning("TestRunner timeout after %.3fs", elapsed) return TestResult( passed=False, output=msg, failures=[msg], ) except Exception as e: elapsed = round(time.perf_counter() - start, 3) msg = f"TestRunner internal error: {e}" logger.error(msg) return TestResult( passed=False, output=msg, failures=[msg], ) # ------------------------------------------------------------------ # # Helpers # # ------------------------------------------------------------------ # def _parse_unittest_output(output: str, returncode: int) -> TestResult: """ Parse the output of a unittest run (verbosity=2) into a TestResult. The verbose unittest output looks like: test_empty_input (test_suite.TestReverse) ... ok test_normal_input (test_suite.TestReverse) ... ok test_single_char (test_suite.TestReverse) ... FAIL ... FAILED (failures=1) or OK We extract: - Total tests run from the final summary line - Pass/fail counts by scanning each test line - Failure messages from the FAIL/ERROR blocks Args: output: Raw stdout/stderr from the unittest subprocess. returncode: Process return code (0 = all passed). Returns: Typed TestResult with counts and structured failure list. """ if not output: # No output at all — something went wrong before tests ran return TestResult( passed=returncode == 0, output="(no output captured)", failures=[] if returncode == 0 else ["No test output captured."], ) lines = output.splitlines() # ── Count individual test outcomes ────────────────────────────── # passed_tests = 0 failed_tests = 0 error_tests = 0 for line in lines: line_lower = line.lower() if " ... ok" in line_lower: passed_tests += 1 elif " ... fail" in line_lower: failed_tests += 1 elif " ... error" in line_lower: error_tests += 1 total_tests = passed_tests + failed_tests + error_tests # ── Try to extract total from the summary line ─────────────────── # # e.g. "Ran 3 tests in 0.001s" ran_match = re.search(r"Ran (\d+) test", output) if ran_match: total_tests = int(ran_match.group(1)) # ── Extract individual failure/error messages ──────────────────── # failures: list[str] = [] # Failure blocks start with "FAIL: test_name" or "ERROR: test_name" # and end before the next "===" separator or end of output fail_pattern = re.compile( r"((?:FAIL|ERROR): .+?)\n" # Header line r"[-=]+\n" # Separator r"(.*?)" # Failure body r"(?=\n[-=]+|\Z)", # Stop at next separator or end re.DOTALL, ) for match in fail_pattern.finditer(output): header = match.group(1).strip() body = match.group(2).strip() # Keep the most useful part — last few lines of the traceback body_lines = body.splitlines() excerpt = "\n".join(body_lines[-5:]) if len(body_lines) > 5 else body failures.append(f"{header}\n{excerpt}") # Fallback: if no structured failures found but returncode != 0 if not failures and returncode != 0: # Surface the last 10 lines of output as the failure message tail = "\n".join(lines[-10:]) failures.append(tail) # Objective: ALL tests must pass. Even a single failure or error # sends the code back through the debug loop for a rewrite. all_passed = (failed_tests == 0) and (returncode == 0) and (total_tests > 0) if all_passed: logger.info("TestRunner: all %d test(s) passed", total_tests) else: logger.info( "TestRunner: %d passed, %d failed, %d errors", passed_tests, failed_tests, error_tests, ) return TestResult( passed = all_passed, total_tests = total_tests, passed_tests = passed_tests, failed_tests = failed_tests + error_tests, output = output, failures = failures, )