""" test_runner.py - pytest 자동 재실행 verifier subprocess로 pytest를 실행하고 결과를 파싱 """ import os import re import subprocess DEFAULT_TIMEOUT = 60 FULL_SUITE_TIMEOUT = 180 def _parse_pytest_output(output: str) -> str: """pytest stdout에서 통과/실패 요약 줄을 추출.""" # 예: "5 passed, 2 failed in 1.23s" 또는 "3 passed in 0.45s" patterns = [ r"(\d+ passed.*?in \d+\.?\d*s)", r"(\d+ failed.*?in \d+\.?\d*s)", r"(no tests ran.*?in \d+\.?\d*s)", r"(\d+ error.*?in \d+\.?\d*s)", ] for line in reversed(output.splitlines()): line = line.strip() for pattern in patterns: match = re.search(pattern, line, re.IGNORECASE) if match: return match.group(1) return "" def verify( test_dir: str = "", test_files: list[str] | None = None, timeout: int = DEFAULT_TIMEOUT, full_suite: bool = False, ) -> dict: """ 지정된 디렉토리 또는 파일 목록에서 pytest를 실행합니다. Args: test_dir: pytest를 실행할 디렉토리 경로 test_files: pytest에 개별 파일로 전달할 경로 목록 (None이면 미사용) timeout: 최대 실행 시간 (초), 초과 시 SKIP full_suite: True이면 워크스페이스 루트의 tests/ 전체를 실행 Returns: {"status": "PASS"|"FAIL"|"SKIP", "details": [...]} """ # full_suite 모드: workspace_root/tests/ 전체 실행 if full_suite: workspace_root = os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace") tests_dir = os.path.join(workspace_root, "tests") if not os.path.exists(tests_dir): return {"status": "FAIL", "details": [f"Test directory not found: {tests_dir}"]} cmd = ["python3", "-m", "pytest", tests_dir, "-q", "--tb=short", "--continue-on-collection-errors"] effective_timeout = timeout if timeout != DEFAULT_TIMEOUT else FULL_SUITE_TIMEOUT details = [f"Running: {' '.join(cmd)}"] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=effective_timeout, ) stdout = result.stdout.strip() stderr = result.stderr.strip() summary = _parse_pytest_output(stdout) if summary: details.append(f"pytest result: {summary}") exit_code = result.returncode if exit_code == 0: details.append(f"Exit code: {exit_code} (all tests passed)") return {"status": "PASS", "details": details} elif exit_code == 5: details.append(f"Exit code: {exit_code} (no tests collected in {tests_dir})") return {"status": "SKIP", "details": details} else: details.append(f"Exit code: {exit_code} (tests failed or error)") if stdout: fail_lines = [ line for line in stdout.splitlines() if "FAILED" in line or "ERROR" in line or "error" in line.lower() ] for line in fail_lines[:10]: details.append(f" {line.strip()}") if stderr: details.append(f"stderr: {stderr[:300]}") return {"status": "FAIL", "details": details} except subprocess.TimeoutExpired: details.append(f"Timeout after {effective_timeout}s — skipping test run") return {"status": "SKIP", "details": details} except FileNotFoundError: return {"status": "FAIL", "details": ["python3 not found — cannot run pytest"]} except Exception as e: return {"status": "FAIL", "details": [f"Unexpected error running pytest: {e}"]} # test_files가 주어진 경우: 개별 파일 모드 if test_files is not None: existing_files = [f for f in test_files if os.path.isfile(f)] if not existing_files: return { "status": "SKIP", "details": ["추론된 테스트 파일 중 존재하는 파일 없음"], } cmd = [ "python3", "-m", "pytest", *existing_files, "-x", "-q", "--tb=short", ] elif not test_dir: return {"status": "SKIP", "details": ["No test directory specified"]} else: if not os.path.exists(test_dir): return {"status": "FAIL", "details": [f"Test directory not found: {test_dir}"]} if not os.path.isdir(test_dir): return {"status": "FAIL", "details": [f"Not a directory: {test_dir}"]} cmd = [ "python3", "-m", "pytest", test_dir, "-x", "-q", "--tb=short", ] details = [f"Running: {' '.join(cmd)}"] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, ) stdout = result.stdout.strip() stderr = result.stderr.strip() # 요약 줄 추출 summary = _parse_pytest_output(stdout) if summary: details.append(f"pytest result: {summary}") # 종료코드 확인 # pytest 종료코드: 0=all passed, 1=some failed, 2=interrupted, 3=internal error # 4=command line usage error, 5=no tests collected exit_code = result.returncode if exit_code == 0: details.append(f"Exit code: {exit_code} (all tests passed)") return {"status": "PASS", "details": details} elif exit_code == 5: details.append(f"Exit code: {exit_code} (no tests collected in {test_dir})") return {"status": "SKIP", "details": details} else: details.append(f"Exit code: {exit_code} (tests failed or error)") # 실패한 테스트 줄 포함 (최대 10줄) if stdout: fail_lines = [ line for line in stdout.splitlines() if "FAILED" in line or "ERROR" in line or "error" in line.lower() ] for line in fail_lines[:10]: details.append(f" {line.strip()}") if stderr: details.append(f"stderr: {stderr[:300]}") return {"status": "FAIL", "details": details} except subprocess.TimeoutExpired: details.append(f"Timeout after {timeout}s — skipping test run") return {"status": "SKIP", "details": details} except FileNotFoundError: return { "status": "FAIL", "details": ["python3 not found — cannot run pytest"], } except Exception as e: return { "status": "FAIL", "details": [f"Unexpected error running pytest: {e}"], } if __name__ == "__main__": import json import sys test_dir = sys.argv[1] if len(sys.argv) > 1 else "" t = int(sys.argv[2]) if len(sys.argv) > 2 else DEFAULT_TIMEOUT result = verify(test_dir=test_dir, timeout=t) print(json.dumps(result, ensure_ascii=False, indent=2))