zyc
aab0312cec
All checks were successful
Build and Deploy Log Center / build-and-deploy (push) Successful in 1m35s
154 lines
4.4 KiB
Python
154 lines
4.4 KiB
Python
"""
|
|
Test Runner - 测试执行
|
|
"""
|
|
import subprocess
|
|
import os
|
|
from typing import Optional
|
|
from dataclasses import dataclass
|
|
from loguru import logger
|
|
|
|
|
|
@dataclass
|
|
class TestResult:
|
|
"""测试结果"""
|
|
success: bool
|
|
output: str
|
|
failed_tests: list[str]
|
|
passed_count: int
|
|
failed_count: int
|
|
|
|
|
|
class TestRunner:
|
|
"""负责运行测试"""
|
|
|
|
def __init__(self, project_path: str, project_id: str):
|
|
self.project_path = project_path
|
|
self.project_id = project_id
|
|
|
|
def _detect_test_command(self) -> list[str]:
|
|
"""检测测试命令"""
|
|
# Django 项目
|
|
if os.path.exists(os.path.join(self.project_path, "manage.py")):
|
|
return ["python", "manage.py", "test", "--keepdb", "-v", "2"]
|
|
|
|
# Python pytest
|
|
if os.path.exists(os.path.join(self.project_path, "pytest.ini")) or \
|
|
os.path.exists(os.path.join(self.project_path, "pyproject.toml")):
|
|
return ["pytest", "-v"]
|
|
|
|
# Node.js
|
|
if os.path.exists(os.path.join(self.project_path, "package.json")):
|
|
return ["npm", "test"]
|
|
|
|
# 默认 pytest
|
|
return ["pytest", "-v"]
|
|
|
|
def run_full_suite(self, timeout: int = 300) -> TestResult:
|
|
"""
|
|
运行完整测试套件
|
|
|
|
Args:
|
|
timeout: 超时时间(秒)
|
|
|
|
Returns:
|
|
TestResult
|
|
"""
|
|
cmd = self._detect_test_command()
|
|
logger.info(f"运行测试: {' '.join(cmd)}")
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
cwd=self.project_path,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
|
|
output = result.stdout + result.stderr
|
|
success = result.returncode == 0
|
|
|
|
# 简单解析失败的测试
|
|
failed_tests = []
|
|
for line in output.split("\n"):
|
|
if "FAILED" in line or "ERROR" in line:
|
|
failed_tests.append(line.strip())
|
|
|
|
logger.info(f"测试{'通过' if success else '失败'}")
|
|
|
|
return TestResult(
|
|
success=success,
|
|
output=output,
|
|
failed_tests=failed_tests,
|
|
passed_count=0, # TODO: 解析具体数量
|
|
failed_count=len(failed_tests),
|
|
)
|
|
|
|
except subprocess.TimeoutExpired:
|
|
logger.error(f"测试超时 ({timeout}秒)")
|
|
return TestResult(
|
|
success=False,
|
|
output="测试执行超时",
|
|
failed_tests=[],
|
|
passed_count=0,
|
|
failed_count=0,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"测试执行失败: {e}")
|
|
return TestResult(
|
|
success=False,
|
|
output=str(e),
|
|
failed_tests=[],
|
|
passed_count=0,
|
|
failed_count=0,
|
|
)
|
|
|
|
def run_specific_test(
|
|
self,
|
|
test_path: str,
|
|
timeout: int = 60,
|
|
) -> TestResult:
|
|
"""
|
|
运行特定测试
|
|
|
|
Args:
|
|
test_path: 测试路径 (如 tests.test_user.TestUserLogin)
|
|
timeout: 超时时间
|
|
|
|
Returns:
|
|
TestResult
|
|
"""
|
|
if self.project_id == "rtc_backend":
|
|
cmd = ["python", "manage.py", "test", test_path, "--keepdb", "-v", "2"]
|
|
else:
|
|
cmd = ["pytest", test_path, "-v"]
|
|
|
|
logger.info(f"运行测试: {' '.join(cmd)}")
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
cwd=self.project_path,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
|
|
return TestResult(
|
|
success=result.returncode == 0,
|
|
output=result.stdout + result.stderr,
|
|
failed_tests=[],
|
|
passed_count=0,
|
|
failed_count=0 if result.returncode == 0 else 1,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"测试执行失败: {e}")
|
|
return TestResult(
|
|
success=False,
|
|
output=str(e),
|
|
failed_tests=[],
|
|
passed_count=0,
|
|
failed_count=0,
|
|
)
|