feat(bot): клиент SourceCraft API — прогоны, логи кубов, sha коммита, ретраи

- SourceCraftClient (httpx): list_runs, get_run, get_cube_logs (постранично), run_commit_sha; ретраи на сетевые ошибки и 5xx с экспоненциальной задержкой, 4xx — SourceCraftError без повторов - модели: RunStatus/EventType из swagger, «1970-01-01» = не задано, failed_cubes без служебных #-кубов - у прогона в API нет sha → куб check/meta в tgit-uploads печатает TGIT_COMMIT_SHA (ADR-014) - CLI: python -m app.sourcecraft.cli runs - тесты на реальных ответах API (фикстуры), порог покрытия 45 → 80 Refs #10 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 18:45:42 +03:00
parent 282920e06b
commit 25735fee96
18 changed files with 1433 additions and 4 deletions
@@ -0,0 +1,28 @@
+"""Debug CLI: `python -m app.sourcecraft.cli runs [repo]` — recent CI runs with their commit and failed cubes."""
+
+import asyncio
+import sys
+
+from app.config import SourceCraftSettings
+from app.sourcecraft.client import SourceCraftClient
+
+
+async def runs(repo: str | None) -> None:
+    settings = SourceCraftSettings()
+    repo = repo or settings.uploads_repo
+    async with SourceCraftClient(settings) as client:
+        for run in await client.list_runs(repo, page_size=10):
+            full = await client.get_run(repo, run.slug)
+            sha = await client.run_commit_sha(repo, full)
+            failed = ", ".join(f"{t}/{c.slug}" for _, t, c in full.failed_cubes()) or "-"
+            print(f"{run.slug}  {run.event_type:<8} {run.status:<10} sha={sha or '?':<40.12}  failed={failed}")
+
+
+def main() -> None:
+    if len(sys.argv) < 2 or sys.argv[1] != "runs":
+        sys.exit("usage: python -m app.sourcecraft.cli runs [org/repo]")
+    asyncio.run(runs(sys.argv[2] if len(sys.argv) > 2 else None))
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,140 @@
+"""SourceCraft REST API client (ARCHITECTURE §2.4). The API drops TLS connections now and then: every call retries.
+
+Runs don't carry a commit sha, so the uploads CI prints it from its first cube (`check/meta`, ADR-014);
+`run_commit_sha` reads it back.
+"""
+
+import asyncio
+import logging
+import re
+from collections.abc import Awaitable, Callable
+from types import TracebackType
+from typing import Any, Self
+
+import httpx
+
+from app.config import SourceCraftSettings
+from app.sourcecraft.models import CubeLog, Run
+
+log = logging.getLogger(__name__)
+
+META_TASK = "check"
+META_CUBE = "meta"
+_SHA_LINE = re.compile(r"^TGIT_COMMIT_SHA=([0-9a-f]{40})\s*$", re.MULTILINE)
+
+Sleep = Callable[[float], Awaitable[None]]
+
+
+class SourceCraftError(Exception):
+    def __init__(self, status: int, code: str, message: str) -> None:
+        super().__init__(f"{status} {code}: {message}")
+        self.status = status
+        self.code = code
+        self.message = message
+
+
+class SourceCraftClient:
+    def __init__(
+        self,
+        settings: SourceCraftSettings,
+        transport: httpx.AsyncBaseTransport | None = None,
+        sleep: Sleep = asyncio.sleep,
+    ) -> None:
+        self._settings = settings
+        self._sleep = sleep
+        self._http = httpx.AsyncClient(
+            base_url=settings.api_url,
+            headers={"Authorization": f"Bearer {settings.token.get_secret_value()}"},
+            timeout=settings.timeout,
+            transport=transport,
+        )
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None
+    ) -> None:
+        await self.aclose()
+
+    async def aclose(self) -> None:
+        await self._http.aclose()
+
+    def run_url(self, repo: str, run: str) -> str:
+        return self._settings.run_url_template.format(repo=repo, run=run)
+
+    async def list_runs(self, repo: str, page_size: int = 20) -> list[Run]:
+        """Newest first. One page is enough for the watcher: it only looks for runs of recent pushes."""
+        data = await self._get(f"/repos/{repo}/cicd/runs", params={"page_size": page_size})
+        return [Run.model_validate(r) for r in data.get("runs", [])]
+
+    async def get_run(self, repo: str, run: str) -> Run:
+        return Run.model_validate(await self._get(f"/repos/{repo}/cicd/runs/{run}"))
+
+    async def get_cube_logs(self, repo: str, run: str, workflow: str, task: str, cube: str) -> CubeLog:
+        """Concatenate log pages in order; a page that isn't complete yet is the last readable one."""
+        parts: list[str] = []
+        done = False
+        page = 1
+        while True:
+            try:
+                data = await self._get(f"/repos/{repo}/cicd/logs/{run}/{workflow}/{task}/{cube}", params={"page": page})
+            except SourceCraftError as e:
+                if e.status == 404 and page > 1:
+                    break
+                raise
+            parts.append(data.get("logs", ""))
+            done = bool(data.get("done", False))
+            if not data.get("page_complete", False):
+                break
+            page += 1
+        return CubeLog(text="".join(parts), done=done)
+
+    async def run_commit_sha(self, repo: str, run: Run) -> str | None:
+        """The commit a run was started for, read from the meta cube's log (None if not printed yet)."""
+        for workflow in run.workflows:
+            if run.task(workflow.slug, META_TASK) is None:
+                continue
+            try:
+                logs = await self.get_cube_logs(repo, run.slug, workflow.slug, META_TASK, META_CUBE)
+            except SourceCraftError as e:
+                if e.status == 404:
+                    return None
+                raise
+            match = _SHA_LINE.search(logs.text)
+            return match.group(1) if match else None
+        return None
+
+    async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
+        attempts = self._settings.max_attempts
+        for attempt in range(1, attempts + 1):
+            try:
+                response = await self._http.get(path, params=params)
+            except httpx.TransportError as e:
+                if attempt == attempts:
+                    raise
+                log.warning("sourcecraft %s: %s, retry %d/%d", path, type(e).__name__, attempt, attempts - 1)
+            else:
+                if response.status_code < 500:
+                    return self._parse(response)
+                if attempt == attempts:
+                    return self._parse(response)
+                log.warning("sourcecraft %s: HTTP %d, retry %d/%d", path, response.status_code, attempt, attempts - 1)
+            await self._sleep(min(0.5 * 2 ** (attempt - 1), 8.0))
+        raise AssertionError("unreachable")
+
+    @staticmethod
+    def _parse(response: httpx.Response) -> dict[str, Any]:
+        try:
+            body = response.json()
+        except ValueError:
+            body = {}
+        if response.status_code >= 400:
+            raise SourceCraftError(
+                response.status_code,
+                str(body.get("error_code", "")),
+                str(body.get("message", response.text[:200])),
+            )
+        if not isinstance(body, dict):
+            raise SourceCraftError(response.status_code, "BadResponse", "expected a JSON object")
+        return body
@@ -0,0 +1,113 @@
+"""Typed SourceCraft CI/CD API responses (only the fields the bot uses)."""
+
+from collections.abc import Iterator
+from datetime import datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+class RunStatus(StrEnum):
+    CREATED = "created"
+    PREPARED = "prepared"
+    PROCESSING = "processing"
+    SUCCESS = "success"
+    FAILED = "failed"
+    CANCELED = "canceled"
+    TIMEOUT = "timeout"
+    SKIPPED = "skipped"
+    AWAITING_APPROVAL = "awaiting_approval"
+    REJECTED = "rejected"
+
+    @property
+    def finished(self) -> bool:
+        return self in _FINISHED
+
+
+_FINISHED = {
+    RunStatus.SUCCESS,
+    RunStatus.FAILED,
+    RunStatus.CANCELED,
+    RunStatus.TIMEOUT,
+    RunStatus.SKIPPED,
+    RunStatus.REJECTED,
+}
+
+
+class EventType(StrEnum):
+    PUSH = "push"
+    PR_UPDATE = "pr_update"
+    MANUAL = "manual"
+    RESTART = "restart"
+    SCHEDULE = "schedule"
+    REPOSITORY_EVENT = "repository_event"
+
+
+class _Model(BaseModel):
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+
+class Dates(_Model):
+    created_at: datetime | None = None
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+
+    @field_validator("created_at", "started_at", "finished_at", mode="after")
+    @classmethod
+    def _epoch_is_unset(cls, value: datetime | None) -> datetime | None:
+        # The API reports "not yet" as 1970-01-01T00:00:00Z
+        return None if value is not None and value.year == 1970 else value
+
+
+class Cube(_Model):
+    slug: str
+    status: RunStatus
+    dates: Dates = Field(default_factory=Dates)
+
+    @property
+    def is_service(self) -> bool:
+        """Platform steps (#git-clone, #prepare-image-for-…): not in our config, logs not readable."""
+        return self.slug.startswith("#")
+
+
+class Task(_Model):
+    slug: str
+    status: RunStatus
+    cubes: list[Cube] = []
+
+
+class Workflow(_Model):
+    slug: str
+    status: RunStatus
+    tasks: list[Task] = []
+
+
+class Run(_Model):
+    slug: str
+    status: RunStatus
+    event_type: EventType
+    dates: Dates = Field(default_factory=Dates)
+    workflows: list[Workflow] = []
+    error_messages: list[str] = []
+
+    def task(self, workflow: str, task: str) -> Task | None:
+        for wf in self.workflows:
+            if wf.slug == workflow:
+                return next((t for t in wf.tasks if t.slug == task), None)
+        return None
+
+    def cubes(self) -> Iterator[tuple[str, str, Cube]]:
+        """(workflow, task, cube) for every user-defined cube."""
+        for wf in self.workflows:
+            for t in wf.tasks:
+                for c in t.cubes:
+                    if not c.is_service:
+                        yield wf.slug, t.slug, c
+
+    def failed_cubes(self) -> list[tuple[str, str, Cube]]:
+        return [entry for entry in self.cubes() if entry[2].status is RunStatus.FAILED]
+
+
+class CubeLog(_Model):
+    text: str
+    done: bool