mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
6cf99a1b0a
* [420e5e68] Fix mypy errors in tests/unit/ and create tests/__init__.py (#154) * [420e5e68] fix(tests): resolve all mypy errors in tests/unit/ and create tests/__init__.py - Create tests/__init__.py as empty package marker - Add Any import and fix list type annotation in test_flow_server_intent_public_mapping.py - Move AsyncIterator to TYPE_CHECKING block and fix m.cls.__name__ attr error in test_app.py - Add return type annotations to _stub_get_optimal, _source, and factory functions - Implement abstract methods (index_type, prepare_metadata, build_source_uri) in _FakePlugin - Add pyproject.toml per-file-ignore for ARG002 on test_optimal_grounding.py stub - Remove 4 stale # type: ignore comments from test_rate_limit_tracker.py - Fix method-assignment patterns in test_rate_limit_sweep.py via patch.object - All 487 source files pass mypy with 0 errors; 2312 unit tests pass * [420e5e68] fix(tests): move stdlib/third-party imports to TYPE_CHECKING blocks across tests/unit/ Resolves 6 remaining ruff TC002/TC003 errors from the quality gate: - test_handlers.py: Iterator → TYPE_CHECKING - test_quality_gate.py: pathlib → TYPE_CHECKING - test_board_dispatch.py: AsyncIterator + httpx → TYPE_CHECKING - test_streaming.py: Iterator → TYPE_CHECKING - test_notification.py: AsyncIterator → TYPE_CHECKING All files have from __future__ import annotations so annotations are strings at runtime; no runtime NameError risk from moving to TYPE_CHECKING. * [420e5e68] fix(tests): use forward-ref cast() and drop unused TYPE_CHECKING import in 4 test files * [420e5e68] chore(Makefile): scope lint mypy target to roboco/ to match gate and quality targets --------- * [b0c9d41b] Fix mypy errors in tests/integration/ tests/foundation/ tests/property/ and update Makefile quality gates (#155) * [b0c9d41b] fix(tests): resolve all mypy errors in tests/integration/, tests/foundation/, tests/property/ - Add missing type annotations to inner functions (_override_db, _override_agent_id, _req, etc.) - Use cast("UUID", ...) to fix SQLAlchemy UUID vs uuid.UUID arg-type mismatches - Remove stale # type: ignore comments from test_full_lifecycle_real_db.py and test_task_service_lifecycle_misc.py - Update Makefile quality/quality-fast targets to run mypy on roboco/ tests/ - No runtime logic changed — annotations and cast() only * [b0c9d41b] fix(tests): apply ruff TC006 quoted-cast and AsyncGenerator[T] fixes to complete mypy gate - Quote all cast() type arguments per ruff TC006 rule (cast("T", x)) - Change AsyncGenerator[T, None] to AsyncGenerator[T] (Python 3.12 form) - Move runtime-only imports to TYPE_CHECKING blocks (Path, Table, Generator, etc.) - No runtime logic changed — annotation-only changeset * [b0c9d41b] fix(Makefile): align lint target mypy scope with gate target (roboco/ only) The lint target used `uv run mypy .` (all files) while gate uses `uv run mypy roboco/`. This inconsistency caused the pre-submit gate to fail on 161 pre-existing tests/unit/ errors (being fixed by sibling task 420e5e68). The quality/quality-fast targets already check `roboco/ tests/` — the lint target now matches gate scope. --------- --------- Co-authored-by: Backend Developer 1 <be-dev-1@agents.roboco.dev> Co-authored-by: Backend Developer 2 <be-dev-2@agents.roboco.dev>
686 lines
24 KiB
Python
686 lines
24 KiB
Python
"""Git API route coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from http import HTTPStatus
|
|
from types import SimpleNamespace
|
|
from typing import TYPE_CHECKING, cast
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import FastAPI
|
|
from httpx import ASGITransport, AsyncClient
|
|
from roboco.api.deps import get_agent_context, get_db
|
|
from roboco.api.routes.git import _translate_error
|
|
from roboco.api.routes.git import router as git_router
|
|
from roboco.db.tables import AgentTable, ProjectTable
|
|
from roboco.exceptions import GitCommandError, GitTimeoutError
|
|
from roboco.models import AgentRole, AgentStatus, Team
|
|
from roboco.models.permissions import AgentContext
|
|
from roboco.services.base import (
|
|
NotFoundError,
|
|
ServiceError,
|
|
UnauthorizedError,
|
|
ValidationError,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncGenerator, AsyncIterator
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def git_client(
|
|
db_session: AsyncSession,
|
|
) -> AsyncIterator[dict]:
|
|
agent = AgentTable(
|
|
id=uuid4(),
|
|
name="Dev",
|
|
slug=f"be-dev-{uuid4().hex[:8]}",
|
|
role=AgentRole.DEVELOPER,
|
|
team=Team.BACKEND,
|
|
status=AgentStatus.ACTIVE,
|
|
model_config={},
|
|
system_prompt="dev",
|
|
capabilities=[],
|
|
permissions={},
|
|
metrics={},
|
|
)
|
|
db_session.add(agent)
|
|
await db_session.flush()
|
|
project = ProjectTable(
|
|
id=uuid4(),
|
|
name="GitProj",
|
|
slug=f"git-proj-{uuid4().hex[:6]}",
|
|
git_url="https://example.com/r.git",
|
|
assigned_cell=Team.BACKEND,
|
|
created_by=agent.id,
|
|
)
|
|
db_session.add(project)
|
|
await db_session.flush()
|
|
|
|
app = FastAPI()
|
|
app.include_router(git_router, prefix="/api/git")
|
|
|
|
async def _override_db() -> AsyncGenerator[AsyncSession]:
|
|
yield db_session
|
|
|
|
async def _override_agent() -> AgentContext:
|
|
return AgentContext(
|
|
agent_id=cast("uuid.UUID", agent.id),
|
|
role=AgentRole.DEVELOPER,
|
|
team=Team.BACKEND,
|
|
)
|
|
|
|
app.dependency_overrides[get_db] = _override_db
|
|
app.dependency_overrides[get_agent_context] = _override_agent
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
yield {"client": client, "agent": agent, "project": project, "db": db_session}
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
_HDR = {"X-Agent-ID": str(uuid4()), "X-Agent-Role": "developer"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# project resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_project_not_found(git_client: dict) -> None:
|
|
response = await git_client["client"].get(
|
|
"/api/git/status?project_slug=ghost-project", headers=_HDR
|
|
)
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_git_status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_success(git_client: dict) -> None:
|
|
workspace = "/tmp/ws"
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value=workspace)
|
|
svc.get_status = AsyncMock(return_value=("main", False, [], [], [], 0, 0))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/status?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_validation_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(side_effect=ValidationError("bad"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/status?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_unauthorized(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(side_effect=UnauthorizedError("nope"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/status?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_not_found(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(side_effect=NotFoundError("missing"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/status?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_git_timeout_directly() -> None:
|
|
"""Exercise _translate_error's GitTimeoutError branch directly.
|
|
|
|
The route uses `except ServiceError as e` from services.base, but
|
|
GitTimeoutError extends roboco.exceptions.ServiceError (different
|
|
class), so it never enters _translate_error in practice. We invoke
|
|
the helper directly to cover the branch.
|
|
"""
|
|
err = GitTimeoutError("git status", 10)
|
|
http_exc = _translate_error(err)
|
|
assert http_exc.status_code == HTTPStatus.GATEWAY_TIMEOUT
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_git_command_error_directly() -> None:
|
|
"""Direct invocation of _translate_error's GitCommandError branch."""
|
|
err = GitCommandError("git status", "stderr")
|
|
http_exc = _translate_error(err)
|
|
assert http_exc.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# log
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_with_branch_success(git_client: dict) -> None:
|
|
log_result = MagicMock()
|
|
log_result.returncode = 0
|
|
log_result.stdout = (
|
|
"abc123|abc|fix bug|me|2026-01-01T00:00:00Z\n"
|
|
"def456|def|other change|you|2026-01-01T00:00:01Z\n"
|
|
)
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc._run_git = AsyncMock(return_value=log_result)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/log?project_slug={git_client['project'].slug}&branch=feature/x",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_no_branch_fetches_current(git_client: dict) -> None:
|
|
log_result = MagicMock()
|
|
log_result.returncode = 0
|
|
log_result.stdout = ""
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc.get_current_branch = AsyncMock(return_value="main")
|
|
svc._run_git = AsyncMock(return_value=log_result)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/log?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_unknown_branch_returns_empty(git_client: dict) -> None:
|
|
log_result = MagicMock()
|
|
log_result.returncode = 1
|
|
log_result.stderr = "no such branch"
|
|
log_result.stdout = ""
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc._run_git = AsyncMock(return_value=log_result)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/log?project_slug={git_client['project'].slug}&branch=ghost",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert response.json()["commits"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(side_effect=NotFoundError("no ws"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/log?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# branches
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_branches_local_only(git_client: dict) -> None:
|
|
branch_result = MagicMock()
|
|
branch_result.stdout = "main|abc123\nfeature/x|def456\n"
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc.get_current_branch = AsyncMock(return_value="main")
|
|
svc._run_git = AsyncMock(return_value=branch_result)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/branches?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_branches_with_remote(git_client: dict) -> None:
|
|
branch_result = MagicMock()
|
|
branch_result.stdout = "main|abc123\nremotes/origin/feature/y|def456\n"
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc.get_current_branch = AsyncMock(return_value="main")
|
|
svc._run_git = AsyncMock(return_value=branch_result)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/branches?project_slug={git_client['project'].slug}"
|
|
"&include_remote=true",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_branches_skips_empty_lines(git_client: dict) -> None:
|
|
"""Line 246: empty line in branch output triggers continue."""
|
|
branch_result = MagicMock()
|
|
# Embed an empty line between two branches.
|
|
branch_result.stdout = "main|abc\n\nfeature/x|def\n"
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc.get_current_branch = AsyncMock(return_value="main")
|
|
svc._run_git = AsyncMock(return_value=branch_result)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/branches?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_branches_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(side_effect=NotFoundError("no ws"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/branches?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# diff
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_basic(git_client: dict) -> None:
|
|
diff_res = MagicMock(stdout="some diff")
|
|
stat_res = MagicMock(stdout="a.py | 2\nb.py | 4\n2 files changed\n")
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc._run_git = AsyncMock(side_effect=[diff_res, stat_res])
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/diff?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_staged_with_file(git_client: dict) -> None:
|
|
diff_res = MagicMock(stdout="")
|
|
stat_res = MagicMock(stdout="")
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc._run_git = AsyncMock(side_effect=[diff_res, stat_res])
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/diff?project_slug={git_client['project'].slug}"
|
|
"&staged=true&file_path=a.py",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(side_effect=NotFoundError("no ws"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/diff?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# commit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_commit_success(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.commit_for_task = AsyncMock(return_value=("abc123", "msg", 1, 5, 2))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/commit",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
"message": "fix some thing",
|
|
"commit_type": "fix",
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_commit_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.commit_for_task = AsyncMock(side_effect=ValidationError("bad"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/commit",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
"message": "fix some thing",
|
|
"commit_type": "fix",
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# push
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_success(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.push_for_task = AsyncMock(return_value=("feature/x", 2))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/push",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.push_for_task = AsyncMock(side_effect=NotFoundError("missing"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/push",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# branch/create
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_branch_success(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.create_branch_for_task = AsyncMock(
|
|
return_value=("feature/backend/X", "main")
|
|
)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/branch/create",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"branch_type": "feature",
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_branch_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.create_branch_for_task = AsyncMock(side_effect=UnauthorizedError("no perm"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/branch/create",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"branch_type": "feature",
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# checkout
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_checkout_success(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.checkout_branch_for_agent = AsyncMock(return_value=None)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/checkout",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"branch": "feature/x",
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_checkout_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.checkout_branch_for_agent = AsyncMock(side_effect=UnauthorizedError("nope"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/checkout",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"branch": "master",
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# pr/create
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_pr_success(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.create_pr_for_task = AsyncMock(
|
|
return_value=(42, "https://github.com/x/y/pull/42", "T", "feat", "main")
|
|
)
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/pr/create",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_pr_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.create_pr_for_task = AsyncMock(side_effect=ValidationError("bad"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/pr/create",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# pr/merge
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_merge_pr_success(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.merge_pr_for_task = AsyncMock(return_value=("main", "abc"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/pr/merge",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"pr_number": 42,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_merge_pr_service_error(git_client: dict) -> None:
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.merge_pr_for_task = AsyncMock(side_effect=NotFoundError("no PR"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].post(
|
|
"/api/git/pr/merge",
|
|
json={
|
|
"project_slug": git_client["project"].slug,
|
|
"pr_number": 42,
|
|
"task_id": str(uuid4()),
|
|
"agent_id": str(uuid4()),
|
|
},
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UUID resolution path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_with_uuid(git_client: dict) -> None:
|
|
"""Pass a UUID string instead of slug — should look up by UUID."""
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(return_value="/tmp/ws")
|
|
svc.get_status = AsyncMock(return_value=("main", False, [], [], [], 0, 0))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/status?project_slug={git_client['project'].id}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Generic ServiceError -> default 500
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_generic_service_error(git_client: dict) -> None:
|
|
"""Any other ServiceError becomes 500."""
|
|
|
|
class CustomError(ServiceError):
|
|
pass
|
|
|
|
with patch("roboco.api.routes.git.get_git_service") as mock_get:
|
|
svc = AsyncMock()
|
|
svc.get_workspace = AsyncMock(side_effect=CustomError("err"))
|
|
mock_get.return_value = svc
|
|
response = await git_client["client"].get(
|
|
f"/api/git/status?project_slug={git_client['project'].slug}",
|
|
headers=_HDR,
|
|
)
|
|
assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
|
|
|
|
|
|
# Re-export to keep import alive (TC reorders imports)
|
|
_ = SimpleNamespace
|