diff --git a/gitea_saas/gitea_client.py b/gitea_saas/gitea_client.py new file mode 100644 index 0000000..e30d356 --- /dev/null +++ b/gitea_saas/gitea_client.py @@ -0,0 +1,162 @@ +"""Gitea API client (mockable seam, env token). + +Shared by the factory front-end (issue #4) and the review gate (issue #5). +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. +""" + +from __future__ import annotations + +import json +import os +import urllib.error +import urllib.request +from typing import Any, Optional + + +class GiteaClient: + """Thin urllib client for the subset of the Gitea API the SaaS needs. + + ``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). + """ + + 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 _request(self, method: str, path: str, body: Optional[dict] = None) -> Any: + """Repo-scoped call: ``path`` = ``/...`` under the owner.""" + return self._api(method, f"repos/{self.owner}/{path}", body) + + # -- API surface used by the factory front-end (issue #4) -------------- + + def repo_exists(self, name: str) -> bool: + """True when ``owner/name`` already exists on the instance. + + A 404 means absent; any other HTTP error propagates (fail-closed: + we never claim a repo is missing when the check itself failed). + """ + try: + self._request("GET", f"{name}") + except urllib.error.HTTPError as exc: + if exc.code == 404: + return False + raise + return True + + def create_org_repo(self, name: str) -> dict: + """Create ``owner/name`` if missing; return the repo record. + + Idempotent by construction: when the repo already exists the result + of ``repo_exists`` is returned immediately (no duplicate creation). + Uses ``orgs/{owner}/repos`` when ``owner`` is an organization and + falls back to the authenticated user's repos (``user/repos``) when + it is a user account -- the self-hosted default. + """ + name = (name or "").strip() + if not name: + raise ValueError("repo name must be a non-empty string") + if self.repo_exists(name): + # Reuse: fetch the record so callers see the canonical shape. + return self._request("GET", f"{name}") or {} + payload: dict = {"name": name, "auto_init": False, "private": True} + try: + return self._api("POST", f"orgs/{self.owner}/repos", payload) + except urllib.error.HTTPError as exc: + if exc.code not in (404, 422): + raise + # ``owner`` is a user account (or lacks org-create rights): fall back + # to creating the repo under the authenticated token user. + return self._api("POST", "user/repos", payload) or {} + + def list_labels(self, repo: str) -> list: + """List ``owner/repo``'s labels: [{id, name, ...}, ...].""" + return self._request("GET", f"{repo}/labels") or [] + + def label_id(self, repo: str, name: str) -> Optional[int]: + """Resolve a label name to its id in ``owner/repo`` (None if absent).""" + name = (name or "").strip() + for label in self.list_labels(repo): + if label.get("name") == name: + return label.get("id") + return None + + def create_issue(self, repo: str, title: str, body: str, + labels: Optional[list] = None) -> int: + """Create an issue in ``owner/repo``; return its number. + + ``labels`` is a list of label NAMES; each is resolved to the repo's + label id and unknown names are skipped (the issue is still created). + """ + title = (title or "").strip() + if not title: + raise ValueError("issue title must be a non-empty string") + label_ids = [] + for name in labels or []: + if not isinstance(name, str) or not name.strip(): + continue + label_id = self.label_id(repo, name) + if label_id is not None and label_id not in label_ids: + label_ids.append(label_id) + payload = {"title": title, "body": body or ""} + if label_ids: + payload["labels"] = label_ids + created = self._request("POST", f"{repo}/issues", payload) or {} + number = created.get("number") + if not isinstance(number, int): + raise RuntimeError(f"Gitea create_issue returned no issue number: {created!r}") + return number + + # -- API surface used by the review gate (issue #5) -------------------- + + def get_issue(self, repo: str, number: int) -> dict: + """Fetch an issue: {title, body, state, labels: [...]}.""" + return self._request("GET", f"{repo}/issues/{number}") or {} + + def get_pr(self, repo: str, number: int) -> dict: + """Fetch a PR: {head_sha, changed_files: [filename, ...]}.""" + pr = self._request("GET", f"{repo}/pulls/{number}") or {} + files = self._request("GET", f"{repo}/pulls/{number}/files") or [] + return { + "head_sha": ((pr.get("head") or {}).get("sha")), + "changed_files": [f.get("filename") for f in files if f.get("filename")], + } + + def get_pr_diff(self, repo: str, number: int) -> str: + """Return the PR's diff text reconstructed from per-file patches.""" + files = self._request("GET", f"{repo}/pulls/{number}/files") or [] + chunks = [] + for f in files: + filename = f.get("filename") or "?" + chunks.append(f"diff --git a/{filename} b/{filename}\n" + f"--- a/{filename}\n+++ b/{filename}\n" + + (f.get("patch") or "")) + return "\n".join(chunks) + + def post_comment(self, repo: str, number: int, body: str) -> dict: + """Post a comment on an issue or PR conversation.""" + return self._request("POST", f"{repo}/issues/{number}/comments", {"body": body}) or {} diff --git a/gitea_saas/intake_factory.py b/gitea_saas/intake_factory.py new file mode 100644 index 0000000..be83715 --- /dev/null +++ b/gitea_saas/intake_factory.py @@ -0,0 +1,465 @@ +"""Factory front-end: client repo provisioning + unfunded issue gating. + +Option-1 front-end from ARCHITECTURE.md. All gates live HERE, inside the +SaaS service -- the lane survey/dispatcher scripts are never touched for +SaaS concepts (owner rule 2026-08-25): unfunded client work is held in a +local ``pending_issues`` queue and only becomes a Gitea issue after payment. + +Owns its schema additions against the SAME SQLite file as the billing core. +``schema.sql`` / ``db.py`` stay owned by issue #1; the factory tables are +created idempotently (``CREATE TABLE IF NOT EXISTS ...``) from this module +so the feature can land without touching the core's files: + + client_repos (client_id, repo, created_at) -- repo->client map + pending_issues (id, client_id, repo, title, body_json, labels_json, + created_at, status) -- queued|shipped + factory_locks (client_id, locked_at) -- explicit gating + +Money rule (fail-closed): a client is funded iff its available balance is +>= the debit floor and no lock is set. Any uncertainty (unknown client, +core error) is treated as UNFUNDED so work is queued, never silently +shipped without payment. +""" + +from __future__ import annotations + +import hmac +import json +import os +import re +import sqlite3 +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +# Header carrying the API key. The shared constant lives in the sibling +# ``gitea_saas/config.py`` (issue #2/#6 auth), which lands in a later PR than +# this one; until that module is merged this file falls back to the same +# canonical value. Keeping the single source of truth means the factory +# routes use ``compare_digest`` on the same header name as ``server.py``. +try: + from .config import API_KEY_HEADER # shared with gitea_saas.server +except ImportError: # config.py not merged yet (sibling auth PRs) + API_KEY_HEADER = "X-API-Key" + +# Default floor for "funded": a client must hold at least this many minor +# units (default = 5.00 USD) before intake will ship issues. +DEFAULT_DEBIT_FLOOR_MINOR = 500 + +# Label applied to every issue flushed by :meth:`IntakeFactory.unlock`. +P1_LABEL = "p1" + +_STATUS_QUEUED = "queued" +_STATUS_SHIPPED = "shipped" + +_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). +_FACTORY_DDL = """ +CREATE TABLE IF NOT EXISTS client_repos ( + client_id INTEGER NOT NULL, + repo TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + PRIMARY KEY (client_id, repo) +); + +CREATE TABLE IF NOT EXISTS pending_issues ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + client_id INTEGER NOT NULL, + repo TEXT NOT NULL, + title TEXT NOT NULL, + body_json TEXT NOT NULL, + labels_json TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'shipped')) +); + +CREATE TABLE IF NOT EXISTS factory_locks ( + client_id INTEGER PRIMARY KEY, + locked_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) +); + +CREATE INDEX IF NOT EXISTS idx_pending_client_status ON pending_issues (client_id, status); +""" + + +def _now_iso() -> str: + """Current UTC time as an ISO-8601 string (matches billing core format).""" + return datetime.now(timezone.utc).isoformat() + + +class FactoryError(Exception): + """Base class for factory domain errors.""" + + +class UnknownClientError(FactoryError): + """Raised when a client_id does not exist in the billing core.""" + + +class InvalidRequestError(FactoryError): + """Raised on malformed inputs (empty names, bad repo names, ...).""" + + +class CoreContractError(FactoryError): + """Raised when the billing core violates its (ok, details) contract.""" + + +class IntakeFactory: + """Provision client repos and gate issue intake on funding. + + Parameters + ---------- + core: + The billing core (issue #1 contract): ``get_client(client_id, + include_balance=...) -> (ok, details)``. Duck-typed; never + imported directly so the module stays composable. + gitea: + A :class:`gitea_saas.gitea_client.GiteaClient` (or in-memory mock): + ``create_org_repo(name)``, ``repo_exists(name)``, + ``create_issue(repo, title, body, labels)``. + config: + Configuration object exposing ``db_path`` (a ``pathlib.Path`` to the + billing SQLite file). Optionally ``debit_floor_minor`` and/or the + ``SAAS_DEBIT_FLOOR_MINOR`` env var override the default floor. + """ + + def __init__(self, core: Any, gitea: Any, config: Any): + 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 InvalidRequestError( + "IntakeFactory requires config.db_path (or SAAS_DB_PATH): " + "the factory persists its queue in the billing SQLite file") + self._core = core + self._gitea = gitea + self._config = config + 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(_FACTORY_DDL) + self._lock = threading.RLock() + + # ------------------------------------------------------------------ + # Config + # ------------------------------------------------------------------ + + @property + def debit_floor_minor(self) -> int: + """The funded threshold in minor units (config attr > env > default).""" + 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 InvalidRequestError(f"invalid debit floor: {value!r}") + return value + + # ------------------------------------------------------------------ + # Billing-core helpers (fail-closed) + # ------------------------------------------------------------------ + + def _client_record(self, client_id: int, include_balance: bool = False) -> Dict[str, Any]: + """Fetch a client record from the core; raise when unknown/broken.""" + ok, details = self._core.get_client(client_id, include_balance=include_balance) + if not ok: + code = (details or {}).get("error", {}).get("code") if isinstance(details, dict) else None + if code and "UNKNOWN" in str(code).upper(): + raise UnknownClientError(f"no client with id {client_id}") + raise CoreContractError(f"core.get_client({client_id}) failed: {details!r}") + if not isinstance(details, dict) or (include_balance and not isinstance( + details.get("balance"), int)): + raise CoreContractError(f"core.get_client({client_id}) violated its contract: {details!r}") + return details + + def _client_balance(self, client_id: int) -> Optional[int]: + """Available balance, or None when funding cannot be confirmed.""" + try: + details = self._client_record(client_id, include_balance=True) + except (UnknownClientError, CoreContractError): + return None + return details.get("balance") + + 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`` -- intake queues instead of shipping unverified work. + """ + balance = self._client_balance(client_id) + if balance is None: + return True + return balance < self.debit_floor_minor + + def _require_client(self, client_id: int) -> Dict[str, Any]: + """Validate existence (no balance needed) and return the record.""" + if isinstance(client_id, bool) or not isinstance(client_id, int) or client_id <= 0: + raise InvalidRequestError("client_id must be a positive integer") + return self._client_record(client_id, include_balance=False) + + def _is_locked(self, client_id: int) -> bool: + row = self._conn.execute( + "SELECT 1 FROM factory_locks WHERE client_id = ?", (client_id,)).fetchone() + return row is not None + + # ------------------------------------------------------------------ + # Repo provisioning + # ------------------------------------------------------------------ + + @staticmethod + def _default_repo_name(client_record: Dict[str, Any]) -> str: + """Derive a repo name from the client record (client name slug). + + Accepts both the historical nested shape (``{client: {...}}``) and + the flat record now returned by :meth:`BillingCore.get_client` + (issue #16: ``{id, name, tier, status, meta, created_at, balance}``) + -- the factory composes with whichever contract the core exposes. + """ + name = (client_record.get("client") or {}).get("name") or client_record.get("name") or "" + slug = re.sub(r"[^A-Za-z0-9._-]+", "-", name.strip().lower()).strip("-") + cid_fallback = (client_record.get("client_id") + or client_record.get("id") + or (client_record.get("client") or {}).get("id") + or "") + return slug[:60] or f"client-{cid_fallback}" + + def provision_client(self, client_id: int, repo_name: Optional[str] = None) -> str: + """Provision ``owner/`` for a client; returns the repo name. + + Idempotent: a second call for the same client reuses the existing + mapping and never creates the repo twice. When ``repo_name`` is + omitted it is derived from the client's name. + """ + record = self._require_client(client_id) + if repo_name is not None: + repo_name = (repo_name or "").strip() + if not _REPO_NAME_RE.match(repo_name): + raise InvalidRequestError( + f"invalid repo name: {repo_name!r} (letters, digits, . _ -; 1-100 chars)") + else: + repo_name = self._default_repo_name(record) + with self._lock: + mapped = self._conn.execute( + "SELECT repo FROM client_repos WHERE client_id = ? AND repo = ?", + (client_id, repo_name)).fetchone() + if mapped is None: + self._gitea.create_org_repo(repo_name) + self._conn.execute( + "INSERT OR IGNORE INTO client_repos (client_id, repo, created_at)" + " VALUES (?, ?, ?)", (client_id, repo_name, _now_iso())) + self._conn.commit() + return repo_name + + # ------------------------------------------------------------------ + # Issue intake (funded shipping vs unfunded queue) + # ------------------------------------------------------------------ + + def submit_issue(self, client_id: int, repo: str, title: str, + acceptance_body: str, labels: Optional[List[str]] = None) -> Dict[str, Any]: + """Submit an issue for a client. + + Funded + unlocked -> created in Gitea immediately, + ``{delivered: True, issue_number: N}``. + Unfunded or locked -> queued in ``pending_issues``, + ``{delivered: False, pending_id: N, reason: 'unfunded'|'locked'}``. + """ + self._require_client(client_id) + repo = (repo or "").strip() + title = (title or "").strip() + if not _REPO_NAME_RE.match(repo): + raise InvalidRequestError(f"invalid repo name: {repo!r}") + if not title: + raise InvalidRequestError("issue title must be a non-empty string") + reason = "locked" if self._is_locked(client_id) else ( + "unfunded" if self.client_unfunded(client_id) else None) + if reason is not None: + with self._lock: + cur = self._conn.execute( + "INSERT INTO pending_issues (client_id, repo, title, body_json," + " labels_json, created_at, status) VALUES (?, ?, ?, ?, ?, ?, ?)", + (client_id, repo, title, + json.dumps(acceptance_body or ""), + json.dumps([str(l) for l in (labels or [])]), + _now_iso(), _STATUS_QUEUED)) + self._conn.commit() + pending_id = cur.lastrowid + return {"delivered": False, "pending_id": pending_id, "reason": reason} + issue_number = self._gitea.create_issue(repo, title, acceptance_body or "", labels) + return {"delivered": True, "issue_number": issue_number} + + # ------------------------------------------------------------------ + # Lock / unlock + # ------------------------------------------------------------------ + + def lock(self, client_id: int) -> Dict[str, Any]: + """Stop shipping: future submits queue instead of creating issues. + + Idempotent. No label games and no dispatcher coupling -- the gate + is this service's local state. + """ + self._require_client(client_id) + with self._lock: + self._conn.execute( + "INSERT OR REPLACE INTO factory_locks (client_id, locked_at)" + " VALUES (?, ?)", (client_id, _now_iso())) + self._conn.commit() + return {"client_id": client_id, "locked": True} + + def unlock(self, client_id: int) -> Dict[str, Any]: + """Flush ALL queued issues for the client into Gitea (with ``p1``). + + Each flushed issue is created with its original labels plus ``p1`` + and marked ``shipped``. Unlocking also clears an active lock so the + client ships again. A second unlock with nothing queued is a + no-op: ``{flushed: 0, issues: []}``. + """ + self._require_client(client_id) + with self._lock: + rows = self._conn.execute( + "SELECT * FROM pending_issues WHERE client_id = ? AND status = ?" + " ORDER BY id", (client_id, _STATUS_QUEUED)).fetchall() + numbers: List[int] = [] + for row in rows: + labels = [str(l) for l in json.loads(row["labels_json"] or "[]")] + if P1_LABEL not in labels: + labels.append(P1_LABEL) + number = self._gitea.create_issue( + row["repo"], row["title"], json.loads(row["body_json"] or '""'), labels) + self._conn.execute( + "UPDATE pending_issues SET status = ? WHERE id = ?", + (_STATUS_SHIPPED, row["id"])) + numbers.append(number) + cleared = self._conn.execute( + "DELETE FROM factory_locks WHERE client_id = ?", (client_id,)).rowcount + self._conn.commit() + return {"flushed": len(numbers), "issues": numbers, "lock_cleared": cleared > 0} + + +# --------------------------------------------------------------------------- +# HTTP surface (mounted by the issue #2 Flask app; see register_factory_routes) +# --------------------------------------------------------------------------- +# +# Routes (same auth discipline as gitea_saas.server -- API_KEY_HEADER via +# hmac.compare_digest, fail-closed): +# POST /api/factory/provision {"client_id": int, "repo_name"?: str} +# POST /api/factory//issues {"repo", "title", "acceptance_body", +# "labels"?: [...]} +# POST /api/factory//unlock {} +# POST /api/factory//lock {} +# +# server.py is owned by issues #2/#6, so the mount point itself stays out of +# this module; creating the app just calls:: + +_HOOK_NOTES = """\ + from gitea_saas.intake_factory import register_factory_routes + app = create_app(core, api_key) + register_factory_routes(app, core, gitea_client, cfg) +""" + + +def register_factory_routes(app: Any, core: Any, gitea: Any, config: Any) -> None: + """Mount the factory endpoints on a Flask app (idempotent). + + The app keeps its own ``CORE``/``API_KEY`` config keys; this function + only adds routes that build an :class:`IntakeFactory` per request (the + factory is stateless apart from its own SQLite connection, which is + cheap to open and thread-safe via WAL + a per-instance lock). + """ + from flask import jsonify, request + + api_key = app.config.get("API_KEY") or getattr(config, "api_key", None) or "" + + def _factory() -> IntakeFactory: + return IntakeFactory(core, gitea, config) + + def _authorized() -> bool: + sent = request.headers.get(API_KEY_HEADER, "") + return bool(api_key) and hmac.compare_digest(sent, api_key) + + def _require_auth(): + if not _authorized(): + resp = jsonify({"error": "missing or invalid API key", "code": "unauthorized"}) + resp.status_code = 401 + return resp + return None + + @app.get("/api/factory/health") + def factory_health(): + return jsonify({"ok": True}) + + @app.post("/api/factory/provision") + def factory_provision(): + if (denied := _require_auth()) is not None: + return denied + body = request.get_json(silent=True) or {} + raw_id = body.get("client_id") + if isinstance(raw_id, bool) or not isinstance(raw_id, int) or raw_id <= 0: + return _json_error(400, "invalid_request", "client_id must be a positive integer") + client_id = raw_id + repo_name = body.get("repo_name") + try: + repo = _factory().provision_client(client_id, repo_name or None) + except Exception as exc: + return _map_factory_error(exc) + return jsonify({"repo": repo}) + + @app.post("/api/factory//issues") + def factory_submit(client_id: int): + if (denied := _require_auth()) is not None: + return denied + body = request.get_json(silent=True) or {} + repo = body.get("repo") + title = body.get("title") + acceptance_body = body.get("acceptance_body") or "" + labels = body.get("labels") + if not isinstance(labels, list): + labels = None + try: + result = _factory().submit_issue( + client_id, repo or "", title or "", acceptance_body, + [str(l) for l in labels] if labels else None) + except Exception as exc: + return _map_factory_error(exc) + resp = jsonify(result) + resp.status_code = 202 if not result["delivered"] else 201 + return resp + + @app.post("/api/factory//unlock") + def factory_unlock(client_id: int): + if (denied := _require_auth()) is not None: + return denied + try: + return jsonify(_factory().unlock(client_id)) + except Exception as exc: + return _map_factory_error(exc) + + @app.post("/api/factory//lock") + def factory_lock(client_id: int): + if (denied := _require_auth()) is not None: + return denied + try: + return jsonify(_factory().lock(client_id)) + except Exception as exc: + return _map_factory_error(exc) + + +def _json_error(status: int, code: str, message: str): + from flask import jsonify + + resp = jsonify({"error": message, "code": code}) + resp.status_code = status + return resp + + +def _map_factory_error(exc: Exception): + """Map factory domain errors to HTTP responses (fail-closed on unknown).""" + if isinstance(exc, UnknownClientError): + return _json_error(404, "unknown_client", str(exc)) + if isinstance(exc, (InvalidRequestError, ValueError)): + return _json_error(400, "invalid_request", str(exc)) + if isinstance(exc, CoreContractError): + return _json_error(500, "core_contract", str(exc)) + return _json_error(500, "factory_error", f"{type(exc).__name__}: {exc}") diff --git a/tests/test_intake_factory.py b/tests/test_intake_factory.py new file mode 100644 index 0000000..f8144e5 --- /dev/null +++ b/tests/test_intake_factory.py @@ -0,0 +1,528 @@ +"""Tests for ``gitea_saas.intake_factory`` (factory intake, issue #4). + +Hermetic by design: an in-memory ``MockGitea`` + a reference billing core +over a temp SQLite DB. No network, no Stripe, no Gitea, no dispatcher +files. + +Covers the issue #4 acceptance checklist: + +1. provision creates repo once; second call reuses (idempotent) +2. funded submit creates Gitea issue immediately, delivered=true +3. unfunded submit queues (no Gitea issue created), delivered=false +4. unlock flushes queued -> Gitea with p1 label, marks shipped, second + unlock is a no-op +5. lock stops shipping: subsequent submits queue again +6. client with balance below floor is ``client_unfunded() == True`` +7. all tests against MockGitea + tmp DB; no network +""" + +from __future__ import annotations + +import json +import sqlite3 +import types +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from gitea_saas.intake_factory import ( + DEFAULT_DEBIT_FLOOR_MINOR, + IntakeFactory, + UnknownClientError, + register_factory_routes, +) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +# --------------------------------------------------------------------------- +# Reference billing core (issue #1 contract subset: (ok, details) tuples) +# --------------------------------------------------------------------------- + +class ReferenceCore: + """Minimal billing core honouring the documented ``(ok, details)`` + contract over a temp SQLite DB (subset used by the factory).""" + + SCHEMA = """ + CREATE TABLE IF NOT EXISTS clients ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + tier TEXT NOT NULL DEFAULT 'standard', + status TEXT NOT NULL DEFAULT 'active', + meta_json TEXT, + created_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS ledger ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + client_id INTEGER NOT NULL REFERENCES clients(id), + kind TEXT NOT NULL CHECK (kind IN ('credit','debit','hold','release','refund')), + amount_minor INTEGER NOT NULL, + currency TEXT NOT NULL DEFAULT 'USD', + idempotency_key TEXT UNIQUE NOT NULL, + ref_type TEXT, + ref_id TEXT, + created_at TEXT NOT NULL + ); + """ + + def __init__(self, db_path: Path): + self._conn = sqlite3.connect(str(db_path)) + self._conn.row_factory = sqlite3.Row + self._conn.executescript(self.SCHEMA) + self._seq = 0 + + def _key(self) -> str: + self._seq += 1 + return f"ref-{self._seq}" + + def create_client(self, name: str, meta: dict | None = None): + try: + cur = self._conn.execute( + "INSERT INTO clients (name, meta_json, created_at) VALUES (?, ?, ?)", + (name, json.dumps(meta or {}), _now_iso())) + except sqlite3.IntegrityError: + return False, {"error": {"code": "DUPLICATE_CLIENT", "message": name}} + self._conn.commit() + return True, {"client_id": cur.lastrowid, "name": name} + + def credit(self, client_id: int, amount_minor: int): + self._conn.execute( + "INSERT INTO ledger (client_id, kind, amount_minor, currency," + " idempotency_key, created_at) VALUES (?, 'credit', ?, 'USD', ?, ?)", + (client_id, amount_minor, self._key(), _now_iso())) + self._conn.commit() + return True, {"client_id": client_id, "balance": self._balance(client_id)} + + def _balance(self, client_id: int) -> int: + row = self._conn.execute( + "SELECT COALESCE(SUM(CASE WHEN kind='credit' THEN amount_minor" + " ELSE -amount_minor END), 0) FROM ledger WHERE client_id = ?", + (client_id,)).fetchone() + return int(row[0]) + + def get_client(self, client_id: int, include_balance: bool = True): + row = self._conn.execute( + "SELECT * FROM clients WHERE id = ?", (client_id,)).fetchone() + if row is None: + return False, {"error": {"code": "UNKNOWN_CLIENT", + "message": f"no client with id {client_id}"}} + client = dict(row) + client["meta"] = json.loads(client.pop("meta_json") or "{}") + details = {"client": client} + if include_balance: + details["balance"] = self._balance(client_id) + return True, details + + def get_balance(self, client_id: int): + row = self._conn.execute( + "SELECT id FROM clients WHERE id = ?", (client_id,)).fetchone() + if row is None: + return False, {"error": {"code": "UNKNOWN_CLIENT", "message": "no client"}} + return True, {"client_id": client_id, "balance": self._balance(client_id)} + + +# --------------------------------------------------------------------------- +# MockGitea (in-memory fake of gitea_saas.gitea_client.GiteaClient) +# --------------------------------------------------------------------------- + +class MockGitea: + """In-memory Gitea: repos, per-repo labels (names -> ids) and issues.""" + + def __init__(self): + self.repos: dict = {} + self.labels: dict = {} + self.created_repo_calls: list = [] + self._label_seq = 0 + + # -- repo provisioning ------------------------------------------------- + + def repo_exists(self, name: str) -> bool: + return name in self.repos + + def create_org_repo(self, name: str) -> dict: + self.created_repo_calls.append(name) + if name not in self.repos: + self.repos[name] = {"name": name, "issues": []} + self.labels[name] = {"p1": self._next_label_id()} + return {"name": name} + + # -- labels ------------------------------------------------------------ + + def _next_label_id(self) -> int: + self._label_seq += 1 + return self._label_seq + + def add_label(self, repo: str, name: str) -> int: + if name not in self.labels.setdefault(repo, {}): + self.labels[repo][name] = self._next_label_id() + return self.labels[repo][name] + + def list_labels(self, repo: str) -> list: + return [{"id": i, "name": n} for n, i in self.labels.get(repo, {}).items()] + + def label_id(self, repo: str, name: str) -> int | None: + return self.labels.get(repo, {}).get(name) + + # -- issues ------------------------------------------------------------ + + def create_issue(self, repo: str, title: str, body: str, + labels: list | None = None) -> int: + rec = self.repos[repo] + number = len(rec["issues"]) + 1 + label_ids = [] + for name in labels or []: + lid = self.label_id(repo, name) + if lid is not None and lid not in label_ids: + label_ids.append(lid) + rec["issues"].append({ + "number": number, "title": title, "body": body, "labels": label_ids, + }) + return number + + def issue(self, repo: str, number: int) -> dict: + return self.repos[repo]["issues"][number - 1] + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture() +def env(tmp_path: Path, monkeypatch): + core = ReferenceCore(tmp_path / "billing.db") + gitea = MockGitea() + cfg = types.SimpleNamespace(db_path=tmp_path / "billing.db", + api_key="test-api-key", + debit_floor_minor=DEFAULT_DEBIT_FLOOR_MINOR) + factory = IntakeFactory(core, gitea, cfg) + return types.SimpleNamespace(core=core, gitea=gitea, cfg=cfg, factory=factory, + db_path=tmp_path / "billing.db") + + +def _client(env, name: str = "Acme Corp", amount_minor: int = 0) -> int: + ok, details = env.core.create_client(name) + assert ok, details + client_id = details["client_id"] + if amount_minor: + ok, _ = env.core.credit(client_id, amount_minor) + assert ok + return client_id + + +def _queued_rows(db_path: Path, client_id: int) -> list: + conn = sqlite3.connect(str(db_path)) + try: + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT * FROM pending_issues WHERE client_id = ? ORDER BY id", + (client_id,)).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +def _mapping_rows(db_path: Path, client_id: int) -> list: + conn = sqlite3.connect(str(db_path)) + try: + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT * FROM client_repos WHERE client_id = ?", (client_id,)).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +# --------------------------------------------------------------------------- +# 1. provision: creates once, reuses on second call (idempotent) +# --------------------------------------------------------------------------- + +def test_provision_creates_repo_once_and_reuses(env): + cid = _client(env, "Acme Corp") + repo = env.factory.provision_client(cid) + assert repo == "acme-corp" + assert env.gitea.repo_exists("acme-corp") + assert env.gitea.created_repo_calls == ["acme-corp"] + + # Second provision: same repo, no duplicate creation, one mapping row. + again = env.factory.provision_client(cid) + assert again == "acme-corp" + assert env.gitea.created_repo_calls == ["acme-corp"] + assert len(_mapping_rows(env.db_path, cid)) == 1 + + +def test_provision_explicit_repo_name_and_cross_client_isolation(env): + cid = _client(env, "Globex") + repo = env.factory.provision_client(cid, "globex-main") + assert repo == "globex-main" + assert env.gitea.repo_exists("globex-main") + # A different client gets its own mapping and repo. + cid2 = _client(env, "Initech") + repo2 = env.factory.provision_client(cid2) + assert repo2 == "initech" + assert len(env.gitea.created_repo_calls) == 2 + assert len(_mapping_rows(env.db_path, cid)) == 1 + assert len(_mapping_rows(env.db_path, cid2)) == 1 + + +def test_provision_rejects_bad_repo_name_and_unknown_client(env): + cid = _client(env) + with pytest.raises(Exception): + env.factory.provision_client(cid, "bad name with spaces") + with pytest.raises(UnknownClientError): + env.factory.provision_client(9999, "whatever-repo") + + +# --------------------------------------------------------------------------- +# 2. funded submit: Gitea issue immediately, delivered=true +# --------------------------------------------------------------------------- + +def test_funded_submit_delivers_immediately(env): + cid = _client(env, "Acme Corp", amount_minor=10_000) + env.factory.provision_client(cid) + env.gitea.add_label("acme-corp", "p2") + result = env.factory.submit_issue( + cid, "acme-corp", "Fix the bug", + "## Acceptance\n- [ ] it works", labels=["p2"]) + assert result == {"delivered": True, "issue_number": 1} + issue = env.gitea.issue("acme-corp", 1) + assert issue["title"] == "Fix the bug" + assert issue["body"] == "## Acceptance\n- [ ] it works" + assert issue["labels"] == [env.gitea.label_id("acme-corp", "p2")] + # Nothing queued for a funded direct delivery. + assert _queued_rows(env.db_path, cid) == [] + + +def test_funded_submit_without_labels(env): + cid = _client(env, "Acme Corp", amount_minor=10_000) + env.factory.provision_client(cid) + result = env.factory.submit_issue(cid, "acme-corp", "No labels", "") + assert result["delivered"] is True + assert env.gitea.issue("acme-corp", 1)["labels"] == [] + + +# --------------------------------------------------------------------------- +# 3. unfunded submit: queued, no Gitea issue, delivered=false +# --------------------------------------------------------------------------- + +def test_unfunded_submit_queues(env): + cid = _client(env, "Acme Corp", amount_minor=0) + env.factory.provision_client(cid) + result = env.factory.submit_issue( + cid, "acme-corp", "Queued work", "## Acceptance\n- [ ] later") + assert result["delivered"] is False + assert result["reason"] == "unfunded" + assert isinstance(result["pending_id"], int) + assert env.gitea.repos["acme-corp"]["issues"] == [] + rows = _queued_rows(env.db_path, cid) + assert len(rows) == 1 + row = rows[0] + assert row["status"] == "queued" + assert row["title"] == "Queued work" + assert json.loads(row["body_json"]) == "## Acceptance\n- [ ] later" + assert json.loads(row["labels_json"]) == [] + + +# --------------------------------------------------------------------------- +# 4. unlock: flush queued -> Gitea with p1 label, shipped, second no-op +# --------------------------------------------------------------------------- + +def test_unlock_flushes_queued_with_p1_and_is_idempotent(env): + cid = _client(env, "Acme Corp", amount_minor=0) + env.factory.provision_client(cid) + env.gitea.add_label("acme-corp", "p3") + r1 = env.factory.submit_issue(cid, "acme-corp", "Task A", "body A", labels=["p3"]) + r2 = env.factory.submit_issue(cid, "acme-corp", "Task B", "body B") + assert r1["delivered"] is False and r2["delivered"] is False + + flushed = env.factory.unlock(cid) + assert flushed["flushed"] == 2 + assert flushed["issues"] == [1, 2] + p1 = env.gitea.label_id("acme-corp", "p1") + p3 = env.gitea.label_id("acme-corp", "p3") + a = env.gitea.issue("acme-corp", 1) + b = env.gitea.issue("acme-corp", 2) + assert a["title"] == "Task A" and a["body"] == "body A" + assert p1 in a["labels"] and p3 in a["labels"] + assert b["labels"] == [p1] + # Rows marked shipped. + rows = _queued_rows(env.db_path, cid) + assert len(rows) == 2 and all(r["status"] == "shipped" for r in rows) + + # Second unlock: nothing queued -> no-op. + again = env.factory.unlock(cid) + assert again == {"flushed": 0, "issues": [], "lock_cleared": False} + assert len(env.gitea.repos["acme-corp"]["issues"]) == 2 + + +# --------------------------------------------------------------------------- +# 5. lock: future submits queue even when funded; unlock clears + flushes +# --------------------------------------------------------------------------- + +def test_lock_queues_funded_submits_and_unlock_resumes(env): + cid = _client(env, "Acme Corp", amount_minor=10_000) + env.factory.provision_client(cid) + assert env.factory.lock(cid) == {"client_id": cid, "locked": True} + + # Funded but locked -> queued, no Gitea issue. + result = env.factory.submit_issue(cid, "acme-corp", "Blocked by lock", "") + assert result == {"delivered": False, "pending_id": 1, "reason": "locked"} + assert env.gitea.repos["acme-corp"]["issues"] == [] + + # unlock flushes the queued issue AND clears the lock. + flushed = env.factory.unlock(cid) + assert flushed["flushed"] == 1 and flushed["lock_cleared"] is True + assert env.gitea.repos["acme-corp"]["issues"][0]["title"] == "Blocked by lock" + + # Lock cleared -> shipping resumes. + again = env.factory.submit_issue(cid, "acme-corp", "Now shipping", "") + assert again == {"delivered": True, "issue_number": 2} + + # lock() twice is idempotent; lock() on unknown client raises. + assert env.factory.lock(cid)["locked"] is True + with pytest.raises(UnknownClientError): + env.factory.lock(9999) + + +# --------------------------------------------------------------------------- +# 6. client_unfunded: below the debit floor (and fail-closed) +# --------------------------------------------------------------------------- + +def test_client_unfunded_boundary(env): + cid = _client(env, "Acme Corp", amount_minor=0) + assert env.factory.client_unfunded(cid) is True + env.core.credit(cid, DEFAULT_DEBIT_FLOOR_MINOR - 1) # 499 + assert env.factory.client_unfunded(cid) is True + env.core.credit(cid, 1) # 500 == floor + assert env.factory.client_unfunded(cid) is False + env.core.credit(cid, 1000) # 1500 + assert env.factory.client_unfunded(cid) is False + + +def test_client_unfunded_fail_closed_on_unknown_client(env): + # Unknown client -> cannot confirm funding -> unfunded (fail-closed). + assert env.factory.client_unfunded(9999) is True + with pytest.raises(UnknownClientError): + env.factory.submit_issue(9999, "acme-corp", "Nope", "") + + +def test_floor_override_via_config(env): + cid = _client(env, "Acme Corp", amount_minor=200) + env.cfg.debit_floor_minor = 100 + assert env.factory.client_unfunded(cid) is False # 200 >= 100 + env.cfg.debit_floor_minor = 300 + assert env.factory.client_unfunded(cid) is True # 200 < 300 + + +# --------------------------------------------------------------------------- +# HTTP surface (register_factory_routes), same X-API-Key discipline +# --------------------------------------------------------------------------- + +@pytest.fixture() +def app_client(env): + from flask import Flask + + app = Flask(__name__) + app.config["API_KEY"] = env.cfg.api_key + register_factory_routes(app, env.core, env.gitea, env.cfg) + return app.test_client() + + +def _post(app, path, payload=None, key="test-api-key"): + headers = {"X-API-Key": key} if key else {} + return app.post(path, json=payload or {}, headers=headers) + + +def test_routes_require_api_key(app_client): + resp = app_client.post("/api/factory/provision", json={"client_id": 1}) + assert resp.status_code == 401 + resp = app_client.post("/api/factory/1/unlock") + assert resp.status_code == 401 + + +def test_route_provision_submit_unlock_lock_flow(env, app_client): + cid = _client(env, "Acme Corp", amount_minor=0) + resp = _post(app_client, "/api/factory/provision", {"client_id": cid}) + assert resp.status_code == 200 + assert resp.get_json() == {"repo": "acme-corp"} + + # Unfunded submit -> 202 queued. + resp = _post(app_client, f"/api/factory/{cid}/issues", + {"repo": "acme-corp", "title": "Queued via API", + "acceptance_body": "## Acceptance"}) + assert resp.status_code == 202 + body = resp.get_json() + assert body["delivered"] is False and body["reason"] == "unfunded" + + # Unlock -> 200 flushed with p1. + resp = _post(app_client, f"/api/factory/{cid}/unlock") + assert resp.status_code == 200 + assert resp.get_json()["flushed"] == 1 + + # Lock -> 200; funded submit afterwards still queues. + env.core.credit(cid, 10_000) + resp = _post(app_client, f"/api/factory/{cid}/lock") + assert resp.get_json() == {"client_id": cid, "locked": True} + resp = _post(app_client, f"/api/factory/{cid}/issues", + {"repo": "acme-corp", "title": "Locked via API"}) + assert resp.get_json()["reason"] == "locked" + + +def test_route_unknown_client_404(env, app_client): + resp = _post(app_client, "/api/factory/provision", {"client_id": 9999}) + assert resp.status_code == 404 + assert resp.get_json()["code"] == "unknown_client" + + +def test_route_validation_400(app_client): + resp = _post(app_client, "/api/factory/provision", {"client_id": -1}) + assert resp.status_code == 400 + + +# --------------------------------------------------------------------------- +# 7. Hermeticity guard +# --------------------------------------------------------------------------- + +def test_hermetic_no_real_gitea_or_network_used(env): + """The whole suite runs against mocks; a real GiteaClient would need a + token and a network. This test pins the factory's seams: every gitea + interaction goes through the injected mock object.""" + assert isinstance(env.gitea, MockGitea) + assert isinstance(env.factory._gitea, MockGitea) + + +def test_real_billing_core_composition(tmp_path: Path): + """The factory composes with the REAL billing core (issue #1, merged). + + Uses the actual ``gitea_saas.billing_core.BillingCore`` over a temp DB + -- still no network: Gitea stays mocked. + """ + from gitea_saas.billing_core import BillingCore + + core = BillingCore(tmp_path / "real.db") + gitea = MockGitea() + cfg = types.SimpleNamespace(db_path=tmp_path / "real.db", api_key="k", + debit_floor_minor=DEFAULT_DEBIT_FLOOR_MINOR) + factory = IntakeFactory(core, gitea, cfg) + + ok, details = core.create_client("Acme Corp") + assert ok + cid = details["id"] + assert core.credit(cid, 10_000, "idem-1")[0] is True + assert factory.provision_client(cid) == "acme-corp" + result = factory.submit_issue(cid, "acme-corp", "Funded with real core", "body") + assert result == {"delivered": True, "issue_number": 1} + + ok, details2 = core.create_client("Poor Corp") + assert ok + cid2 = details2["id"] + assert factory.provision_client(cid2) == "poor-corp" + queued = factory.submit_issue(cid2, "poor-corp", "Queued", "body") + assert queued["delivered"] is False and queued["reason"] == "unfunded" + + flushed = factory.unlock(cid2) + assert flushed["flushed"] == 1 + p1 = gitea.label_id("poor-corp", "p1") + assert p1 in gitea.issue("poor-corp", 1)["labels"]