mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
refactor: remove dead Test/CI route stack
Trashed roboco/api/routes/test.py, roboco/api/schemas/test.py, and roboco/services/test_runner.py — 744 LOC across 3 files. The routes (/test/run, /test/lint, /test/format, /test/typecheck, /test/build) were the API surface for the deleted roboco/mcp/test/test_server.py MCP. With that MCP server gone (Phase 4 T9) no caller remains: panel never used /test/*, and no Python service imports TestRunnerService outside the deleted MCP.
This commit is contained in:
Renn F
@@ -32,7 +32,6 @@ from roboco.api.routes.provider import router as provider_router
|
||||
from roboco.api.routes.sessions import router as sessions_router
|
||||
from roboco.api.routes.stream import router as stream_router
|
||||
from roboco.api.routes.tasks import router as tasks_router
|
||||
from roboco.api.routes.test import router as test_router
|
||||
from roboco.api.routes.v2 import do as do_module
|
||||
from roboco.api.routes.v2 import flow_auditor as flow_auditor_module
|
||||
from roboco.api.routes.v2 import flow_board as flow_board_module
|
||||
@@ -267,13 +266,6 @@ def create_app() -> FastAPI:
|
||||
tags=["Git Operations"],
|
||||
)
|
||||
|
||||
# Test/CI Operations
|
||||
app.include_router(
|
||||
test_router,
|
||||
prefix=f"{api_prefix}/test",
|
||||
tags=["Test Operations"],
|
||||
)
|
||||
|
||||
# Project Management
|
||||
app.include_router(
|
||||
project_router,
|
||||
|
||||
@@ -21,7 +21,6 @@ from roboco.api.routes import (
|
||||
sessions,
|
||||
stream,
|
||||
tasks,
|
||||
test,
|
||||
work_session,
|
||||
)
|
||||
|
||||
@@ -42,6 +41,5 @@ __all__ = [
|
||||
"sessions",
|
||||
"stream",
|
||||
"tasks",
|
||||
"test",
|
||||
"work_session",
|
||||
]
|
||||
|
||||
@@ -1,136 +0,0 @@
|
||||
"""
|
||||
Test API Routes
|
||||
|
||||
CI/CD operations for agents working on code tasks. Thin HTTP plumbing —
|
||||
workspace resolution, subprocess dispatch, and output parsing live in
|
||||
`TestRunnerService`.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query, status
|
||||
|
||||
from roboco.api.deps import CurrentAgentContext, DbSession
|
||||
from roboco.api.schemas.test import (
|
||||
BuildRequest,
|
||||
BuildResponse,
|
||||
FormatRequest,
|
||||
FormatResponse,
|
||||
LintRequest,
|
||||
LintResponse,
|
||||
TestRunRequest,
|
||||
TestRunResponse,
|
||||
TestStatusResponse,
|
||||
TypecheckRequest,
|
||||
TypecheckResponse,
|
||||
)
|
||||
from roboco.services.base import (
|
||||
NotFoundError,
|
||||
ServiceError,
|
||||
ServiceUnavailableError,
|
||||
UnauthorizedError,
|
||||
ValidationError,
|
||||
)
|
||||
from roboco.services.test_runner import get_test_runner_service
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
def _translate_error(e: ServiceError) -> HTTPException:
|
||||
"""Service errors → HTTP status."""
|
||||
if isinstance(e, NotFoundError):
|
||||
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=e.message)
|
||||
if isinstance(e, UnauthorizedError):
|
||||
return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=e.message)
|
||||
if isinstance(e, ValidationError):
|
||||
return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=e.message)
|
||||
if isinstance(e, ServiceUnavailableError):
|
||||
return HTTPException(
|
||||
status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail=e.message
|
||||
)
|
||||
return HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=e.message
|
||||
)
|
||||
|
||||
|
||||
@router.get("/status", response_model=TestStatusResponse)
|
||||
async def get_test_status(
|
||||
db: DbSession,
|
||||
agent: CurrentAgentContext,
|
||||
project_slug: str = Query(...),
|
||||
_task_id: str | None = Query(default=None),
|
||||
) -> TestStatusResponse:
|
||||
"""Get test status for a project."""
|
||||
service = get_test_runner_service(db)
|
||||
try:
|
||||
return await service.get_status(project_slug, agent.agent_id)
|
||||
except ServiceError as e:
|
||||
raise _translate_error(e) from e
|
||||
|
||||
|
||||
@router.post("/run", response_model=TestRunResponse)
|
||||
async def run_tests(
|
||||
data: TestRunRequest,
|
||||
db: DbSession,
|
||||
agent: CurrentAgentContext,
|
||||
) -> TestRunResponse:
|
||||
"""Run tests for a project."""
|
||||
service = get_test_runner_service(db)
|
||||
try:
|
||||
return await service.run_tests(agent.agent_id, data)
|
||||
except ServiceError as e:
|
||||
raise _translate_error(e) from e
|
||||
|
||||
|
||||
@router.post("/lint", response_model=LintResponse)
|
||||
async def run_lint(
|
||||
data: LintRequest,
|
||||
db: DbSession,
|
||||
agent: CurrentAgentContext,
|
||||
) -> LintResponse:
|
||||
"""Run linter for a project."""
|
||||
service = get_test_runner_service(db)
|
||||
try:
|
||||
return await service.run_lint(agent.agent_id, data)
|
||||
except ServiceError as e:
|
||||
raise _translate_error(e) from e
|
||||
|
||||
|
||||
@router.post("/format", response_model=FormatResponse)
|
||||
async def run_format(
|
||||
data: FormatRequest,
|
||||
db: DbSession,
|
||||
agent: CurrentAgentContext,
|
||||
) -> FormatResponse:
|
||||
"""Run formatter for a project."""
|
||||
service = get_test_runner_service(db)
|
||||
try:
|
||||
return await service.run_format(agent.agent_id, data)
|
||||
except ServiceError as e:
|
||||
raise _translate_error(e) from e
|
||||
|
||||
|
||||
@router.post("/typecheck", response_model=TypecheckResponse)
|
||||
async def run_typecheck(
|
||||
data: TypecheckRequest,
|
||||
db: DbSession,
|
||||
agent: CurrentAgentContext,
|
||||
) -> TypecheckResponse:
|
||||
"""Run type checker for a project."""
|
||||
service = get_test_runner_service(db)
|
||||
try:
|
||||
return await service.run_typecheck(agent.agent_id, data)
|
||||
except ServiceError as e:
|
||||
raise _translate_error(e) from e
|
||||
|
||||
|
||||
@router.post("/build", response_model=BuildResponse)
|
||||
async def run_build(
|
||||
data: BuildRequest,
|
||||
db: DbSession,
|
||||
agent: CurrentAgentContext,
|
||||
) -> BuildResponse:
|
||||
"""Run build command for a project."""
|
||||
service = get_test_runner_service(db)
|
||||
try:
|
||||
return await service.run_build(agent.agent_id, data)
|
||||
except ServiceError as e:
|
||||
raise _translate_error(e) from e
|
||||
@@ -1,189 +0,0 @@
|
||||
"""
|
||||
Test API Schemas
|
||||
|
||||
Request/response models for test/CI operation endpoints.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
# =============================================================================
|
||||
# STATUS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class TestStatusResponse(BaseModel):
|
||||
"""Test status response."""
|
||||
|
||||
project_slug: str
|
||||
passed: bool
|
||||
summary: str
|
||||
last_run: datetime | None = None
|
||||
passed_count: int = 0
|
||||
failed_count: int = 0
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# TEST RUN
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class TestRunRequest(BaseModel):
|
||||
"""Request to run tests."""
|
||||
|
||||
project_slug: str
|
||||
task_id: str
|
||||
agent_id: str
|
||||
test_path: str | None = None
|
||||
verbose: bool = False
|
||||
|
||||
|
||||
class TestRunResponse(BaseModel):
|
||||
"""Response from test run.
|
||||
|
||||
`skipped=True` + `skip_reason` indicates the project opted out of
|
||||
this check (e.g., no test_command configured). That's distinct
|
||||
from a failure — QA should treat it as "nothing to verify here,
|
||||
move on" rather than a gate violation.
|
||||
"""
|
||||
|
||||
project_slug: str
|
||||
passed: bool
|
||||
passed_count: int = 0
|
||||
failed_count: int = 0
|
||||
skipped_count: int = 0
|
||||
duration_seconds: float = 0
|
||||
output: str = ""
|
||||
failures: list[str] = []
|
||||
skipped: bool = False
|
||||
skip_reason: str | None = None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# LINT
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class LintRequest(BaseModel):
|
||||
"""Request to run linter."""
|
||||
|
||||
project_slug: str
|
||||
task_id: str
|
||||
agent_id: str
|
||||
fix: bool = False
|
||||
path: str | None = None
|
||||
|
||||
|
||||
class LintIssue(BaseModel):
|
||||
"""A lint issue found."""
|
||||
|
||||
file: str
|
||||
line: int
|
||||
column: int
|
||||
code: str
|
||||
message: str
|
||||
|
||||
|
||||
class LintResponse(BaseModel):
|
||||
"""Response from lint run.
|
||||
|
||||
See `TestRunResponse` for `skipped` / `skip_reason` semantics.
|
||||
"""
|
||||
|
||||
project_slug: str
|
||||
passed: bool
|
||||
issues: list[LintIssue] = []
|
||||
fixed_count: int = 0
|
||||
skipped: bool = False
|
||||
skip_reason: str | None = None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FORMAT
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class FormatRequest(BaseModel):
|
||||
"""Request to run formatter."""
|
||||
|
||||
project_slug: str
|
||||
task_id: str
|
||||
agent_id: str
|
||||
check_only: bool = False
|
||||
path: str | None = None
|
||||
|
||||
|
||||
class FormatResponse(BaseModel):
|
||||
"""Response from format run.
|
||||
|
||||
See `TestRunResponse` for `skipped` / `skip_reason` semantics.
|
||||
"""
|
||||
|
||||
project_slug: str
|
||||
files_modified: int = 0
|
||||
files_unchanged: int = 0
|
||||
skipped: bool = False
|
||||
skip_reason: str | None = None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# TYPECHECK
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class TypecheckRequest(BaseModel):
|
||||
"""Request to run type checker."""
|
||||
|
||||
project_slug: str
|
||||
task_id: str
|
||||
agent_id: str
|
||||
path: str | None = None
|
||||
|
||||
|
||||
class TypecheckError(BaseModel):
|
||||
"""A type check error found."""
|
||||
|
||||
file: str
|
||||
line: int
|
||||
message: str
|
||||
|
||||
|
||||
class TypecheckResponse(BaseModel):
|
||||
"""Response from type check run.
|
||||
|
||||
See `TestRunResponse` for `skipped` / `skip_reason` semantics.
|
||||
"""
|
||||
|
||||
project_slug: str
|
||||
passed: bool
|
||||
errors: list[TypecheckError] = []
|
||||
skipped: bool = False
|
||||
skip_reason: str | None = None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# BUILD
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class BuildRequest(BaseModel):
|
||||
"""Request to run build."""
|
||||
|
||||
project_slug: str
|
||||
task_id: str
|
||||
agent_id: str
|
||||
|
||||
|
||||
class BuildResponse(BaseModel):
|
||||
"""Response from build run.
|
||||
|
||||
See `TestRunResponse` for `skipped` / `skip_reason` semantics.
|
||||
"""
|
||||
|
||||
project_slug: str
|
||||
success: bool
|
||||
duration_seconds: float = 0
|
||||
output: str = ""
|
||||
skipped: bool = False
|
||||
skip_reason: str | None = None
|
||||
@@ -1,419 +0,0 @@
|
||||
"""
|
||||
Test Runner Service
|
||||
|
||||
Orchestrates CI/CD-style commands (tests, lint, format, typecheck, build)
|
||||
in agent workspaces. The API routes are thin adapters over this service —
|
||||
all workspace resolution, subprocess dispatch, and output parsing happens
|
||||
here so routes only handle HTTP translation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
import shlex
|
||||
import subprocess
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, ClassVar
|
||||
|
||||
from roboco.api.schemas.test import (
|
||||
BuildRequest,
|
||||
BuildResponse,
|
||||
FormatRequest,
|
||||
FormatResponse,
|
||||
LintIssue,
|
||||
LintRequest,
|
||||
LintResponse,
|
||||
TestRunRequest,
|
||||
TestRunResponse,
|
||||
TestStatusResponse,
|
||||
TypecheckError,
|
||||
TypecheckRequest,
|
||||
TypecheckResponse,
|
||||
)
|
||||
from roboco.config import settings
|
||||
from roboco.services.base import (
|
||||
BaseService,
|
||||
NotFoundError,
|
||||
ServiceError,
|
||||
ServiceUnavailableError,
|
||||
ValidationError,
|
||||
)
|
||||
from roboco.services.project import get_project_service
|
||||
from roboco.services.workspace import WorkspaceError, get_workspace_service
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from uuid import UUID
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
# Command timeout in seconds — long enough for full test runs.
|
||||
_CMD_TIMEOUT = 300
|
||||
|
||||
# Minimum parts for lint output parsing (file:line:col:message).
|
||||
_LINT_PARTS_MIN = 4
|
||||
|
||||
# Minimum parts for type error parsing (file:line:message).
|
||||
_TYPE_ERROR_PARTS_MIN = 3
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _ProjectContext:
|
||||
"""Resolved (project, workspace) pair for a command run."""
|
||||
|
||||
project: Any
|
||||
workspace: Path
|
||||
|
||||
|
||||
class TestRunnerService(BaseService):
|
||||
"""Runs project CI commands in an agent's workspace."""
|
||||
|
||||
service_name: ClassVar[str] = "test_runner"
|
||||
|
||||
# =========================================================================
|
||||
# WORKSPACE RESOLUTION
|
||||
# =========================================================================
|
||||
|
||||
async def _resolve_legacy_workspace(self, project: Any, project_slug: str) -> Path:
|
||||
"""Legacy single-path project config (no agent_id given)."""
|
||||
workspace_path = getattr(project, "workspace_path", None)
|
||||
if not workspace_path:
|
||||
raise ValidationError(
|
||||
f"Project '{project_slug}' has no workspace configured and "
|
||||
"no agent_id provided for dynamic workspace resolution"
|
||||
)
|
||||
workspace = Path(workspace_path)
|
||||
if not workspace.exists():
|
||||
raise ValidationError(f"Workspace path does not exist: {workspace}")
|
||||
return workspace
|
||||
|
||||
async def _resolve_agent_workspace(
|
||||
self, project: Any, project_slug: str, agent_id: UUID
|
||||
) -> Path:
|
||||
"""Resolve workspace via WorkspaceService (ensure/resolve per setting)."""
|
||||
workspace_service = get_workspace_service(self.session)
|
||||
try:
|
||||
if settings.workspace_auto_clone:
|
||||
return await workspace_service.ensure_workspace(
|
||||
project_slug=project_slug,
|
||||
agent_id=agent_id,
|
||||
git_url=project.git_url,
|
||||
default_branch=project.default_branch or "main",
|
||||
)
|
||||
workspace = await workspace_service.resolve_workspace(
|
||||
project_slug=project_slug,
|
||||
agent_id=agent_id,
|
||||
)
|
||||
if not workspace.exists():
|
||||
raise ValidationError(
|
||||
f"Workspace does not exist: {workspace}. "
|
||||
"Clone the repository first or enable auto_clone."
|
||||
)
|
||||
return workspace
|
||||
except WorkspaceError as e:
|
||||
raise ValidationError(str(e)) from e
|
||||
|
||||
async def _load_project_and_workspace(
|
||||
self,
|
||||
project_slug: str,
|
||||
agent_id: UUID | None,
|
||||
) -> _ProjectContext:
|
||||
"""Fetch project + resolve its workspace. Raises typed errors."""
|
||||
service = get_project_service(self.session)
|
||||
project = await service.get_by_slug(project_slug)
|
||||
if not project:
|
||||
raise NotFoundError(resource_type="Project", resource_id=project_slug)
|
||||
|
||||
if agent_id is None:
|
||||
workspace = await self._resolve_legacy_workspace(project, project_slug)
|
||||
else:
|
||||
workspace = await self._resolve_agent_workspace(
|
||||
project, project_slug, agent_id
|
||||
)
|
||||
return _ProjectContext(project=project, workspace=workspace)
|
||||
|
||||
# =========================================================================
|
||||
# COMMAND RUNNER
|
||||
# =========================================================================
|
||||
|
||||
async def _run_command(
|
||||
self,
|
||||
workspace: Path,
|
||||
command: str,
|
||||
timeout: int = _CMD_TIMEOUT,
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
"""Run a shell command in the workspace (non-blocking).
|
||||
|
||||
Commands are tokenized with shlex — no shell features (pipes,
|
||||
redirects, env expansion) supported. Project-configured commands
|
||||
are expected to be simple exe + args.
|
||||
"""
|
||||
argv = shlex.split(command)
|
||||
|
||||
def _run() -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
argv,
|
||||
check=False,
|
||||
cwd=workspace,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
try:
|
||||
return await asyncio.to_thread(_run)
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise ServiceUnavailableError(
|
||||
service_name="test_runner",
|
||||
reason=f"Command timed out after {timeout}s: {command}",
|
||||
) from e
|
||||
except FileNotFoundError as e:
|
||||
binary = argv[0] if argv else command
|
||||
raise ValidationError(
|
||||
f"Command binary not found: '{binary}'. "
|
||||
"Update the project's configured command (e.g., replace 'make test' "
|
||||
"with 'uv run pytest') or ensure the binary is installed in the "
|
||||
"runtime environment."
|
||||
) from e
|
||||
|
||||
def _project_cmd(self, project: Any, attr: str) -> str | None:
|
||||
"""Fetch a configured command for the project, or None if unset.
|
||||
|
||||
Prior behavior raised ValidationError on a missing command, but
|
||||
QA agents ended up hitting 400s on every probe when the project
|
||||
simply hadn't opted into that CI tool (e.g., lint_command=null
|
||||
because the task is a README edit). Now callers get `None` and
|
||||
return a `skipped=True` success response — no gate violation,
|
||||
clear signal to the agent that there's nothing to run.
|
||||
"""
|
||||
cmd = getattr(project, attr, None)
|
||||
return str(cmd) if cmd else None
|
||||
|
||||
@staticmethod
|
||||
def _skip_reason(project_slug: str, label: str) -> str:
|
||||
return (
|
||||
f"Project '{project_slug}' has no {label} configured — check "
|
||||
"skipped. Review the change manually if it's relevant."
|
||||
)
|
||||
|
||||
# =========================================================================
|
||||
# OUTPUT PARSERS
|
||||
# =========================================================================
|
||||
|
||||
@staticmethod
|
||||
def _parse_pytest_counts(output: str) -> tuple[int, int, int]:
|
||||
"""Parse (passed, failed, skipped) from pytest-style output."""
|
||||
if "passed" not in output:
|
||||
return 0, 0, 0
|
||||
|
||||
def _first_int(pattern: str) -> int:
|
||||
match = re.search(pattern, output)
|
||||
return int(match.group(1)) if match else 0
|
||||
|
||||
return (
|
||||
_first_int(r"(\d+) passed"),
|
||||
_first_int(r"(\d+) failed"),
|
||||
_first_int(r"(\d+) skipped"),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _parse_lint_line(line: str) -> LintIssue | None:
|
||||
"""Parse a single ruff-style lint output line."""
|
||||
if "::" in line or not line.strip():
|
||||
return None
|
||||
parts = line.split(":", 3)
|
||||
if len(parts) < _LINT_PARTS_MIN:
|
||||
return None
|
||||
try:
|
||||
return LintIssue(
|
||||
file=parts[0],
|
||||
line=int(parts[1]),
|
||||
column=int(parts[2]),
|
||||
code=parts[3].split()[0] if parts[3].strip() else "E",
|
||||
message=parts[3].strip(),
|
||||
)
|
||||
except (ValueError, IndexError):
|
||||
return None
|
||||
|
||||
def _parse_lint_output(self, output: str) -> list[LintIssue]:
|
||||
return [
|
||||
issue
|
||||
for line in output.split("\n")
|
||||
if (issue := self._parse_lint_line(line)) is not None
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _parse_typecheck_line(line: str) -> TypecheckError | None:
|
||||
"""Parse a single mypy-style type error line."""
|
||||
if ": error:" not in line:
|
||||
return None
|
||||
parts = line.split(":", 2)
|
||||
if len(parts) < _TYPE_ERROR_PARTS_MIN:
|
||||
return None
|
||||
try:
|
||||
return TypecheckError(
|
||||
file=parts[0],
|
||||
line=int(parts[1]),
|
||||
message=parts[2].replace(" error:", "").strip(),
|
||||
)
|
||||
except (ValueError, IndexError):
|
||||
return None
|
||||
|
||||
def _parse_typecheck_output(self, output: str) -> list[TypecheckError]:
|
||||
return [
|
||||
err
|
||||
for line in output.split("\n")
|
||||
if (err := self._parse_typecheck_line(line)) is not None
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _build_format_cmd(base_cmd: str, data: FormatRequest) -> str:
|
||||
cmd = base_cmd
|
||||
if data.check_only:
|
||||
cmd = f"{cmd} --check"
|
||||
if data.path:
|
||||
cmd = f"{cmd} {data.path}"
|
||||
return cmd
|
||||
|
||||
# =========================================================================
|
||||
# PUBLIC ORCHESTRATION
|
||||
# =========================================================================
|
||||
|
||||
async def get_status(self, project_slug: str, agent_id: UUID) -> TestStatusResponse:
|
||||
"""Placeholder status endpoint — validates workspace resolution."""
|
||||
await self._load_project_and_workspace(project_slug, agent_id)
|
||||
return TestStatusResponse(
|
||||
project_slug=project_slug,
|
||||
passed=True,
|
||||
summary=(
|
||||
"No test results stored yet. Run roboco_test_run() to execute tests."
|
||||
),
|
||||
last_run=None,
|
||||
)
|
||||
|
||||
async def run_tests(self, agent_id: UUID, data: TestRunRequest) -> TestRunResponse:
|
||||
ctx = await self._load_project_and_workspace(data.project_slug, agent_id)
|
||||
base = self._project_cmd(ctx.project, "test_command")
|
||||
if not base:
|
||||
return TestRunResponse(
|
||||
project_slug=data.project_slug,
|
||||
passed=True,
|
||||
skipped=True,
|
||||
skip_reason=self._skip_reason(data.project_slug, "test_command"),
|
||||
)
|
||||
cmd = base
|
||||
if data.test_path:
|
||||
cmd = f"{cmd} {data.test_path}"
|
||||
if data.verbose:
|
||||
cmd = f"{cmd} -v"
|
||||
|
||||
result = await self._run_command(ctx.workspace, cmd)
|
||||
output = result.stdout + result.stderr
|
||||
passed, failed, skipped = self._parse_pytest_counts(output)
|
||||
|
||||
return TestRunResponse(
|
||||
project_slug=data.project_slug,
|
||||
passed=result.returncode == 0,
|
||||
passed_count=passed,
|
||||
failed_count=failed,
|
||||
skipped_count=skipped,
|
||||
output=output[:10000],
|
||||
failures=[],
|
||||
)
|
||||
|
||||
async def run_lint(self, agent_id: UUID, data: LintRequest) -> LintResponse:
|
||||
ctx = await self._load_project_and_workspace(data.project_slug, agent_id)
|
||||
base = self._project_cmd(ctx.project, "lint_command")
|
||||
if not base:
|
||||
return LintResponse(
|
||||
project_slug=data.project_slug,
|
||||
passed=True,
|
||||
skipped=True,
|
||||
skip_reason=self._skip_reason(data.project_slug, "lint_command"),
|
||||
)
|
||||
cmd = base
|
||||
if data.fix:
|
||||
cmd = f"{cmd} --fix"
|
||||
if data.path:
|
||||
cmd = f"{cmd} {data.path}"
|
||||
|
||||
result = await self._run_command(ctx.workspace, cmd)
|
||||
output = result.stdout + result.stderr
|
||||
return LintResponse(
|
||||
project_slug=data.project_slug,
|
||||
passed=result.returncode == 0,
|
||||
issues=self._parse_lint_output(output),
|
||||
fixed_count=0,
|
||||
)
|
||||
|
||||
async def run_format(self, agent_id: UUID, data: FormatRequest) -> FormatResponse:
|
||||
ctx = await self._load_project_and_workspace(data.project_slug, agent_id)
|
||||
base = self._project_cmd(ctx.project, "format_command")
|
||||
if not base:
|
||||
return FormatResponse(
|
||||
project_slug=data.project_slug,
|
||||
skipped=True,
|
||||
skip_reason=self._skip_reason(data.project_slug, "format_command"),
|
||||
)
|
||||
result = await self._run_command(
|
||||
ctx.workspace, self._build_format_cmd(base, data)
|
||||
)
|
||||
output = result.stdout + result.stderr
|
||||
files_modified = output.count("reformatted") if not data.check_only else 0
|
||||
files_unchanged = output.count("unchanged") or output.count("already formatted")
|
||||
return FormatResponse(
|
||||
project_slug=data.project_slug,
|
||||
files_modified=files_modified,
|
||||
files_unchanged=files_unchanged,
|
||||
)
|
||||
|
||||
async def run_typecheck(
|
||||
self, agent_id: UUID, data: TypecheckRequest
|
||||
) -> TypecheckResponse:
|
||||
ctx = await self._load_project_and_workspace(data.project_slug, agent_id)
|
||||
base = self._project_cmd(ctx.project, "typecheck_command")
|
||||
if not base:
|
||||
return TypecheckResponse(
|
||||
project_slug=data.project_slug,
|
||||
passed=True,
|
||||
skipped=True,
|
||||
skip_reason=self._skip_reason(data.project_slug, "typecheck_command"),
|
||||
)
|
||||
cmd = f"{base} {data.path}" if data.path else base
|
||||
result = await self._run_command(ctx.workspace, cmd)
|
||||
output = result.stdout + result.stderr
|
||||
return TypecheckResponse(
|
||||
project_slug=data.project_slug,
|
||||
passed=result.returncode == 0,
|
||||
errors=self._parse_typecheck_output(output),
|
||||
)
|
||||
|
||||
async def run_build(self, agent_id: UUID, data: BuildRequest) -> BuildResponse:
|
||||
ctx = await self._load_project_and_workspace(data.project_slug, agent_id)
|
||||
base = self._project_cmd(ctx.project, "build_command")
|
||||
if not base:
|
||||
return BuildResponse(
|
||||
project_slug=data.project_slug,
|
||||
success=True,
|
||||
skipped=True,
|
||||
skip_reason=self._skip_reason(data.project_slug, "build_command"),
|
||||
)
|
||||
start = time.time()
|
||||
result = await self._run_command(ctx.workspace, base)
|
||||
duration = time.time() - start
|
||||
return BuildResponse(
|
||||
project_slug=data.project_slug,
|
||||
success=result.returncode == 0,
|
||||
duration_seconds=round(duration, 2),
|
||||
output=(result.stdout + result.stderr)[:10000],
|
||||
)
|
||||
|
||||
|
||||
def get_test_runner_service(session: AsyncSession) -> TestRunnerService:
|
||||
"""Factory for TestRunnerService."""
|
||||
return TestRunnerService(session)
|
||||
|
||||
|
||||
__all__ = ["ServiceError", "TestRunnerService", "get_test_runner_service"]
|
||||
Reference in New Issue
Block a user