bot-bottle/bot_bottle/contrib/gitea/client.py

"""Gitea HTTP client + `GiteaForge` (PRD forge-native-integration, chunk 3).

`GiteaClient` is the thin stdlib-only HTTP transport (mirrors
`deploy_key_provisioner.py`: `urllib.request`, bounded timeouts,
structured error bodies). `GiteaForge` adapts it to the provider-agnostic
`Forge` surface.

Unlike the option-2 design, the token is held here (the sidecar process
owns it) and passed to the client directly — there is no agent-side
cred-proxy route, because the agent never makes forge calls. The HTTP
client is the one piece shared with `GiteaDeployKeyProvisioner`; the two
are deliberately *not* unified behind a common abstract base (see the
deferral note in the PRD).
"""

from __future__ import annotations

import json
import urllib.error
import urllib.request
from typing import Any

from ..forge.base import Comment, Forge, Issue, PullRequest

# Bound every Gitea call: a hung instance must not stall the sidecar.
_API_TIMEOUT_SECS = 30


class GiteaClient:
    """Thin authenticated HTTP client for one repo's Gitea API.

    `api_url` is the API base *including* `/api/v1` (matching the
    `FORGE_GITEA_API` env var), e.g. `https://gitea.example.com/api/v1`.
    """

    def __init__(self, *, api_url: str, owner: str, repo: str, token: str) -> None:
        self._api_url = api_url.rstrip("/")
        self._owner = owner
        self._repo = repo
        self._token = token

    # --- low-level request -------------------------------------------------

    def _request(
        self, method: str, path: str, *, body: dict[str, Any] | None = None
    ) -> tuple[int, Any]:
        """Issue an authenticated request. Returns `(status, parsed_json)`;
        parsed_json is None when the response has no body. Raises
        `RuntimeError` on any non-2xx except where callers special-case
        the HTTPError themselves (membership 404)."""
        url = f"{self._api_url}{path}"
        data = json.dumps(body).encode() if body is not None else None
        headers = {"Authorization": f"token {self._token}"}
        if data is not None:
            headers["Content-Type"] = "application/json"
        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        with urllib.request.urlopen(req, timeout=_API_TIMEOUT_SECS) as resp:
            raw = resp.read()
            parsed = json.loads(raw) if raw else None
            return resp.status, parsed

    def _repo_path(self, suffix: str) -> str:
        return f"/repos/{self._owner}/{self._repo}{suffix}"

    # --- operations --------------------------------------------------------

    def is_org_member(self, org: str, username: str) -> bool:
        """GET /orgs/{org}/members/{username}: 2xx → member, 404 → not.
        Other errors propagate so a misconfigured token fails loudly."""
        url = f"{self._api_url}/orgs/{org}/members/{username}"
        req = urllib.request.Request(
            url, headers={"Authorization": f"token {self._token}"}, method="GET"
        )
        try:
            with urllib.request.urlopen(req, timeout=_API_TIMEOUT_SECS):
                return True
        except urllib.error.HTTPError as exc:
            if exc.code == 404:
                return False
            raise RuntimeError(
                f"org membership check failed for {org}/{username}: "
                f"HTTP {exc.code} — {_read_error_body(exc)}"
            ) from exc

    def get_issue(self, number: int) -> dict[str, Any]:
        _status, body = self._request("GET", self._repo_path(f"/issues/{number}"))
        return body or {}

    def get_comments(self, number: int) -> list[dict[str, Any]]:
        _status, body = self._request(
            "GET", self._repo_path(f"/issues/{number}/comments")
        )
        return body or []

    def post_comment(self, number: int, body: str) -> None:
        self._request(
            "POST",
            self._repo_path(f"/issues/{number}/comments"),
            body={"body": body},
        )

    def patch_issue_body(self, number: int, body: str) -> None:
        self._request(
            "PATCH", self._repo_path(f"/issues/{number}"), body={"body": body}
        )

    def get_pull(self, number: int) -> dict[str, Any]:
        _status, body = self._request("GET", self._repo_path(f"/pulls/{number}"))
        return body or {}


class GiteaForge(Forge):
    """`Forge` over a `GiteaClient`."""

    def __init__(self, client: GiteaClient) -> None:
        self._client = client

    def read_issue(self, number: int) -> Issue:
        raw = self._client.get_issue(number)
        return Issue(
            number=int(raw.get("number", number)),
            title=str(raw.get("title", "")),
            body=str(raw.get("body", "") or ""),
            state=str(raw.get("state", "")),
        )

    def read_pr(self, number: int) -> PullRequest:
        raw = self._client.get_pull(number)
        return PullRequest(
            number=int(raw.get("number", number)),
            title=str(raw.get("title", "")),
            body=str(raw.get("body", "") or ""),
            state=str(raw.get("state", "")),
            merged=bool(raw.get("merged", False)),
        )

    def read_comments(self, number: int) -> list[Comment]:
        return [
            Comment(
                id=int(c.get("id", 0)),
                user=str((c.get("user") or {}).get("login", "")),
                body=str(c.get("body", "") or ""),
            )
            for c in self._client.get_comments(number)
        ]

    def post_comment(self, number: int, body: str) -> None:
        self._client.post_comment(number, body)

    def update_description(self, number: int, body: str) -> None:
        self._client.patch_issue_body(number, body)

    def is_org_member(self, org: str, username: str) -> bool:
        return self._client.is_org_member(org, username)

    def get_pr_for_issue(self, number: int) -> int | None:
        """Gitea models a PR as an issue with the same number, exposing a
        `pull_request` object on the issue. When the queried number is
        itself a PR, return it; otherwise None. (The orchestrator tracks
        the issue→PR mapping in forge state for the cross-number case.)"""
        raw = self._client.get_issue(number)
        if raw.get("pull_request"):
            return int(raw.get("number", number))
        return None

    def is_pr_open(self, number: int) -> bool:
        return self.read_pr(number).state == "open"


def _read_error_body(exc: urllib.error.HTTPError) -> str:
    try:
        return exc.read().decode("utf-8", errors="replace")
    except Exception:  # pylint: disable=broad-exception-caught
        return ""