ElebobaandClaude Opus 5.5
95a72437ec
CI / ruff · mypy · pytest (pull_request) Failing after 3m40s
- app/pipeline: протокол CIProvider, клиент Gitea Actions (прогон по head_sha, упавший шаг, лог джоба, ретраи), PipelineWatcher — опрос только при незавершённых push, сообщение о push обновляется CI ⏳ → ✅/❌ → CD 🚀/⛔, автору — отдельные уведомления - деплой сервисов — только после зелёного CI; старый push не откатывает сервис, который развернёт более новый; после рестарта бот продолжает с того же места - CI_PROVIDER=none — прежнее поведение (деплой сразу после push) - шаблон tgit-uploads: .gitea/workflows/check.yml; .gitea/ защищён от правок из бота - миграция 0002: старые push помечены skipped, чтобы не слать им «таймаут CI» - ADR-016, ARCHITECTURE §2.5/§5.2, README, DEPLOY, .env.example Refs #11 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.config import CISettings
|
|
from app.db.models import PipelineStatus
|
|
from app.pipeline.gitea import GiteaCI, GiteaError
|
|
from app.pipeline.provider import CIRun
|
|
|
|
API = "https://gitea.example.test"
|
|
RUNS = f"{API}/api/v1/repos/HSE/up/actions/runs"
|
|
SHA = "eda569427093acb3b39ad5b301216cb5b7400ab8"
|
|
|
|
|
|
def run(
|
|
id: int, path: str = "check.yml@refs/heads/main", status: str = "completed", conclusion: Any = "success"
|
|
) -> Any:
|
|
"""Shape of a real Gitea 1.27 run (fields the client doesn't read trimmed)."""
|
|
run = {"id": id, "path": path, "status": status, "head_sha": SHA, "html_url": f"{API}/HSE/up/actions/runs/{id}"}
|
|
return run | ({"conclusion": conclusion} if conclusion else {}) # in-progress runs have no conclusion key
|
|
|
|
|
|
def job(id: int, conclusion: str, steps: list[tuple[str, str]]) -> Any:
|
|
return {
|
|
"id": id,
|
|
"name": "check",
|
|
"status": "completed",
|
|
"conclusion": conclusion,
|
|
"steps": [{"name": n, "number": i, "status": "completed", "conclusion": c} for i, (n, c) in enumerate(steps)],
|
|
}
|
|
|
|
|
|
async def no_sleep(_: float) -> None:
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
async def ci() -> Any:
|
|
c = GiteaCI(CISettings(token="secret", api_url=API, repo="HSE/up", max_attempts=3), sleep=no_sleep) # type: ignore[arg-type]
|
|
yield c
|
|
await c.aclose()
|
|
|
|
|
|
@respx.mock
|
|
async def test_finds_newest_run_of_the_workflow(ci: GiteaCI) -> None:
|
|
other = run(9, path="other.yml@refs/heads/main")
|
|
route = respx.get(RUNS).respond(
|
|
json={"workflow_runs": [run(5), other, run(7, status="in_progress", conclusion=None)]}
|
|
)
|
|
|
|
found = await ci.find(SHA)
|
|
|
|
assert found == CIRun("7", PipelineStatus.RUNNING, f"{API}/HSE/up/actions/runs/7")
|
|
assert route.calls.last.request.url.params["head_sha"] == SHA
|
|
assert route.calls.last.request.headers["Authorization"] == "token secret"
|
|
|
|
|
|
@respx.mock
|
|
async def test_no_run_yet(ci: GiteaCI) -> None:
|
|
respx.get(RUNS).respond(json={"workflow_runs": [], "total_count": 0})
|
|
|
|
assert await ci.find(SHA) is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "conclusion", "expected"),
|
|
[
|
|
("queued", None, PipelineStatus.PENDING),
|
|
("waiting", None, PipelineStatus.PENDING),
|
|
("in_progress", None, PipelineStatus.RUNNING),
|
|
("completed", "success", PipelineStatus.SUCCESS),
|
|
("completed", "skipped", PipelineStatus.SKIPPED),
|
|
],
|
|
)
|
|
@respx.mock
|
|
async def test_status_mapping(ci: GiteaCI, status: str, conclusion: str | None, expected: PipelineStatus) -> None:
|
|
respx.get(RUNS).respond(json={"workflow_runs": [run(1, status=status, conclusion=conclusion)]})
|
|
|
|
found = await ci.find(SHA)
|
|
|
|
assert found is not None and found.status is expected
|
|
|
|
|
|
@respx.mock
|
|
async def test_failure_names_the_step_and_reads_its_log(ci: GiteaCI) -> None:
|
|
respx.get(RUNS).respond(json={"workflow_runs": [run(3, conclusion="failure")]})
|
|
steps = [("Run actions/checkout@v4", "success"), ("lint", "failure"), ("tests", "skipped")]
|
|
respx.get(f"{RUNS}/3/jobs").respond(json={"jobs": [job(31, "failure", steps)], "total_count": 1})
|
|
log = "2026-10-18T20:44:56.7721431Z ruff check .\n2026-10-18T20:44:57.1Z F821 undefined name 'x'\n"
|
|
respx.get(f"{API}/api/v1/repos/HSE/up/actions/jobs/31/logs").respond(text=log)
|
|
|
|
found = await ci.find(SHA)
|
|
assert found is not None
|
|
text = await ci.failure_log(found)
|
|
|
|
assert (found.status, found.failed_step) == (PipelineStatus.FAILED, "check / lint")
|
|
assert text == "ruff check .\nF821 undefined name 'x'\n"
|
|
|
|
|
|
@respx.mock
|
|
async def test_cancelled_run_is_a_failure(ci: GiteaCI) -> None:
|
|
respx.get(RUNS).respond(json={"workflow_runs": [run(4, conclusion="cancelled")]})
|
|
respx.get(f"{RUNS}/4/jobs").respond(json={"jobs": [job(41, "cancelled", [("lint", "success")])]})
|
|
|
|
found = await ci.find(SHA)
|
|
|
|
assert found is not None and (found.status, found.failed_step) == (PipelineStatus.FAILED, "check / cancelled")
|
|
|
|
|
|
@respx.mock
|
|
async def test_retries_transport_errors_and_5xx(ci: GiteaCI) -> None:
|
|
route = respx.get(RUNS)
|
|
route.side_effect = [
|
|
httpx.ConnectError("reset"),
|
|
httpx.Response(502),
|
|
httpx.Response(200, json={"workflow_runs": []}),
|
|
]
|
|
|
|
assert await ci.find(SHA) is None
|
|
assert route.call_count == 3
|
|
|
|
|
|
@respx.mock
|
|
async def test_gives_up_with_a_typed_error(ci: GiteaCI) -> None:
|
|
respx.get(RUNS).respond(404, json={"message": "not found"})
|
|
|
|
with pytest.raises(GiteaError, match="404"):
|
|
await ci.find(SHA)
|