diff --git a/gitea_saas/gitea_client.py b/gitea_saas/gitea_client.py new file mode 100644 index 0000000..20c7a23 --- /dev/null +++ b/gitea_saas/gitea_client.py @@ -0,0 +1,132 @@ +"""Gitea API client (mockable seam, env token). + +Shared surface for the SaaS front-ends: the factory front-end (issue #4), +the review gate (issue #5) and the eval farm (issue #8) all drive Gitea +through this seam. Real HTTP only when a token is available; the token is +read from ``GITEA_TOKEN`` (or passed in) and is never printed or placed on +a command line. Tests use an in-memory fake, so this module is never +exercised over the network in CI. + +``owner`` defaults to the ``GITEA_OWNER`` environment variable and then to +``pbwraith`` (the self-hosted instance's user). Repo-scoped calls build +``/api/v1/repos/{owner}/{repo}/...``; org-repo creation uses +``/api/v1/orgs/{owner}/repos`` and falls back to the authenticated user's +repos when ``owner`` is a user account (orgs endpoint 404s). +""" + +from __future__ import annotations + +import json +import os +import urllib.error +import urllib.request +from typing import Any, List, Optional + + +class GiteaClient: + """Thin urllib client for the subset of the Gitea API the SaaS needs.""" + + def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None, + owner: Optional[str] = None): + self.base_url = (base_url or os.environ.get( + "GITEA_BASE_URL", "https://gitea.185.214.135.234.nip.io")).rstrip("/") + self.token = token if token is not None else os.environ.get("GITEA_TOKEN", "") + self.owner = (owner or os.environ.get("GITEA_OWNER") or "pbwraith").strip() or "pbwraith" + + # -- low-level --------------------------------------------------------- + + def _api(self, method: str, api_path: str, body: Optional[dict] = None) -> Any: + """Perform an API call under ``/api/v1/{api_path}`` and parse JSON.""" + data = json.dumps(body).encode() if body is not None else None + url = f"{self.base_url}/api/v1/{api_path}" + req = urllib.request.Request( + url, data=data, method=method, + headers={"Authorization": "token " + self.token, + "Accept": "application/json", + "Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=30) as resp: + raw = resp.read() + return json.loads(raw) if raw else None + + def _repo_api(self, method: str, repo: str, path: str = "", + body: Optional[dict] = None) -> Any: + return self._api(method, f"repos/{self.owner}/{repo}{path}", body) + + # -- repos ------------------------------------------------------------- + + def repo_exists(self, name: str) -> bool: + """True when ``owner/`` exists (404 -> False).""" + try: + self._repo_api("GET", name) + return True + except urllib.error.HTTPError as e: + if e.code == 404: + return False + raise + + def create_org_repo(self, name: str, private: bool = True) -> dict: + """Create ``name`` under the owner (org endpoint, user fallback).""" + payload = {"name": name, "private": private, + "description": "internal SaaS eval/task repo", + "auto_init": True} + try: + created = self._api("POST", f"orgs/{self.owner}/repos", payload) + except urllib.error.HTTPError: + created = self._api("POST", "user/repos", payload) + return created or {"name": name} + + # -- issues ------------------------------------------------------------ + + def _label_ids(self, repo: str, labels: List[str]) -> List[int]: + """Resolve label names to ids, creating missing labels first.""" + if not labels: + return [] + try: + existing = self._repo_api("GET", repo, "/labels?limit=50") or [] + except urllib.error.HTTPError: + existing = [] + by_name = {l.get("name"): l.get("id") for l in existing if isinstance(l, dict)} + ids: List[int] = [] + for name in labels: + label_id = by_name.get(name) + if label_id is None: + try: + created = self._repo_api( + "POST", repo, "/labels", + {"name": name, "color": "#1d76db", + "description": "SaaS worker label"}) + label_id = created.get("id") + except urllib.error.HTTPError: + label_id = None + if label_id is not None: + ids.append(int(label_id)) + return ids + + def create_issue(self, repo: str, title: str, body: str, + labels: Optional[List[str]] = None) -> dict: + """Create an issue; returns the created issue record.""" + payload: dict = {"title": title, "body": body or ""} + label_ids = self._label_ids(repo, list(labels or [])) + if label_ids: + payload["labels"] = label_ids + return self._repo_api("POST", repo, "/issues", payload) + + def get_issue(self, repo: str, number: int) -> dict: + """Fetch an issue: {title, body, state, labels: [name, ...]}.""" + issue = self._repo_api("GET", repo, f"/issues/{number}") or {} + return { + "title": issue.get("title"), + "body": issue.get("body"), + "state": issue.get("state"), + "labels": [l.get("name") for l in issue.get("labels", []) + if isinstance(l, dict)], + } + + def list_issue_comments(self, repo: str, number: int) -> List[dict]: + """All comments on an issue conversation: [{body, ...}, ...].""" + return self._repo_api("GET", repo, f"/issues/{number}/comments") or [] + + def post_comment(self, repo: str, number: int, body: str) -> dict: + """Post a comment on an issue or PR conversation.""" + return self._repo_api("POST", repo, f"/issues/{number}/comments", + {"body": body}) diff --git a/gitea_saas/intake_eval.py b/gitea_saas/intake_eval.py new file mode 100644 index 0000000..93c82df --- /dev/null +++ b/gitea_saas/intake_eval.py @@ -0,0 +1,466 @@ +"""Eval farm front-end (issue #8): batch intake -> lane drain -> results emitter. + +Turns the lane machine into a paid eval/QA grind (option 2 from +ARCHITECTURE.md). An eval batch is a set of tasks; each task becomes one +Gitea issue (labelled ``eval-task``, carrying a ``batch: `` marker +in its body) so the lane fleet picks it up like any funded work. Results +are collected by polling the task issues for closure, then a JSON report +is emitted to a callback webhook (retry x3 with backoff) or dropped as a +file into ``SAAS_EVAL_OUTPUT_DIR``. + +Money rule (fail-closed, mirroring issue #4's factory semantics): a batch +is only accepted when the client's available balance is at or above the +debit floor. Unfunded/unknown clients get a 402-style structured refusal +and NO Gitea issues are ever created for them. + +Schema additions are owned here (``schema.sql`` belongs to issue #1): +``eval_batches`` and ``eval_tasks`` are bootstrapped idempotently against +the same SQLite file as the billing core, exactly like the factory does. +""" + +from __future__ import annotations + +import json +import os +import re +import sqlite3 +import threading +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +# Default funded threshold: a client must hold at least 5.00 USD before an +# eval batch is accepted (same default as the factory front-end, issue #4). +DEFAULT_DEBIT_FLOOR_MINOR = 500 + +DEFAULT_EVAL_REPO = "eval-runs" +EVAL_LABEL = "eval-task" + +_STATUS_OPEN = "open" +_STATUS_DONE = "done" +_STATUS_EMITTED = "emitted" +_STATUS_FAILED = "failed" + +_TASK_PENDING = "pending" +_TASK_DONE = "done" + +_REPO_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$") + +# Idempotent by construction (same convention as the billing schema). +_EVAL_DDL = """ +CREATE TABLE IF NOT EXISTS eval_batches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + client_id INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'open' + CHECK (status IN ('open', 'done', 'emitted', 'failed')), + callback_url TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) +); + +CREATE TABLE IF NOT EXISTS eval_tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + batch_id INTEGER NOT NULL REFERENCES eval_batches(id), + repo TEXT NOT NULL, + issue_number INTEGER NOT NULL, + task_id TEXT NOT NULL, + payload_json TEXT NOT NULL DEFAULT '{}', + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'done')), + result_json TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE (batch_id, task_id) +); + +CREATE INDEX IF NOT EXISTS idx_eval_tasks_batch ON eval_tasks (batch_id, status); +""" + + +class EvalIntakeError(ValueError): + """Base class for eval intake domain errors (malformed input only).""" + + +def _now_iso() -> str: + """Current UTC time as an ISO-8601 string (matches billing core format).""" + return datetime.now(timezone.utc).isoformat() + + +class HttpEmitter: + """Webhook emitter: POST a JSON report; retry x3 with backoff. + + Returns ``True`` only on a 2xx response. ``sleep_fn`` is injectable so + tests stay hermetic (no real backoff sleeps). + """ + + def __init__(self, timeout: int = 20, max_attempts: int = 3, + sleep_fn=None): + self.timeout = timeout + self.max_attempts = max_attempts + self._sleep = sleep_fn or time.sleep + + def push(self, report: dict, url: str) -> bool: + import urllib.request + + attempts = 0 + while attempts < self.max_attempts: + attempts += 1 + req = urllib.request.Request( + url, data=json.dumps(report).encode(), method="POST", + headers={"Content-Type": "application/json", + "Accept": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + if 200 <= resp.status < 300: + return True + except Exception: + pass # retry below + if attempts < self.max_attempts: + self._sleep(2 ** attempts) # 2s, 4s backoff + return False + + +class EvalIntake: + """Batch eval intake: submit -> lane drain -> poll -> emit. + + Parameters + ---------- + core: + The billing core (issue #1 contract): ``get_client(client_id, + include_balance=True) -> (ok, details)``. Duck-typed; never + imported directly. + gitea: + A :class:`gitea_saas.gitea_client.GiteaClient` (or in-memory mock): + ``repo_exists``, ``create_org_repo``, ``create_issue``, + ``get_issue``, ``list_issue_comments``. + config: + Configuration object exposing ``db_path`` (a ``pathlib.Path`` to + the billing SQLite file). Optional attributes: ``eval_repo`` + (shared eval repo name), ``output_dir`` (file-drop destination), + ``debit_floor_minor`` (funded threshold), ``per_batch_repos``. + Env fallbacks: ``SAAS_DB_PATH``, ``SAAS_EVAL_REPO``, + ``SAAS_EVAL_OUTPUT_DIR``, ``SAAS_DEBIT_FLOOR_MINOR``. + emitter: + Optional duck-typed emitter used for callback delivery: + ``push(report, url) -> bool``. Defaults to :class:`HttpEmitter`. + sleep_fn: + Backoff sleep override for hermetic tests (default ``time.sleep``). + """ + + def __init__(self, core: Any, gitea: Any, config: Any, + emitter: Optional[Any] = None, sleep_fn=None): + db_path = getattr(config, "db_path", None) + if db_path is None: + raw = os.environ.get("SAAS_DB_PATH", "").strip() + db_path = Path(raw).expanduser() if raw else None + if db_path is None: + raise EvalIntakeError( + "EvalIntake requires config.db_path (or SAAS_DB_PATH): " + "the batch/task queue persists in the billing SQLite file") + self._core = core + self._gitea = gitea + self._config = config + self._emitter = emitter if emitter is not None else HttpEmitter( + sleep_fn=sleep_fn) + self._sleep = sleep_fn or time.sleep + self._conn = sqlite3.connect(str(Path(db_path)), check_same_thread=False) + self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA foreign_keys=ON") + self._conn.executescript(_EVAL_DDL) + self._lock = threading.RLock() + + # ------------------------------------------------------------------ + # Config + # ------------------------------------------------------------------ + + @property + def debit_floor_minor(self) -> int: + value = getattr(self._config, "debit_floor_minor", None) + if value is None: + raw = os.environ.get("SAAS_DEBIT_FLOOR_MINOR", "").strip() + value = int(raw) if raw else DEFAULT_DEBIT_FLOOR_MINOR + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise EvalIntakeError(f"invalid debit floor: {value!r}") + return value + + @property + def eval_repo(self) -> str: + value = getattr(self._config, "eval_repo", None) + if value is None: + value = os.environ.get("SAAS_EVAL_REPO", "").strip() or DEFAULT_EVAL_REPO + return str(value).strip() or DEFAULT_EVAL_REPO + + @property + def output_dir(self) -> Optional[Path]: + value = getattr(self._config, "output_dir", None) + if value is None: + raw = os.environ.get("SAAS_EVAL_OUTPUT_DIR", "").strip() + value = Path(raw).expanduser() if raw else None + return Path(value) if value else None + + @property + def per_batch_repos(self) -> bool: + return bool(getattr(self._config, "per_batch_repos", False)) + + # ------------------------------------------------------------------ + # Billing helpers (fail-closed, factory semantics) + # ------------------------------------------------------------------ + + def _client_balance(self, client_id: int) -> Optional[int]: + """Available balance, or None when funding cannot be confirmed.""" + try: + ok, details = self._core.get_client(client_id, include_balance=True) + except Exception: + return None + if not ok or not isinstance(details, dict): + return None + balance = details.get("balance") + return balance if isinstance(balance, int) else None + + def client_unfunded(self, client_id: int) -> bool: + """True when the client's balance is below the debit floor. + + Fail-closed: an unknown client or a core error also reports True. + """ + balance = self._client_balance(client_id) + if balance is None: + return True + return balance < self.debit_floor_minor + + def _require_valid_inputs(self, client_id: int, tasks: List[dict]) -> None: + if isinstance(client_id, bool) or not isinstance(client_id, int) \ + or client_id <= 0: + raise EvalIntakeError("client_id must be a positive integer") + if not isinstance(tasks, list) or not tasks: + raise EvalIntakeError("tasks must be a non-empty list") + seen: set = set() + for t in tasks: + if not isinstance(t, dict): + raise EvalIntakeError("each task must be an object") + task_id = str(t.get("task_id") or "").strip() + if not task_id: + raise EvalIntakeError("each task needs a non-empty task_id") + if task_id in seen: + raise EvalIntakeError(f"duplicate task_id {task_id!r}") + seen.add(task_id) + if not str(t.get("title") or "").strip(): + raise EvalIntakeError("each task needs a title") + + # ------------------------------------------------------------------ + # Submitting a batch + # ------------------------------------------------------------------ + + def submit_batch(self, client_id: int, tasks: List[dict], + callback_url: Optional[str] = None) -> Dict[str, Any]: + """Accept a funded batch; one eval-task issue per task. + + Returns ``{ok: True, batch_id, repo, issues: [{task_id, + issue_number}]}``. Unfunded/unknown clients get a 402-style + ``{ok: False, reason: 'unfunded', status: 402}`` and NO Gitea + issues are created. + """ + self._require_valid_inputs(client_id, tasks) + if self.client_unfunded(client_id): + return {"ok": False, "reason": "unfunded", "status": 402} + callback_url = (callback_url or "").strip() or None + + with self._lock: + cur = self._conn.execute( + "INSERT INTO eval_batches (client_id, status, callback_url, created_at)" + " VALUES (?, ?, ?, ?)", + (client_id, _STATUS_OPEN, callback_url, _now_iso())) + batch_id = int(cur.lastrowid) + + # Shared eval repo (SAAS_EVAL_REPO) or one repo per batch. + if self.per_batch_repos: + repo = f"eval-{client_id}-{batch_id}" + else: + repo = self.eval_repo + if not _REPO_NAME_RE.match(repo): + repo = DEFAULT_EVAL_REPO + try: + if not self._gitea.repo_exists(repo): + self._gitea.create_org_repo(repo) + issues = [] + for t in tasks: + task_id = str(t["task_id"]).strip() + title = str(t["title"]).strip() + body = str(t.get("body") or "").strip() + body = f"{body}\n\n---\nbatch: {batch_id}".strip() + issue = self._gitea.create_issue( + repo, title, body, labels=[EVAL_LABEL]) + issue_number = int(issue["number"]) + self._conn.execute( + "INSERT INTO eval_tasks (batch_id, repo, issue_number," + " task_id, payload_json, status, created_at)" + " VALUES (?, ?, ?, ?, ?, ?, ?)", + (batch_id, repo, issue_number, task_id, + json.dumps({"title": title, "body": body}), + _TASK_PENDING, _now_iso())) + issues.append({"task_id": task_id, + "issue_number": issue_number}) + self._conn.commit() + except Exception as exc: # fail-closed: no partial batches + self._conn.execute( + "DELETE FROM eval_tasks WHERE batch_id = ?", (batch_id,)) + self._conn.execute( + "DELETE FROM eval_batches WHERE id = ?", (batch_id,)) + self._conn.commit() + return {"ok": False, "reason": "intake_failed", + "error": str(exc), "status": 500} + + return {"ok": True, "batch_id": batch_id, "repo": repo, + "issues": issues} + + # ------------------------------------------------------------------ + # Polling completed tasks + # ------------------------------------------------------------------ + + def _batch_row(self, batch_id: int) -> Optional[sqlite3.Row]: + return self._conn.execute( + "SELECT * FROM eval_batches WHERE id = ?", (batch_id,)).fetchone() + + def _task_rows(self, batch_id: int) -> List[sqlite3.Row]: + return self._conn.execute( + "SELECT * FROM eval_tasks WHERE batch_id = ? ORDER BY id", + (batch_id,)).fetchall() + + def poll_completed(self, batch_id: int) -> Dict[str, Any]: + """Collect results from closed task issues; skip open ones. + + A task is done when its issue is closed (the lane merged); the + result record captures the final conversation comment (the lane's + report) plus the issue's labels. + """ + batch = self._batch_row(batch_id) + if batch is None: + return {"ok": False, "error": "unknown_batch"} + done = 0 + pending = 0 + for row in self._task_rows(batch_id): + if row["status"] == _TASK_DONE: + done += 1 + continue + issue = self._gitea.get_issue(row["repo"], row["issue_number"]) + if not issue or issue.get("state") != "closed": + pending += 1 + continue + comments = self._gitea.list_issue_comments( + row["repo"], row["issue_number"]) or [] + summary = "" + if comments: + last = comments[-1] + summary = last.get("body") if isinstance(last, dict) else str(last) + result = { + "status": "done", + "repo": row["repo"], + "issue_number": row["issue_number"], + "labels": list(issue.get("labels") or []), + "summary": (summary or "").strip(), + } + with self._lock: + self._conn.execute( + "UPDATE eval_tasks SET status = ?, result_json = ?" + " WHERE id = ?", + (_TASK_DONE, json.dumps(result), row["id"])) + remaining = self._conn.execute( + "SELECT COUNT(*) FROM eval_tasks WHERE batch_id = ?" + " AND status = 'pending'", (batch_id,)).fetchone()[0] + if remaining == 0 and batch["status"] == _STATUS_OPEN: + self._conn.execute( + "UPDATE eval_batches SET status = ? WHERE id = ?", + (_STATUS_DONE, batch_id)) + self._conn.commit() + done += 1 + return {"ok": True, "batch_id": batch_id, "done": done, + "pending": pending} + + # ------------------------------------------------------------------ + # Emitting results + # ------------------------------------------------------------------ + + def _report(self, batch: sqlite3.Row) -> dict: + results = [] + for row in self._task_rows(batch["id"]): + try: + payload = json.loads(row["payload_json"] or "{}") + except ValueError: + payload = {} + try: + result = json.loads(row["result_json"] or "{}") + except ValueError: + result = {} + results.append({ + "task_id": row["task_id"], + "status": "done" if row["status"] == _TASK_DONE else row["status"], + "summary": result.get("summary") or "", + "title": payload.get("title") or "", + }) + return { + "batch_id": batch["id"], + "client_id": batch["client_id"], + "status": batch["status"], + "results": results, + } + + def emit(self, batch_id: int) -> Dict[str, Any]: + """Emit the full batch report once every task is done. + + Callback mode: POST to ``callback_url`` (retry x3, backoff) via the + emitter seam. File mode: write ``SAAS_EVAL_OUTPUT_DIR/.json``. + A batch already emitted re-emits idempotently (no second POST/write). + """ + batch = self._batch_row(batch_id) + if batch is None: + return {"ok": False, "error": "unknown_batch"} + pending = self._conn.execute( + "SELECT COUNT(*) FROM eval_tasks WHERE batch_id = ?" + " AND status = 'pending'", (batch_id,)).fetchone()[0] + if pending > 0: + return {"ok": False, "reason": "tasks_pending", + "pending": int(pending)} + if batch["status"] == _STATUS_EMITTED: + return {"ok": True, "idempotent": True, "batch_id": batch_id, + "report": self._report(batch)} + report = self._report(batch) + + if batch["callback_url"]: + delivered = False + attempts = 0 + while attempts < 3 and not delivered: + attempts += 1 + delivered = bool(self._emitter.push(report, batch["callback_url"])) + if not delivered and attempts < 3: + self._sleep(2 ** attempts) + if not delivered: + return {"ok": False, "reason": "emit_failed", + "attempts": attempts} + else: + out_dir = self.output_dir + if out_dir is None: + return {"ok": False, "reason": "no_callback_no_output_dir"} + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / f"{batch_id}.json").write_text( + json.dumps(report, indent=2), encoding="utf-8") + + with self._lock: + self._conn.execute( + "UPDATE eval_batches SET status = ? WHERE id = ?", + (_STATUS_EMITTED, batch_id)) + self._conn.commit() + return {"ok": True, "batch_id": batch_id, "emitted": True, + "report": report} + + # ------------------------------------------------------------------ + # Reads (for server/UI wiring — issue #2's file may expose these) + # ------------------------------------------------------------------ + + def get_batch(self, batch_id: int) -> Dict[str, Any]: + batch = self._batch_row(batch_id) + if batch is None: + return {"ok": False, "error": "unknown_batch"} + tasks = [dict(r) for r in self._task_rows(batch_id)] + for t in tasks: + t["payload_json"] = json.loads(t["payload_json"] or "{}") + t["result_json"] = json.loads(t["result_json"] or "null") + return {"ok": True, "batch": dict(batch), "tasks": tasks} diff --git a/tests/test_intake_eval.py b/tests/test_intake_eval.py new file mode 100644 index 0000000..678ab52 --- /dev/null +++ b/tests/test_intake_eval.py @@ -0,0 +1,364 @@ +"""Hermetic tests for the eval farm intake (issue #8 acceptance). + +MockGitea (in-memory repos/issues/comments/labels), a scripted emitter +(retry/backoff verified via call counts), and a real BillingCore on a tmp +DB (funded vs unfunded gating is real). No network, no sleeps (a noop +``sleep_fn`` is injected everywhere). + +Run: python3 -m pytest tests/test_intake_eval.py -q +""" +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import pytest + +from gitea_saas.billing_core import BillingCore +from gitea_saas.intake_eval import EvalIntake, EvalIntakeError, EVAL_LABEL + + +class Config: + """Tiny config object for the intake (db_path + optional overrides).""" + + def __init__(self, db_path, **kwargs): + self.db_path = Path(db_path) + for k, v in kwargs.items(): + setattr(self, k, v) + + +class MockGitea: + """In-memory Gitea seam recording every mutation.""" + + def __init__(self): + self.repos = {} # repo -> {"issues": {n: dict}, "comments": {n: [str]}} + self.next_number = 0 # global issue numbering + self.created_repos = [] # repo names created + self.created_issues = [] # (repo, title, body, labels) + + def _repo(self, name): + if name not in self.repos: + self.repos[name] = {"issues": {}, "comments": {}} + return self.repos[name] + + def repo_exists(self, name): + return name in self.repos + + def create_org_repo(self, name): + self._repo(name) + self.created_repos.append(name) + + def create_issue(self, repo, title, body, labels=None): + r = self._repo(repo) + self.next_number += 1 + n = self.next_number + r["issues"][n] = {"title": title, "body": body, + "state": "open", "labels": list(labels or [])} + r["comments"][n] = [] + self.created_issues.append((repo, title, body, list(labels or []))) + return {"number": n, "title": title, "body": body, "state": "open"} + + def get_issue(self, repo, number): + r = self._repo(repo) + issue = r["issues"].get(number) + if issue is None: + return None + return dict(issue) + + def list_issue_comments(self, repo, number): + return [{"body": b} for b in self._repo(repo)["comments"].get(number, [])] + + # -- test helpers: simulate the lane draining a task ------------------- + def add_comment(self, repo, number, body): + self._repo(repo)["comments"].get(number, []).append(body) + + def close_issue(self, repo, number): + self._repo(repo)["issues"][number]["state"] = "closed" + + +class MockEmitter: + """Scripted emitter: fails the first ``fail_first`` pushes.""" + + def __init__(self, fail_first=0): + self.fail_first = fail_first + self.push_count = 0 + self.push_payloads = [] # (report, url) + + def push(self, report, url): + self.push_count += 1 + if self.push_count <= self.fail_first: + return False + self.push_payloads.append((report, url)) + return True + + +def noop_sleep(_seconds): + return None + + +@pytest.fixture +def funded(tmp_path): + core = BillingCore(tmp_path / "billing.db") + ok, c = core.create_client("eval-client") + assert ok + ok, _ = core.credit(c["client_id"], 10_000, "fund-1") + assert ok + return core, c["client_id"] + + +@pytest.fixture +def unfunded(tmp_path): + core = BillingCore(tmp_path / "billing.db") + ok, c = core.create_client("poor-client") + assert ok + return core, c["client_id"] + + +def make_intake(core, tmp_path, gitea=None, emitter=None, **config_kwargs): + gitea = gitea or MockGitea() + cfg = Config(tmp_path / "saas.db", **config_kwargs) + return EvalIntake(core, gitea, cfg, emitter=emitter, sleep_fn=noop_sleep) + + +TASKS = [ + {"task_id": "t1", "title": "Eval task one", + "body": "## Spec\nrun X\n## Acceptance\n- [ ] a\n- [ ] b"}, + {"task_id": "t2", "title": "Eval task two", "body": "## Spec\nrun Y"}, +] + + +# -------------------------------------------------------------------------- # +# Submission # +# -------------------------------------------------------------------------- # + +class TestSubmitBatch: + def test_funded_batch_creates_one_issue_per_task(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea) + r = intake.submit_batch(client_id, TASKS) + assert r["ok"] is True + assert r["batch_id"] > 0 + assert r["repo"] == "eval-runs" + assert [i["task_id"] for i in r["issues"]] == ["t1", "t2"] + # shared repo exists; two issues; each carries the eval-task label + assert "eval-runs" in gitea.created_repos + assert len(gitea.created_issues) == 2 + for repo, title, body, labels in gitea.created_issues: + assert repo == "eval-runs" + assert labels == [EVAL_LABEL] + assert f"batch: {r['batch_id']}" in body + # both tasks persisted as pending + batch = intake.get_batch(r["batch_id"]) + assert batch["ok"] and len(batch["tasks"]) == 2 + assert all(t["status"] == "pending" for t in batch["tasks"]) + assert batch["batch"]["status"] == "open" + + def test_unfunded_batch_402_and_no_gitea_mutations(self, unfunded, tmp_path): + core, client_id = unfunded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea) + r = intake.submit_batch(client_id, TASKS) + assert r["ok"] is False + assert r["reason"] == "unfunded" + assert r["status"] == 402 + assert gitea.created_issues == [] # no issues created + assert gitea.created_repos == [] # no repo created either + assert gitea.next_number == 0 + + def test_unknown_client_is_fail_closed_unfunded(self, tmp_path): + core = BillingCore(tmp_path / "billing.db") + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea) + r = intake.submit_batch(9999, TASKS) + assert r["ok"] is False and r["reason"] == "unfunded" + assert r["status"] == 402 + assert gitea.created_issues == [] + + def test_duplicate_task_id_rejected(self, funded, tmp_path): + core, client_id = funded + intake = make_intake(core, tmp_path) + with pytest.raises(EvalIntakeError): + intake.submit_batch(client_id, [TASKS[0], dict(TASKS[0])]) + + def test_intake_failure_creates_no_partial_batch(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + + class FlakyGitea(MockGitea): + def create_issue(self, *a, **kw): + if self.next_number >= 1: # second issue fails + raise RuntimeError("gitea hiccup") + return super().create_issue(*a, **kw) + + intake = make_intake(core, tmp_path, FlakyGitea()) + r = intake.submit_batch(client_id, TASKS) + assert r["ok"] is False and r["reason"] == "intake_failed" + # no batch row left behind + assert intake.get_batch(1)["ok"] is False + + +# -------------------------------------------------------------------------- # +# Polling # +# -------------------------------------------------------------------------- # + +class TestPollCompleted: + def test_poll_stores_closed_results_and_skips_open(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea) + r = intake.submit_batch(client_id, TASKS) + n1 = r["issues"][0]["issue_number"] + gitea.add_comment("eval-runs", n1, "lane report: all checks green") + gitea.close_issue("eval-runs", n1) # task 1 drained; task 2 still open + out = intake.poll_completed(r["batch_id"]) + assert out["done"] == 1 and out["pending"] == 1 + batch = intake.get_batch(r["batch_id"]) + done_task = [t for t in batch["tasks"] if t["task_id"] == "t1"][0] + pending_task = [t for t in batch["tasks"] if t["task_id"] == "t2"][0] + assert done_task["status"] == "done" + result = done_task["result_json"] + assert result["summary"] == "lane report: all checks green" + assert result["labels"] == [EVAL_LABEL] + assert pending_task["status"] == "pending" + assert batch["batch"]["status"] == "open" # not all done yet + + def test_poll_marks_batch_done_when_all_closed(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea) + r = intake.submit_batch(client_id, TASKS) + for issue in r["issues"]: + gitea.add_comment("eval-runs", issue["issue_number"], "ok") + gitea.close_issue("eval-runs", issue["issue_number"]) + out = intake.poll_completed(r["batch_id"]) + assert out["done"] == 2 and out["pending"] == 0 + assert intake.get_batch(r["batch_id"])["batch"]["status"] == "done" + + def test_poll_unknown_batch(self, funded, tmp_path): + core, client_id = funded + intake = make_intake(core, tmp_path) + out = intake.poll_completed(123) + assert out["ok"] is False and out["error"] == "unknown_batch" + + +# -------------------------------------------------------------------------- # +# Emitting # +# -------------------------------------------------------------------------- # + +class TestEmit: + def test_emit_posts_report_to_callback_with_retry(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + # fails twice (500s), succeeds on the third attempt + emitter = MockEmitter(fail_first=2) + intake = make_intake(core, tmp_path, gitea, emitter=emitter) + r = intake.submit_batch(client_id, TASKS, + callback_url="https://client.example/hook") + for issue in r["issues"]: + gitea.add_comment("eval-runs", issue["issue_number"], + "done: all checks passed") + gitea.close_issue("eval-runs", issue["issue_number"]) + intake.poll_completed(r["batch_id"]) + + out = intake.emit(r["batch_id"]) + assert out["ok"] is True and out["emitted"] is True + assert emitter.push_count == 3 # retry x3 total + assert len(emitter.push_payloads) == 1 # delivered once + report, url = emitter.push_payloads[0] + assert url == "https://client.example/hook" + assert report["batch_id"] == r["batch_id"] + assert report["client_id"] == client_id + assert [res["task_id"] for res in report["results"]] == ["t1", "t2"] + assert all(res["status"] == "done" for res in report["results"]) + assert all("summary" in res for res in report["results"]) + assert intake.get_batch(r["batch_id"])["batch"]["status"] == "emitted" + + def test_emit_file_mode_writes_json_drop(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea, + output_dir=tmp_path / "drops") + r = intake.submit_batch(client_id, TASKS) # no callback_url + for issue in r["issues"]: + gitea.close_issue("eval-runs", issue["issue_number"]) + intake.poll_completed(r["batch_id"]) + out = intake.emit(r["batch_id"]) + assert out["ok"] is True and out["emitted"] is True + drop = tmp_path / "drops" / f"{r['batch_id']}.json" + assert drop.exists() + payload = json.loads(drop.read_text(encoding="utf-8")) + assert payload["batch_id"] == r["batch_id"] + assert len(payload["results"]) == 2 + + def test_emit_refuses_before_all_tasks_done(self, funded, tmp_path): + core, client_id = funded + intake = make_intake(core, tmp_path) + r = intake.submit_batch(client_id, TASKS) + out = intake.emit(r["batch_id"]) + assert out["ok"] is False and out["reason"] == "tasks_pending" + assert intake.get_batch(r["batch_id"])["batch"]["status"] == "open" + + def test_reemit_is_idempotent_no_double_post(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + emitter = MockEmitter() + intake = make_intake(core, tmp_path, gitea, emitter=emitter) + r = intake.submit_batch(client_id, TASKS, + callback_url="https://cb.example/x") + for issue in r["issues"]: + gitea.close_issue("eval-runs", issue["issue_number"]) + intake.poll_completed(r["batch_id"]) + assert intake.emit(r["batch_id"])["ok"] is True + pushes_after_first = emitter.push_count + again = intake.emit(r["batch_id"]) + assert again["ok"] is True and again.get("idempotent") is True + assert emitter.push_count == pushes_after_first # no second POST + + def test_callback_failure_keeps_batch_emittable(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + emitter = MockEmitter(fail_first=99) # always fails + intake = make_intake(core, tmp_path, gitea, emitter=emitter) + r = intake.submit_batch(client_id, TASKS, + callback_url="https://cb.example/x") + for issue in r["issues"]: + gitea.close_issue("eval-runs", issue["issue_number"]) + intake.poll_completed(r["batch_id"]) + out = intake.emit(r["batch_id"]) + assert out["ok"] is False and out["reason"] == "emit_failed" + assert out["attempts"] == 3 + assert intake.get_batch(r["batch_id"])["batch"]["status"] == "done" + + +# -------------------------------------------------------------------------- # +# Persistence + config # +# -------------------------------------------------------------------------- # + +class TestPersistence: + def test_batch_survives_intake_reopen(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea) + r = intake.submit_batch(client_id, TASKS) + intake2 = make_intake(core, tmp_path, gitea) # fresh connection + batch = intake2.get_batch(r["batch_id"]) + assert batch["ok"] and batch["batch"]["client_id"] == client_id + assert len(batch["tasks"]) == 2 + + def test_shared_repo_name_from_config(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea, eval_repo="qa-runs") + r = intake.submit_batch(client_id, TASKS) + assert r["repo"] == "qa-runs" + assert "qa-runs" in gitea.created_repos + + def test_per_batch_repo_mode(self, funded, tmp_path): + core, client_id = funded + gitea = MockGitea() + intake = make_intake(core, tmp_path, gitea, per_batch_repos=True) + r = intake.submit_batch(client_id, TASKS) + assert r["repo"] == f"eval-{client_id}-{r['batch_id']}" + assert r["repo"] in gitea.created_repos