pbwraith/gitea-saas-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into issue-2

Browse Source 
# Conflicts:
#	requirements.txt
This commit is contained in:
Claude 2026-08-25 06:29:19 -05:00
commit 4d3123af9e
9 changed files with 1284 additions and 396 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
ARCHITECTURE.md
View File

@ -8,9 +8,12 @@ is the delivery machine; this service is the client-facing moat around it.
- **Billing core** (shared): clients, ledger, holds, Stripe top-ups, credits. Pure Python + SQLite.
No knowledge of lanes or Gitea.
- **Factory front-end** (option 1): client repo provisioning, issue intake with acceptance
criteria, `unfunded` gating, delivery receipts on merge.
criteria, **funding gate at intake**, delivery receipts on merge.
- **Eval farm front-end** (option 2): batch task intake -> internal `eval-task` issues,
results table, emitter (webhook or file drop).
- **Review gate** (shared, THE quality moat): automated PR reviewer. Parses the linked
issue's acceptance criteria, runs tests/build/diff checks, posts APPROVE/BOUNCE verdicts,
auto-merges per client tier policy. Replaces the operator's manual review for client work.
## Billing invariants (fail-closed)
@ -18,15 +21,31 @@ is the delivery machine; this service is the client-facing moat around it.
- every mutation is idempotent via an `idempotency_key` (Stripe event id, intake uuid, etc.)
- holds expire (default 30 min) and release back automatically
- debit happens exactly once per delivered issue (key = gitea repo + issue number + merge sha)
- client with balance below the debit floor is **unfunded**: the dispatcher must never claim
its issues (gate = `unfunded` label skipped by the lane survey)
## Gating model (no dispatcher state)
## Gating model (intake-side, no dispatcher coupling)
The dispatcher scans all repos under the owner. Intake creates issues carrying the
`unfunded` label; payment removes it and applies the tier's priority label. The lane survey
already skips a known label set — `unfunded` joins it. Zero-balance re-locks by re-adding
`unfunded` on the next delivery check. No new dispatcher machinery, no shared DB reads.
A client's work NEVER becomes a Gitea issue while unfunded. `submit_issue` checks the
client balance: funded -> issue created immediately; unfunded -> row in `pending_issues`
queued locally. `unlock(client_id)` flushes the queue to Gitea with the tier's priority
label. The dispatcher scans all repos but only ever sees *funded* work — zero changes to
the lane/dispatcher scripts, no label-gating in the survey, no coupling between the
monetization service and deployment infrastructure.
## Review gate
`reviewer.py` polls PRs (or is webhook-driven) that carry `in-review`. For each:
1. Fetch linked issue, parse the acceptance-criteria checklist.
2. Checkout head in a worktree; run the repo's test command + build + `git diff --check`.
3. Optional secret scan (reuse worker-ops patterns).
4. LLM review pass against the acceptance criteria (files allowed per issue spec,
scope-creep check, correctness, edge cases) — evidence-quoted verdict.
5. Post verdict comment: APPROVE / BOUNCE (with findings) / NEEDS-GUIDANCE (cannot verify).
6. Policy per client tier (config): auto-merge on APPROVE+green for gold tier; manual
confirm for standard tier; BOUNCE routes back to the implementing lane.
The review gate applies to client repos; the operator's internal fleet can opt in per
repo. Bounced PRs re-enter the normal lane flow (existing `bounced` label machinery).
## Layout (target)
@ -35,9 +54,10 @@ already skips a known label set — `unfunded` joins it. Zero-balance re-locks b
server.py # Flask app (billing API + dashboards)
stripe_client.py # Stripe adapter (mockable seam, env keys)
gitea_client.py # Gitea API client (mockable seam, env token)
intake_factory.py # option 1: provision client repo + issues + unlock
intake_factory.py # option 1: provision client repo + funded issue creation
intake_eval.py # option 2: eval batches -> issues + results + emitter
delivery.py # merge event -> receipt debit + comment + relock check
reviewer.py # review gate: acceptance parse, tests, verdicts, policy
delivery.py # merge event -> receipt debit + comment + lock-check
config.py # tier config, rates (loaded from env/config, never hardcoded)
schema.sql / db.py # SQLite migration
tests/ # pytest, hermetic (mock Stripe + Gitea, tmp SQLite)
@ -46,11 +66,12 @@ already skips a known label set — `unfunded` joins it. Zero-balance re-locks b
## Environment (provided by operator at deploy time, mocked in CI)
- STRIPE_SECRET_KEY / STRIPE_WEBHOOK_SECRET (optional: absent -> mock mode)
- GITEA_BASE_URL / GITEA_TOKEN (factory/eval front-ends)
- GITEA_BASE_URL / GITEA_TOKEN (factory/eval/reviewer)
- SAAS_DB_PATH, SAAS_API_KEY (X-API-Key auth), SAAS_PORT
## Tier/rate config (operator sets later; centralize in config.py)
Enterprise default: priority p1, auto-merge on green (opt-in per client), rate per issue.
Rates are config, not code. First client onboarding is manual; self-serve (Stripe checkout
-> auto-provision) is the stated next step after core + both front-ends land.
Gold: priority p1, auto-merge on reviewer APPROVE+green. Standard: p2, manual merge
confirm. Rates per issue, config not code. First client onboarding is manual; self-serve
(Stripe checkout -> auto-provision) is the stated next step after core + both front-ends
land.

534
gitea_saas/billing_core.py Normal file
View File

@ -0,0 +1,534 @@
"""Billing ledger core for gitea-saas-core.
Pure Python + SQLite (stdlib only). No HTTP, no Stripe, no Gitea. This is the
shared billing core everything else builds on.
Invariants (fail-closed)
-----------------------
* ``balance = sum(credits) + sum(refunds) - sum(debits) - sum(active holds)``
and may never go negative: any mutation that would push it below zero is
rejected with a structured error and writes NO rows.
* Every mutating operation is idempotent via ``idempotency_key`` (UNIQUE in
the ledger). Replaying a key that already exists is a no-op that returns
the original entry -- never an error.
* Holds expire (default 30 minutes) and are auto-released on the next
read/query, so a stale hold can never lock balance forever.
Public API contract
-------------------
Every operation returns ``(ok, details)`` -- a bool and a dict. Failures are
structured and never raise: ``(False, {"error": {"code": ..., "message":
...}})``. Success payloads carry ``ledger_id``/``hold_id`` and the client's
``balance`` after the mutation.
Ledger audit convention for holds: the ``hold``/``release`` ledger rows use
``ref_type='hold', ref_id=str(hold_id)`` (a system pointer) so idempotent
replays resolve the exact hold row. The caller-facing semantic ref supplied
to ``reserve_hold`` lives on the holds row itself.
"""
from __future__ import annotations
import json
import sqlite3
import threading
from datetime import datetime, timedelta, timezone
from pathlib import Path
from .db import SCHEMA_PATH, connect, create_tables
# Structured error codes (public contract).
UNKNOWN_CLIENT = "UNKNOWN_CLIENT"
INVALID_AMOUNT = "INVALID_AMOUNT"
INVALID_KEY = "INVALID_KEY"
INVALID_CURRENCY = "INVALID_CURRENCY"
INVALID_NAME = "INVALID_NAME"
INVALID_HOLD_TTL = "INVALID_HOLD_TTL"
INSUFFICIENT_BALANCE = "INSUFFICIENT_BALANCE"
DUPLICATE_CLIENT = "DUPLICATE_CLIENT"
HOLD_NOT_FOUND = "HOLD_NOT_FOUND"
HOLD_ALREADY_RELEASED = "HOLD_ALREADY_RELEASED"
IDEMPOTENCY_CONFLICT = "IDEMPOTENCY_CONFLICT"
INTERNAL = "INTERNAL"
KINDS = frozenset({"credit", "debit", "hold", "release", "refund"})
LEDGER_CREDIT_KINDS = ("credit", "refund") # money-in kinds counted in balance
DEFAULT_HOLD_MINUTES = 30
DEFAULT_CURRENCY = "USD"
def _ok(details: dict) -> tuple[bool, dict]:
return True, details
def _err(code: str, message: str) -> tuple[bool, dict]:
return False, {"error": {"code": code, "message": message}}
def _coerce_ts(value) -> str:
"""Accept an ISO string or a (possibly naive) datetime -> ISO string."""
if isinstance(value, datetime):
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.isoformat()
return str(value)
def _load_meta(raw: str | None) -> dict:
try:
parsed = json.loads(raw or "{}")
return parsed if isinstance(parsed, dict) else {}
except (ValueError, TypeError):
return {}
class BillingCore:
"""Thread-safe billing ledger backed by a single SQLite file.
``db_path`` is a ``pathlib.Path`` (tests use temp-dir DBs). All access is
serialized by one re-entrant lock; WAL keeps readers non-blocking.
"""
def __init__(self, db_path: Path, schema_path: Path | None = None,
now_fn=None):
"""Open (or create) the ledger DB.
``db_path`` is a ``pathlib.Path`` (tests use temp-dir DBs).
``now_fn`` is a test seam: a zero-arg callable returning the current
UTC ``datetime``; defaults to the real clock. All access is
serialized by one re-entrant lock; WAL keeps readers non-blocking.
"""
self._path = Path(db_path)
self._schema_path = Path(schema_path) if schema_path is not None else SCHEMA_PATH
self._now_fn = now_fn or (lambda: datetime.now(timezone.utc))
self._lock = threading.RLock()
self._conn = connect(self._path)
self.create_tables()
# ------------------------------------------------------------------
# Time helpers (injectable clock)
# ------------------------------------------------------------------
def _ts(self) -> str:
"""Current time as a sortable ISO-8601 UTC string."""
return self._now_fn().isoformat()
def _ts_plus_minutes(self, minutes: float) -> str:
"""ISO-8601 UTC timestamp ``minutes`` from now (negative => the past)."""
return (self._now_fn() + timedelta(minutes=minutes)).isoformat()
# ------------------------------------------------------------------
# Schema / low-level helpers
# ------------------------------------------------------------------
def create_tables(self) -> None:
"""Create the schema idempotently (safe to call any number of times)."""
with self._lock:
create_tables(self._conn, schema_path=self._schema_path)
def _client_exists(self, client_id: int) -> bool:
row = self._conn.execute("SELECT id FROM clients WHERE id = ?", (client_id,)).fetchone()
return row is not None
def _active_hold_sum(self, client_id: int) -> int:
row = self._conn.execute(
"SELECT COALESCE(SUM(amount_minor), 0) FROM holds"
" WHERE client_id = ? AND released_at IS NULL AND expires_at > ?",
(client_id, self._ts()),
).fetchone()
return int(row[0])
def _ledger_sum(self, client_id: int, kinds: tuple[str, ...]) -> int:
placeholders = ",".join("?" * len(kinds))
row = self._conn.execute(
f"SELECT COALESCE(SUM(amount_minor), 0) FROM ledger"
f" WHERE client_id = ? AND kind IN ({placeholders})",
(client_id, *kinds),
).fetchone()
return int(row[0])
def _balance(self, client_id: int) -> int:
"""Available balance: credits + refunds - debits - active holds."""
money_in = self._ledger_sum(client_id, LEDGER_CREDIT_KINDS)
debits = self._ledger_sum(client_id, ("debit",))
holds = self._active_hold_sum(client_id)
return money_in - debits - holds
def _expire_stale_holds(self, client_id: int | None = None) -> list[dict]:
"""Release expired holds, writing deterministic 'release' ledger rows.
Idempotent: only ``released_at IS NULL`` holds are touched, and the
auto-release ledger key is derived from the hold id, so re-running is
a no-op.
"""
now = self._ts()
if client_id is None:
rows = self._conn.execute(
"SELECT * FROM holds WHERE released_at IS NULL AND expires_at <= ?",
(now,),
).fetchall()
else:
rows = self._conn.execute(
"SELECT * FROM holds WHERE client_id = ? AND released_at IS NULL AND expires_at <= ?",
(client_id, now),
).fetchall()
expired = []
for h in rows:
hold_id = h["id"]
cur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, 'release', ?, ?, ?, 'hold', ?, ?)",
(h["client_id"], h["amount_minor"], DEFAULT_CURRENCY,
f"auto-release:{hold_id}", str(hold_id), now),
)
self._conn.execute("UPDATE holds SET released_at = ? WHERE id = ?", (now, hold_id))
expired.append({"hold_id": hold_id, "amount_minor": h["amount_minor"],
"ledger_id": cur.lastrowid})
return expired
# ------------------------------------------------------------------
# Transaction wrapper (fail-closed)
# ------------------------------------------------------------------
def _mutate(self, fn) -> tuple[bool, dict]:
"""Run ``fn`` inside BEGIN IMMEDIATE with structured error isolation.
Writes are atomic: on any SQLite failure the whole transaction is
rolled back and a structured INTERNAL error is returned -- no partial
rows, and no exceptions leak to callers.
"""
with self._lock:
try:
self._conn.execute("BEGIN IMMEDIATE")
except sqlite3.Error as e:
return _err(INTERNAL, f"could not start transaction: {e}")
try:
result = fn()
except sqlite3.Error as e:
try:
self._conn.rollback()
except sqlite3.Error:
pass
return _err(INTERNAL, f"database error: {e}")
try:
self._conn.commit()
except sqlite3.Error as e:
try:
self._conn.rollback()
except sqlite3.Error:
pass
return _err(INTERNAL, f"could not commit transaction: {e}")
return result
@staticmethod
def _validate_amount(amount_minor) -> str | None:
"""Return an error code if the amount is invalid, else None."""
if not isinstance(amount_minor, int) or isinstance(amount_minor, bool) or amount_minor <= 0:
return INVALID_AMOUNT
return None
@staticmethod
def _validate_key(idempotency_key) -> str | None:
if not isinstance(idempotency_key, str) or not idempotency_key.strip():
return INVALID_KEY
return None
@staticmethod
def _validate_currency(currency) -> str | None:
if not isinstance(currency, str) or not currency.strip():
return INVALID_CURRENCY
return None
# ------------------------------------------------------------------
# Clients
# ------------------------------------------------------------------
def create_client(self, name: str, tier: str = "free", status: str = "active",
meta: dict | None = None) -> tuple[bool, dict]:
"""Create a client. Duplicate names are rejected (UNIQUE)."""
name = (name or "").strip()
if not name:
return _err(INVALID_NAME, "client name must be a non-empty string")
meta = meta if isinstance(meta, dict) else {}
meta_json = json.dumps(meta, sort_keys=True)
def fn() -> tuple[bool, dict]:
now = self._ts()
try:
cur = self._conn.execute(
"INSERT INTO clients (name, tier, status, meta_json, created_at)"
" VALUES (?, ?, ?, ?, ?)",
(name, tier, status, meta_json, now),
)
except sqlite3.IntegrityError:
return _err(DUPLICATE_CLIENT, f"client name already exists: {name}")
return _ok({"client_id": cur.lastrowid, "name": name, "tier": tier,
"status": status, "meta": meta, "created_at": now})
return self._mutate(fn)
def get_client(self, client_id: int, include_balance: bool = True) -> tuple[bool, dict]:
"""Fetch a client. Stale holds auto-release before the balance is read."""
def fn() -> tuple[bool, dict]:
row = self._conn.execute("SELECT * FROM clients WHERE id = ?", (client_id,)).fetchone()
if row is None:
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
client = dict(row)
client["meta"] = _load_meta(client.pop("meta_json"))
self._expire_stale_holds(client_id)
details: dict = {"client": client}
if include_balance:
details["balance"] = self._balance(client_id)
return _ok(details)
return self._mutate(fn)
def get_balance(self, client_id: int) -> tuple[bool, dict]:
"""Available balance: credits + refunds - debits - active holds."""
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
return _ok({"client_id": client_id, "balance": self._balance(client_id)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Ledger mutations (credit / debit / refund)
# ------------------------------------------------------------------
def credit(self, client_id: int, amount_minor: int, idempotency_key: str,
currency: str = DEFAULT_CURRENCY, ref_type: str | None = None,
ref_id: str | None = None) -> tuple[bool, dict]:
"""Add money. Replaying the same ``idempotency_key`` is a no-op."""
return self._apply_ledger(client_id, "credit", amount_minor, idempotency_key,
currency, ref_type, ref_id)
def debit(self, client_id: int, amount_minor: int, idempotency_key: str,
currency: str = DEFAULT_CURRENCY, ref_type: str | None = None,
ref_id: str | None = None) -> tuple[bool, dict]:
"""Remove money. Rejected with INSUFFICIENT_BALANCE (no rows written)
if it would push the balance below zero."""
return self._apply_ledger(client_id, "debit", amount_minor, idempotency_key,
currency, ref_type, ref_id)
def refund(self, client_id: int, amount_minor: int, idempotency_key: str,
currency: str = DEFAULT_CURRENCY, ref_type: str | None = None,
ref_id: str | None = None) -> tuple[bool, dict]:
"""Add money back (audited as 'refund'). Preserves idempotency."""
return self._apply_ledger(client_id, "refund", amount_minor, idempotency_key,
currency, ref_type, ref_id)
def _apply_ledger(self, client_id: int, kind: str, amount_minor: int,
idempotency_key: str, currency: str, ref_type: str | None,
ref_id: str | None) -> tuple[bool, dict]:
if kind not in KINDS or kind in ("hold", "release"):
return _err(INTERNAL, f"invalid ledger kind: {kind}")
if (code := self._validate_amount(amount_minor)) is not None:
return _err(code, "amount_minor must be a positive integer (minor units, e.g. cents)")
if (code := self._validate_key(idempotency_key)) is not None:
return _err(code, "idempotency_key must be a non-empty string")
if (code := self._validate_currency(currency)) is not None:
return _err(code, "currency must be a non-empty string (default USD)")
currency = currency.strip().upper()
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
existing = self._conn.execute(
"SELECT * FROM ledger WHERE idempotency_key = ?", (idempotency_key,)
).fetchone()
if existing is not None:
return self._replay(existing, client_id)
balance = self._balance(client_id)
if kind == "debit" and amount_minor > balance:
return _err(INSUFFICIENT_BALANCE,
f"debit {amount_minor} exceeds available balance {balance}")
cur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(client_id, kind, amount_minor, currency, idempotency_key,
ref_type, ref_id, self._ts()),
)
return _ok({"ledger_id": cur.lastrowid, "kind": kind, "client_id": client_id,
"amount_minor": amount_minor, "currency": currency,
"ref_type": ref_type, "ref_id": ref_id,
"balance": self._balance(client_id)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Holds
# ------------------------------------------------------------------
def reserve_hold(self, client_id: int, amount_minor: int, idempotency_key: str,
ref_type: str | None = None, ref_id: str | None = None,
expires_in_minutes: float = DEFAULT_HOLD_MINUTES,
currency: str = DEFAULT_CURRENCY) -> tuple[bool, dict]:
"""Reserve balance with a hold. Replaying the same key is a no-op.
``expires_in_minutes`` defaults to 30; <= 0 means already expired
(deterministic expiry for tests / backdated holds).
"""
if (code := self._validate_amount(amount_minor)) is not None:
return _err(code, "amount_minor must be a positive integer (minor units, e.g. cents)")
if (code := self._validate_key(idempotency_key)) is not None:
return _err(code, "idempotency_key must be a non-empty string")
if (code := self._validate_currency(currency)) is not None:
return _err(code, "currency must be a non-empty string (default USD)")
if not isinstance(expires_in_minutes, (int, float)) or isinstance(expires_in_minutes, bool):
return _err(INVALID_HOLD_TTL, "expires_in_minutes must be a number")
currency = currency.strip().upper()
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
existing = self._conn.execute(
"SELECT * FROM ledger WHERE idempotency_key = ?", (idempotency_key,)
).fetchone()
if existing is not None:
return self._replay(existing, client_id)
balance = self._balance(client_id)
if amount_minor > balance:
return _err(INSUFFICIENT_BALANCE,
f"hold {amount_minor} exceeds available balance {balance}")
expires_at = self._ts_plus_minutes(float(expires_in_minutes))
cur = self._conn.execute(
"INSERT INTO holds (client_id, amount_minor, ref_type, ref_id, expires_at)"
" VALUES (?, ?, ?, ?, ?)",
(client_id, amount_minor, ref_type, ref_id, expires_at),
)
hold_id = cur.lastrowid
# System pointer so idempotent replays resolve the exact hold row.
lcur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, 'hold', ?, ?, ?, 'hold', ?, ?)",
(client_id, amount_minor, currency, idempotency_key, str(hold_id), self._ts()),
)
return _ok({"hold_id": hold_id, "client_id": client_id,
"amount_minor": amount_minor, "currency": currency,
"ref_type": ref_type, "ref_id": ref_id,
"expires_at": expires_at, "ledger_id": lcur.lastrowid,
"balance": self._balance(client_id)})
return self._mutate(fn)
def release_hold(self, hold_id: int, idempotency_key: str) -> tuple[bool, dict]:
"""Release a hold, returning its amount to the balance.
Replaying the same ``idempotency_key`` returns the original release
(no-op). Releasing an already-released hold with a *different* key is
a structured HOLD_ALREADY_RELEASED error.
"""
if (code := self._validate_key(idempotency_key)) is not None:
return _err(code, "idempotency_key must be a non-empty string")
def fn() -> tuple[bool, dict]:
hold = self._conn.execute("SELECT * FROM holds WHERE id = ?", (hold_id,)).fetchone()
if hold is None:
return _err(HOLD_NOT_FOUND, f"no hold with id {hold_id}")
client_id = hold["client_id"]
existing = self._conn.execute(
"SELECT * FROM ledger WHERE idempotency_key = ?", (idempotency_key,)
).fetchone()
if existing is not None:
return self._replay(existing, client_id)
if hold["released_at"] is not None:
return _err(HOLD_ALREADY_RELEASED,
f"hold {hold_id} already released at {hold['released_at']}")
# The hold may be stale: expire it first, then re-check state.
self._expire_stale_holds(client_id)
hold = self._conn.execute("SELECT * FROM holds WHERE id = ?", (hold_id,)).fetchone()
if hold["released_at"] is not None:
return _err(HOLD_ALREADY_RELEASED,
f"hold {hold_id} auto-released on expiry at {hold['released_at']}")
now = self._ts()
self._conn.execute("UPDATE holds SET released_at = ? WHERE id = ?", (now, hold_id))
lcur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, 'release', ?, ?, ?, 'hold', ?, ?)",
(client_id, hold["amount_minor"], DEFAULT_CURRENCY,
idempotency_key, str(hold_id), now),
)
return _ok({"hold_id": hold_id, "client_id": client_id,
"amount_minor": hold["amount_minor"],
"released_at": now, "ledger_id": lcur.lastrowid,
"balance": self._balance(client_id)})
return self._mutate(fn)
def expire_stale_holds(self, client_id: int | None = None) -> tuple[bool, dict]:
"""Explicitly release every expired hold (optionally for one client)."""
def fn() -> tuple[bool, dict]:
if client_id is not None and not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
expired = self._expire_stale_holds(client_id)
return _ok({"expired": expired, "count": len(expired)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Queries
# ------------------------------------------------------------------
def list_ledger(self, client_id: int, from_=None, to=None) -> tuple[bool, dict]:
"""Ledger entries for a client, newest-compatible ordering.
``from_``/``to`` bound ``created_at`` (ISO string or datetime).
Stale holds auto-release before the read.
"""
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
sql = "SELECT * FROM ledger WHERE client_id = ?"
params: list = [client_id]
if from_ is not None:
sql += " AND created_at >= ?"
params.append(_coerce_ts(from_))
if to is not None:
sql += " AND created_at <= ?"
params.append(_coerce_ts(to))
sql += " ORDER BY created_at, id"
rows = self._conn.execute(sql, params).fetchall()
entries = [dict(r) for r in rows]
return _ok({"entries": entries, "count": len(entries)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Idempotent replay
# ------------------------------------------------------------------
def _replay(self, row, client_id: int) -> tuple[bool, dict]:
"""Return the original result for a replayed idempotency key."""
if row["client_id"] != client_id:
return _err(IDEMPOTENCY_CONFLICT,
f"idempotency_key already used for client {row['client_id']}")
original = {"ledger_id": row["id"], "kind": row["kind"], "client_id": row["client_id"],
"amount_minor": row["amount_minor"], "currency": row["currency"],
"ref_type": row["ref_type"], "ref_id": row["ref_id"],
"created_at": row["created_at"]}
details: dict = {"replayed": True, "original": original,
"client_id": client_id, "balance": self._balance(client_id)}
if row["kind"] in ("hold", "release") and row["ref_id"] is not None:
try:
hold = self._conn.execute(
"SELECT * FROM holds WHERE id = ?", (int(row["ref_id"]),)
).fetchone()
except (ValueError, TypeError):
hold = None
if hold is not None:
details["hold_id"] = hold["id"]
if row["kind"] == "release":
details["released_at"] = hold["released_at"]
return _ok(details)

37
gitea_saas/db.py Normal file
View File

@ -0,0 +1,37 @@
"""SQLite connection + schema bootstrap for gitea-saas-core.
Pure stdlib (sqlite3, pathlib). No HTTP, no Stripe, no Gitea.
"""
from __future__ import annotations
import sqlite3
from pathlib import Path
SCHEMA_PATH = Path(__file__).resolve().parent / "schema.sql"
def connect(db_path: Path) -> sqlite3.Connection:
"""Open a SQLite connection tuned for the billing core.
- WAL journal: readers never block writers (writes are additionally
serialized by BillingCore's single lock).
- Foreign keys enforced.
- ``check_same_thread=False``: BillingCore serializes all access with
one lock, so a single long-lived connection is safe across threads.
"""
path = Path(db_path)
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(path), check_same_thread=False)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA busy_timeout=5000")
return conn
def create_tables(conn: sqlite3.Connection, schema_path: Path | None = None) -> None:
"""Apply schema.sql idempotently (safe to call any number of times)."""
path = Path(schema_path) if schema_path is not None else SCHEMA_PATH
with open(path, encoding="utf-8") as f:
conn.executescript(f.read())

49
gitea_saas/schema.sql Normal file
View File

@ -0,0 +1,49 @@
-- gitea-saas-core billing schema (SQLite).
--
-- Idempotent by construction: every object is created with IF NOT EXISTS,
-- so executing this file any number of times is safe.
--
-- Conventions:
-- * All amounts are INTEGER minor units (cents). No floats, ever.
-- * Currency defaults to USD.
-- * Timestamps are ISO-8601 UTC text written by the application layer;
-- the DEFAULT expressions are fallbacks for manual/one-off inserts only.
-- * ledger.kind is constrained to {credit, debit, hold, release, refund}.
-- * ledger.idempotency_key is globally UNIQUE: every mutation is replay-safe.
CREATE TABLE IF NOT EXISTS clients (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
tier TEXT NOT NULL DEFAULT 'free',
status TEXT NOT NULL DEFAULT 'active',
meta_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
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 CHECK (amount_minor >= 0),
currency TEXT NOT NULL DEFAULT 'USD',
idempotency_key TEXT NOT NULL UNIQUE,
ref_type TEXT,
ref_id TEXT,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE TABLE IF NOT EXISTS holds (
id INTEGER PRIMARY KEY AUTOINCREMENT,
client_id INTEGER NOT NULL REFERENCES clients(id),
amount_minor INTEGER NOT NULL CHECK (amount_minor >= 0),
ref_type TEXT,
ref_id TEXT,
expires_at TEXT NOT NULL,
released_at TEXT
);
-- Balance / listing lookups.
CREATE INDEX IF NOT EXISTS idx_ledger_client_created ON ledger (client_id, created_at, id);
CREATE INDEX IF NOT EXISTS idx_ledger_client_kind ON ledger (client_id, kind);
CREATE INDEX IF NOT EXISTS idx_holds_client ON holds (client_id);
CREATE INDEX IF NOT EXISTS idx_holds_stale ON holds (client_id) WHERE released_at IS NULL;

148
gitea_saas/server.py
View File

@ -2,26 +2,39 @@
Consumed core interface (issue #1, ``gitea_saas/billing_core.py``):
All operations return ``(ok, details)`` and never raise for domain
Every operation returns ``(ok, details)`` and never raises for domain
failures; ``ok`` is bool and ``details`` is a dict. Domain errors carry
``details = {"code": "...", "error": "..."}`` with codes such as
``not_found``, ``negative_balance``, ``insufficient_funds``.
``details = {"error": {"code": "...", "message": "..."}}`` with codes
such as ``UNKNOWN_CLIENT``, ``INSUFFICIENT_BALANCE``, ``INVALID_AMOUNT``,
``HOLD_NOT_FOUND``, ``INTERNAL`` (see ``billing_core`` for the full set).
create_client(name, tier=..., meta=...) -> client record (incl. balance)
get_client(client_id, include_balance=True) -> client record
create_client(name, tier="free", status="active", meta=None)
-> {"client_id", "name", "tier",
"status", "meta", "created_at"}
get_client(client_id, include_balance=True)
-> {"client": {...}, "balance": N}
credit(client_id, amount_minor, idempotency_key,
ref_type=None, ref_id=None) -> {"client_id", "balance", ...}
debit(client_id, amount_minor, idempotency_key,
ref_type=None, ref_id=None) -> {"client_id", "balance", ...}
reserve_hold(client_id, amount_minor, ref_type=None,
ref_id=None, ttl_s=None) -> {"hold_id", "amount_minor", ...}
release_hold(hold_id) -> {"hold_id", "released": True, ...}
list_clients() -> {"clients": [client record, ...]}
list_ledger(client_id, from_=None, to=None) -> {"rows": [ledger row, ...]}
currency="USD", ref_type=None, ref_id=None)
-> {"ledger_id", "client_id",
"balance", ...}
debit(...) -> same shape as credit
reserve_hold(client_id, amount_minor, idempotency_key,
ref_type=None, ref_id=None, expires_in_minutes=30)
-> {"hold_id", "client_id",
"amount_minor", "balance", ...}
release_hold(hold_id, idempotency_key)
-> {"hold_id", "released_at",
"balance", ...}
list_ledger(client_id, from_=None, to=None)
-> {"entries": [ledger row, ...],
"count": N}
No ``list_clients`` exists on the core (issue #1 did not land one), so this
API has no list-all-clients endpoint.
Duplicate ``idempotency_key`` mutations are no-ops that return the original
result with ``ok=True`` (never an error). Balances and amounts are integers
in minor units.
result with ``ok=True`` and ``replayed=True`` (never an error). Balances and
amounts are integers in minor units.
The factory is deliberately pure HTTP: the core is injected, which keeps the
tests hermetic (Flask test client + temp DB) and free of any network/Stripe.
@ -31,6 +44,7 @@ from __future__ import annotations
import hmac
import sys
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from flask import Flask, jsonify, request
@ -88,15 +102,40 @@ def _call(app: Flask, fn: str, **kwargs) -> Dict[str, Any]:
def _map_core_error(fn: str, details: Any) -> ApiError:
code = details.get("code") if isinstance(details, dict) else None
message = details.get("error") if isinstance(details, dict) else str(details)
"""Map a real-core error payload to an HTTP error.
The real ``BillingCore`` reports failures as
``(False, {"error": {"code": ..., "message": ...}})`` with codes such as
``UNKNOWN_CLIENT`` / ``INSUFFICIENT_BALANCE`` / ``INVALID_AMOUNT`` /
``HOLD_NOT_FOUND``. Translate those to HTTP statuses while keeping the
response body a plain ``{"error", "code"}`` pair.
"""
err = details.get("error") if isinstance(details, dict) else None
if isinstance(err, dict):
code = err.get("code")
message = err.get("message")
else:
# Defensive: tolerate the older flat {code, error} shape.
code = details.get("code") if isinstance(details, dict) else None
message = details.get("error") if isinstance(details, dict) else str(details)
message = message or f"{fn} failed"
if code in ("not_found", "unknown_client", "unknown_hold"):
return ApiError(404, code, message)
if code in ("negative_balance", "insufficient_funds", "insufficient"):
return ApiError(409, code, message)
if code in ("validation", "invalid", "invalid_request"):
return ApiError(400, code, message)
status_map = {
"UNKNOWN_CLIENT": 404,
"HOLD_NOT_FOUND": 404,
"INSUFFICIENT_BALANCE": 409,
"DUPLICATE_CLIENT": 409,
"HOLD_ALREADY_RELEASED": 409,
"IDEMPOTENCY_CONFLICT": 409,
"INVALID_AMOUNT": 400,
"INVALID_KEY": 400,
"INVALID_CURRENCY": 400,
"INVALID_NAME": 400,
"INVALID_HOLD_TTL": 400,
"INTERNAL": 500,
}
if code in status_map:
return ApiError(status_map[code], code.lower(), message)
# Unknown structured errors are still surfaced (fail-closed), not 500'd.
if code:
return ApiError(400, code, message)
@ -142,10 +181,31 @@ def _json_body() -> Dict[str, Any]:
def _client_out(client: Dict[str, Any]) -> Dict[str, Any]:
"""Normalise a core client record for the wire (defensive)."""
allowed = ("id", "name", "tier", "status", "meta", "created_at", "balance")
return {k: client.get(k) for k in allowed if k in client} if isinstance(
client, dict) else client
"""Normalise a real-core client payload for the wire.
The real ``BillingCore`` returns two shapes:
* flat from ``create_client``: ``{client_id, name, tier, status,
meta, created_at}``
* nested from ``get_client``: ``{"client": {...}, "balance": N}``
This handles both and always exposes the API's ``id`` field (the core
calls it ``client_id``) plus the balance when present. Unknown keys are
passed through rather than dropped, so a contract mismatch becomes a
visible test failure instead of silent data loss.
"""
if not isinstance(client, dict):
return client
# Unwrap the nested get_client shape.
inner = client.get("client")
if isinstance(inner, dict):
out = dict(inner)
if "balance" in client:
out["balance"] = client["balance"]
client = out
out = dict(client)
if "client_id" in out and "id" not in out:
out["id"] = out.pop("client_id")
return out
# --------------------------------------------------------------------------
@ -199,17 +259,15 @@ def create_app(core: Any, api_key: str) -> Flask:
if meta is not None:
kwargs["meta"] = meta
created = _call(app, "create_client", **kwargs)
resp = jsonify(_client_out(created))
client_id = created.get("client_id") or created.get("id")
# Re-fetch through the real core so the response carries the API's
# `id` field plus the current balance (the real create_client does
# not return a balance).
client = _call(app, "get_client", client_id=client_id, include_balance=True)
resp = jsonify(_client_out(client))
resp.status_code = 201
return resp
@app.get("/api/clients")
def list_clients():
mapped = _call(app, "list_clients")
clients = mapped.get("clients", []) if isinstance(mapped, dict) else []
return jsonify({"clients": [_client_out(c) for c in clients],
"count": len(list(clients))})
@app.get("/api/clients/<int:client_id>")
def get_client(client_id: int):
client = _call(app, "get_client", client_id=client_id, include_balance=True)
@ -243,14 +301,16 @@ def create_app(core: Any, api_key: str) -> Flask:
kwargs: Dict[str, Any] = {
"client_id": client_id,
"amount_minor": _require_int_field(body, "amount_minor"),
"idempotency_key": _require_key(body, "idempotency_key"),
"ref_type": _optional_str(body, "ref_type"),
"ref_id": _optional_str(body, "ref_id"),
}
ttl = body.get("ttl_s")
if ttl is not None:
if isinstance(ttl, bool) or not isinstance(ttl, int) or ttl <= 0:
raise _bad_request("'ttl_s' must be a positive integer (seconds)")
kwargs["ttl_s"] = ttl
if isinstance(ttl, bool) or not isinstance(ttl, (int, float)) or ttl <= 0:
raise _bad_request("'ttl_s' must be a positive number (seconds)")
# Real core takes expires_in_minutes (a float, minutes).
kwargs["expires_in_minutes"] = ttl / 60.0
held = _call(app, "reserve_hold", **kwargs)
resp = jsonify(held)
resp.status_code = 201
@ -258,7 +318,10 @@ def create_app(core: Any, api_key: str) -> Flask:
@app.post("/api/holds/<int:hold_id>/release")
def release_hold(hold_id: int):
return jsonify(_call(app, "release_hold", hold_id=hold_id))
body = _json_body()
return jsonify(_call(app, "release_hold",
hold_id=hold_id,
idempotency_key=_require_key(body, "idempotency_key")))
# -- ledger ----------------------------------------------------------
@app.get("/api/clients/<int:client_id>/ledger")
@ -267,8 +330,8 @@ def create_app(core: Any, api_key: str) -> Flask:
to_ = request.args.get("to")
result = _call(app, "list_ledger", client_id=client_id,
from_=from_ or None, to=to_ or None)
rows = result.get("rows", []) if isinstance(result, dict) else []
return jsonify({"rows": rows, "count": len(list(rows))})
entries = result.get("entries", []) if isinstance(result, dict) else []
return jsonify({"rows": entries, "count": len(list(entries))})
return app
@ -276,10 +339,11 @@ def create_app(core: Any, api_key: str) -> Flask:
def _build_core(cfg):
"""Construct the real billing core from config (issue #1 module)."""
from gitea_saas import billing_core # lazy import: pure-server may omit it
db_path = cfg.db_path or Path(":memory:")
for ctor_name in ("BillingCore", "Core", "LedgerCore"):
ctor = getattr(billing_core, ctor_name, None)
if ctor is not None:
return ctor(cfg.db_path)
return ctor(db_path)
raise RuntimeError(
"gitea_saas.billing_core has no known constructor "
"(expected BillingCore/Core/LedgerCore)"

8
pyproject.toml Normal file
View File

@ -0,0 +1,8 @@
[project]
name = "gitea-saas-core"
version = "0.1.0"
description = "Shared billing core (clients/ledger/holds, idempotent, fail-closed) for gitea-saas"
requires-python = ">=3.10"
[tool.pytest.ini_options]
testpaths = ["tests"]

2
requirements.txt
View File

@ -1,2 +1,4 @@
# Runtime: Flask for the billing HTTP API (#2). Core (#1) is pure stdlib.
# Test/dev dependency:
flask>=3.0
pytest>=8.0

408
tests/test_billing_core.py Normal file
View File

@ -0,0 +1,408 @@
"""Hermetic tests for the billing ledger core.
Zero reliance on network / Stripe / Gitea: every test uses a temp-dir SQLite
file and pure stdlib. Run with: python -m pytest tests/test_billing_core.py -q
"""
import sys
import threading
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
# The package is a PEP 420 namespace package; make the project root importable
# regardless of how pytest was invoked.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from gitea_saas.billing_core import ( # noqa: E402
BillingCore,
DUPLICATE_CLIENT,
HOLD_ALREADY_RELEASED,
HOLD_NOT_FOUND,
IDEMPOTENCY_CONFLICT,
INSUFFICIENT_BALANCE,
INVALID_AMOUNT,
INVALID_KEY,
UNKNOWN_CLIENT,
)
class FakeClock:
"""Deterministic clock for hold-expiry tests (injected via now_fn)."""
def __init__(self, start: datetime | None = None):
self.now = start or datetime(2026, 1, 1, tzinfo=timezone.utc)
def __call__(self) -> datetime:
return self.now
def advance(self, minutes: float) -> None:
self.now += timedelta(minutes=minutes)
@pytest.fixture
def core(tmp_path):
return BillingCore(tmp_path / "billing.db")
def make_client(core, name="alice"):
ok, details = core.create_client(name)
assert ok, details
return details["client_id"]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
def test_create_tables_idempotent(core, tmp_path):
# Constructor already created them; calling again must be safe and
# non-destructive.
core.create_tables()
core.create_tables()
cid = make_client(core)
ok, details = core.get_client(cid)
assert ok and details["client"]["id"] == cid
# ---------------------------------------------------------------------------
# Credit / debit
# ---------------------------------------------------------------------------
def test_credit_balance_up_debit_balance_down(core):
cid = make_client(core)
ok, d = core.credit(cid, 500, "c1")
assert ok and d["kind"] == "credit" and d["amount_minor"] == 500
assert d["balance"] == 500
ok, d = core.credit(cid, 250, "c2")
assert ok and d["balance"] == 750
ok, d = core.debit(cid, 300, "d1")
assert ok and d["kind"] == "debit" and d["balance"] == 450
ok, details = core.get_client(cid, include_balance=True)
assert ok and details["balance"] == 450
# Debit down to exactly zero is allowed (never *negative*).
ok, d = core.debit(cid, 450, "d2")
assert ok and d["balance"] == 0
def test_debit_below_zero_rejected_and_no_row_written(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
ok, details = core.debit(cid, 200, "d1")
assert ok is False
assert details["error"]["code"] == INSUFFICIENT_BALANCE
# No debit row was written; balance unchanged.
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 1
assert ledger["entries"][0]["kind"] == "credit"
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 100
def test_duplicate_idempotency_key_credit_applied_once(core):
cid = make_client(core)
ok, first = core.credit(cid, 100, "k1")
assert ok and first["balance"] == 100
ok, second = core.credit(cid, 100, "k1")
assert ok and second["replayed"] is True
assert second["original"]["ledger_id"] == first["ledger_id"]
assert second["original"]["amount_minor"] == 100
assert second["balance"] == 100 # applied once
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 1
def test_duplicate_idempotency_key_debit_applied_once(core):
cid = make_client(core)
core.credit(cid, 200, "c1")
ok, d1 = core.debit(cid, 50, "k1")
assert ok and d1["balance"] == 150
ok, d2 = core.debit(cid, 50, "k1")
assert ok and d2["replayed"] is True and d2["balance"] == 150
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 2 # one credit, one debit
def test_idempotency_key_conflict_across_clients_is_an_error(core):
cid1, cid2 = make_client(core, "a"), make_client(core, "b")
ok, _ = core.credit(cid1, 100, "shared-key")
assert ok
ok, details = core.credit(cid2, 100, "shared-key")
assert ok is False and details["error"]["code"] == IDEMPOTENCY_CONFLICT
# ---------------------------------------------------------------------------
# Holds
# ---------------------------------------------------------------------------
def test_hold_reserves_release_returns(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, d = core.reserve_hold(cid, 300, "h1")
assert ok and d["hold_id"] and d["balance"] == 700
ok, details = core.get_client(cid)
assert ok and details["balance"] == 700
ok, d = core.release_hold(d["hold_id"], "r1")
assert ok and d["amount_minor"] == 300 and d["balance"] == 1000
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 3 # credit + hold + release
def test_hold_cannot_exceed_available_balance(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, _ = core.reserve_hold(cid, 700, "h1")
assert ok
ok, details = core.reserve_hold(cid, 400, "h2")
assert ok is False and details["error"]["code"] == INSUFFICIENT_BALANCE
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 300
def test_hold_expiry_auto_releases_on_next_query(tmp_path):
clock = FakeClock()
core = BillingCore(tmp_path / "billing.db", now_fn=clock)
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, d = core.reserve_hold(cid, 200, "h1") # default 30 min
assert ok and d["balance"] == 800 # reserved
# 31 minutes later the hold has expired; the next read auto-releases it.
clock.advance(minutes=31)
ok, details = core.get_client(cid)
assert ok and details["balance"] == 1000
ok, ledger = core.list_ledger(cid)
assert ledger["count"] == 3 # credit + hold + auto-release
kinds = {e["kind"] for e in ledger["entries"]}
assert kinds == {"credit", "hold", "release"}
assert any(e["idempotency_key"].startswith("auto-release:") for e in ledger["entries"])
def test_expire_stale_holds_releases_only_expired(tmp_path):
clock = FakeClock()
core = BillingCore(tmp_path / "billing.db", now_fn=clock)
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, hold_live = core.reserve_hold(cid, 200, "h-live", expires_in_minutes=60)
assert ok
clock.advance(minutes=30)
ok, hold_stale = core.reserve_hold(cid, 400, "h-stale", expires_in_minutes=10)
assert ok
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 400 # both holds active
clock.advance(minutes=11) # h-stale expired (t+41 > t+40), h-live not (t+41 < t+90)
ok, d = core.expire_stale_holds(cid)
assert ok and d["count"] == 1 and d["expired"][0]["hold_id"] == hold_stale["hold_id"]
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 800 # only the live hold still reserves
# Re-running is a no-op.
ok, d = core.expire_stale_holds(cid)
assert ok and d["count"] == 0
def test_hold_expired_at_reservation_never_reserves(tmp_path):
# A hold born already-expired (negative TTL / backdated) does not lock
# balance and is cleaned up on the next query.
clock = FakeClock()
core = BillingCore(tmp_path / "billing.db", now_fn=clock)
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, d = core.reserve_hold(cid, 200, "h1", expires_in_minutes=-1)
assert ok and d["balance"] == 1000 # already expired: no reservation
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 1000
ok, ledger = core.list_ledger(cid)
assert ledger["count"] == 3 # credit + hold + auto-release
def test_reserve_hold_replay_is_noop(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, first = core.reserve_hold(cid, 300, "h1")
assert ok
ok, second = core.reserve_hold(cid, 300, "h1")
assert ok and second["replayed"] is True
assert second["hold_id"] == first["hold_id"]
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 700
ok, ledger = core.list_ledger(cid)
assert ledger["count"] == 2 # credit + one hold row
def test_release_hold_replay_and_double_release(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, hold = core.reserve_hold(cid, 300, "h1")
ok, rel = core.release_hold(hold["hold_id"], "r1")
assert ok and rel["balance"] == 1000
# Same key: idempotent replay, not an error.
ok, rel2 = core.release_hold(hold["hold_id"], "r1")
assert ok and rel2["replayed"] is True and rel2["balance"] == 1000
# Different key on an already-released hold: structured error.
ok, details = core.release_hold(hold["hold_id"], "r2")
assert ok is False and details["error"]["code"] == HOLD_ALREADY_RELEASED
def test_release_unknown_hold(core):
ok, details = core.release_hold(99999, "r1")
assert ok is False and details["error"]["code"] == HOLD_NOT_FOUND
# ---------------------------------------------------------------------------
# Refunds
# ---------------------------------------------------------------------------
def test_refund_adds_back_and_preserves_idempotency(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
core.debit(cid, 100, "d1")
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 0
ok, d = core.refund(cid, 100, "rf1")
assert ok and d["kind"] == "refund" and d["balance"] == 100
ok, d2 = core.refund(cid, 100, "rf1")
assert ok and d2["replayed"] is True and d2["balance"] == 100
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 3
# ---------------------------------------------------------------------------
# Structured errors / validation (never raise, always tuples)
# ---------------------------------------------------------------------------
def test_unknown_client_structured_errors(core):
for op in (
lambda: core.get_client(999),
lambda: core.get_balance(999),
lambda: core.credit(999, 100, "x"),
lambda: core.debit(999, 100, "x"),
lambda: core.refund(999, 100, "x"),
lambda: core.reserve_hold(999, 100, "x"),
lambda: core.list_ledger(999),
lambda: core.expire_stale_holds(999),
):
ok, details = op()
assert ok is False
assert details["error"]["code"] == UNKNOWN_CLIENT
def test_invalid_amounts_rejected(core):
cid = make_client(core)
for bad in (0, -5, True, "10", 1.5, None):
ok, details = core.credit(cid, bad, f"k-{bad}")
assert ok is False and details["error"]["code"] == INVALID_AMOUNT
def test_missing_idempotency_key_rejected(core):
cid = make_client(core)
for bad in ("", " ", None, 42):
ok, details = core.credit(cid, 100, bad)
assert ok is False and details["error"]["code"] == INVALID_KEY
def test_duplicate_client_name_rejected(core):
make_client(core, "bob")
ok, details = core.create_client("bob")
assert ok is False and details["error"]["code"] == DUPLICATE_CLIENT
def test_empty_client_name_rejected(core):
ok, details = core.create_client(" ")
assert ok is False
def test_all_operations_return_tuples(core):
# Smoke check across the whole public surface: (bool, dict) always.
cid = make_client(core)
ops = [
core.get_client(cid),
core.get_balance(cid),
core.credit(cid, 50, "c1"),
core.debit(cid, 10, "d1"),
core.refund(cid, 5, "rf1"),
core.reserve_hold(cid, 5, "h1"),
core.list_ledger(cid),
core.expire_stale_holds(cid),
]
ok, hold = core.reserve_hold(cid, 5, "h2")
ops.append(core.release_hold(hold["hold_id"], "r1"))
ops.append(core.get_client(424242))
for result in ops:
assert isinstance(result, tuple) and len(result) == 2
assert isinstance(result[0], bool)
assert isinstance(result[1], dict)
# ---------------------------------------------------------------------------
# Queries / misc
# ---------------------------------------------------------------------------
def test_list_ledger_filters(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
core.credit(cid, 50, "c2")
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 2
ok, ledger = core.list_ledger(cid, to="2000-01-01T00:00:00+00:00")
assert ok and ledger["count"] == 0
ok, ledger = core.list_ledger(cid, from_="3000-01-01T00:00:00+00:00")
assert ok and ledger["count"] == 0
def test_get_client_include_balance_false(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
ok, details = core.get_client(cid, include_balance=False)
assert ok and "balance" not in details
assert details["client"]["name"] == "alice"
def test_create_client_meta_roundtrip(core):
cid = make_client(core)
ok, carol = core.create_client("carol", tier="enterprise", status="active",
meta={"plan": "pro", "seats": 3})
assert ok
ok, details = core.get_client(cid)
assert ok and details["client"]["tier"] == "free" # default tier
ok, details = core.get_client(carol["client_id"])
assert ok and details["client"]["name"] == "carol"
assert details["client"]["meta"] == {"plan": "pro", "seats": 3}
def test_ledger_entries_expose_all_columns(core):
cid = make_client(core)
core.credit(cid, 100, "c1", ref_type="issue", ref_id="42")
ok, ledger = core.list_ledger(cid)
entry = ledger["entries"][0]
assert entry["ref_type"] == "issue" and entry["ref_id"] == "42"
assert entry["currency"] == "USD"
assert entry["amount_minor"] == 100
assert entry["client_id"] == cid
# ---------------------------------------------------------------------------
# Thread safety
# ---------------------------------------------------------------------------
def test_concurrent_credits_are_serialized(core):
cid = make_client(core)
n_threads, per_thread = 8, 25
errors = []
def worker(t):
for i in range(per_thread):
ok, details = core.credit(cid, 1, f"t{t}-i{i}")
if not ok:
errors.append(details)
threads = [threading.Thread(target=worker, args=(t,)) for t in range(n_threads)]
for t in threads:
t.start()
for t in threads:
t.join()
assert errors == []
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == n_threads * per_thread
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == n_threads * per_thread

445
tests/test_server.py
View File

@ -1,321 +1,42 @@
"""Tests for gitea_saas.server against the issue #1 core contract.
"""Tests for gitea_saas.server against the real BillingCore (issue #1).
Hermetic by design: every test builds a temp SQLite DB and drives the app
through Flask's test client — no network, no Stripe, no Gitea.
The billing core (issue #1) is not yet on ``main``, so this file ships with a
compact reference implementation of the documented core contract
(``ReferenceCore``) used by the main suite. When the real core lands, the
``test_real_core_integration`` test (skipped until then) exercises the app
against it.
Issue #1 (the billing core) has landed on ``main``, so this suite runs
against the real ``gitea_saas.billing_core.BillingCore``. The only doubles
left are two minimal stubs that force error paths the real core cannot
produce on demand: a core missing an op (501) and a core that raises (500).
Every endpoint has at least one test against the real core.
"""
from __future__ import annotations
import datetime
import json
import sqlite3
import threading
from pathlib import Path
import pytest
from gitea_saas import server
from gitea_saas import billing_core, server
from gitea_saas.config import API_KEY_HEADER
API_KEY = "test-api-key-secret"
AUTH = {API_KEY_HEADER: API_KEY}
def _now_iso() -> str:
return datetime.datetime.now(datetime.timezone.utc).isoformat()
def _parse_dt(value: str) -> datetime.datetime:
return datetime.datetime.fromisoformat(value)
# --------------------------------------------------------------------------
# Reference implementation of the issue #1 core contract (SQLite-backed)
# Real-core app fixture
# --------------------------------------------------------------------------
class CoreError(Exception):
"""Internal marker; real contract is (ok, details) tuples."""
class ReferenceCore:
"""Minimal billing core honouring the documented ``(ok, details)``
contract, schema from issue #1, over a temp/in-memory SQLite DB."""
SCHEMA = """
CREATE TABLE IF NOT EXISTS clients (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
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
);
CREATE TABLE IF NOT EXISTS holds (
id INTEGER PRIMARY KEY AUTOINCREMENT,
client_id INTEGER NOT NULL REFERENCES clients(id),
amount_minor INTEGER NOT NULL,
ref_type TEXT,
ref_id TEXT,
expires_at TEXT NOT NULL,
released_at TEXT
);
"""
DEFAULT_HOLD_TTL_S = 30 * 60
def __init__(self, db_path: Path | str = ":memory:"):
self._lock = threading.RLock()
self._conn = sqlite3.connect(str(db_path), check_same_thread=False)
self._conn.row_factory = sqlite3.Row
self._conn.executescript(self.SCHEMA)
self._conn.commit()
def close(self) -> None:
with self._lock:
self._conn.close()
# -- internals --------------------------------------------------------
def _client_row(self, client_id: int):
row = self._conn.execute(
"SELECT * FROM clients WHERE id = ?", (client_id,)).fetchone()
return row
def _release_expired_holds(self, client_id: int | None = None) -> None:
now = _now_iso()
if client_id is None:
self._conn.execute(
"UPDATE holds SET released_at = ? WHERE released_at IS NULL "
"AND expires_at <= ?", (now, now))
else:
self._conn.execute(
"UPDATE holds SET released_at = ? WHERE client_id = ? AND "
"released_at IS NULL AND expires_at <= ?",
(now, client_id, now))
def _balance(self, client_id: int) -> int:
self._release_expired_holds(client_id)
pos, neg, held = self._conn.execute(
"SELECT "
" COALESCE((SELECT SUM(amount_minor) FROM ledger WHERE client_id = ? "
" AND kind IN ('credit','refund')), 0),"
" COALESCE((SELECT SUM(amount_minor) FROM ledger WHERE client_id = ? "
" AND kind = 'debit'), 0),"
" COALESCE((SELECT SUM(amount_minor) FROM holds WHERE client_id = ? "
" AND released_at IS NULL), 0)",
(client_id, client_id, client_id)).fetchone()
return int(pos) - int(neg) - int(held)
def _client_record(self, row: sqlite3.Row, include_balance: bool):
rec = {
"id": row["id"],
"name": row["name"],
"tier": row["tier"],
"status": row["status"],
"meta": json.loads(row["meta_json"]) if row["meta_json"] else None,
"created_at": row["created_at"],
}
if include_balance:
rec["balance"] = self._balance(row["id"])
return rec
def _idempotent(self, key: str, kind: str, client_id: int, amount: int):
"""Return (ok, row) if the key was already applied."""
row = self._conn.execute(
"SELECT * FROM ledger WHERE idempotency_key = ?",
(key,)).fetchone()
if row is None:
return False, None
return True, {
"client_id": row["client_id"],
"kind": row["kind"],
"amount_minor": row["amount_minor"],
"balance": self._balance(row["client_id"]),
"idempotency_key": key,
"duplicate": True,
}
# -- contract operations ----------------------------------------------
def create_client(self, name: str, tier: str = "standard",
meta=None) -> tuple:
with self._lock:
cur = self._conn.execute(
"INSERT INTO clients (name, tier, status, meta_json, created_at) "
"VALUES (?, ?, 'active', ?, ?)",
(name, tier, json.dumps(meta) if meta is not None else None,
_now_iso()))
self._conn.commit()
if cur.lastrowid is None: # pragma: no cover - sqlite contract
raise CoreError("insert produced no rowid")
row = self._client_row(int(cur.lastrowid))
return True, self._client_record(row, include_balance=True)
def get_client(self, client_id: int, include_balance: bool = True) -> tuple:
with self._lock:
row = self._client_row(client_id)
if row is None:
return False, {"code": "not_found",
"error": f"client {client_id} not found"}
return True, self._client_record(row, include_balance)
def credit(self, client_id: int, amount_minor: int, idempotency_key: str,
ref_type: str | None = None, ref_id: str | None = None) -> tuple:
with self._lock:
if self._client_row(client_id) is None:
return False, {"code": "not_found",
"error": f"client {client_id} not found"}
dup, details = self._idempotent(idempotency_key, "credit",
client_id, amount_minor)
if dup:
return True, details
self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency, "
"idempotency_key, ref_type, ref_id, created_at) "
"VALUES (?, 'credit', ?, 'USD', ?, ?, ?, ?)",
(client_id, amount_minor, idempotency_key, ref_type, ref_id,
_now_iso()))
self._conn.commit()
return True, {"client_id": client_id, "kind": "credit",
"amount_minor": amount_minor,
"balance": self._balance(client_id),
"idempotency_key": idempotency_key,
"duplicate": False}
def debit(self, client_id: int, amount_minor: int, idempotency_key: str,
ref_type: str | None = None, ref_id: str | None = None) -> tuple:
with self._lock:
if self._client_row(client_id) is None:
return False, {"code": "not_found",
"error": f"client {client_id} not found"}
dup, details = self._idempotent(idempotency_key, "debit",
client_id, amount_minor)
if dup:
return True, details
if self._balance(client_id) < amount_minor:
return False, {"code": "negative_balance",
"error": "debit would drive balance negative"}
self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency, "
"idempotency_key, ref_type, ref_id, created_at) "
"VALUES (?, 'debit', ?, 'USD', ?, ?, ?, ?)",
(client_id, amount_minor, idempotency_key, ref_type, ref_id,
_now_iso()))
self._conn.commit()
return True, {"client_id": client_id, "kind": "debit",
"amount_minor": amount_minor,
"balance": self._balance(client_id),
"idempotency_key": idempotency_key,
"duplicate": False}
def reserve_hold(self, client_id: int, amount_minor: int,
ref_type: str | None = None, ref_id: str | None = None,
ttl_s: int | None = None) -> tuple:
with self._lock:
if self._client_row(client_id) is None:
return False, {"code": "not_found",
"error": f"client {client_id} not found"}
if self._balance(client_id) < amount_minor:
return False, {"code": "insufficient_funds",
"error": "insufficient balance for hold"}
ttl = ttl_s if ttl_s is not None else self.DEFAULT_HOLD_TTL_S
expires = (datetime.datetime.now(datetime.timezone.utc)
+ datetime.timedelta(seconds=ttl)).isoformat()
cur = self._conn.execute(
"INSERT INTO holds (client_id, amount_minor, ref_type, ref_id, "
"expires_at) VALUES (?, ?, ?, ?, ?)",
(client_id, amount_minor, ref_type, ref_id, expires))
self._conn.commit()
if cur.lastrowid is None: # pragma: no cover - sqlite contract
raise CoreError("insert produced no rowid")
return True, {"hold_id": int(cur.lastrowid),
"client_id": client_id,
"amount_minor": amount_minor,
"expires_at": expires,
"status": "held"}
def release_hold(self, hold_id: int) -> tuple:
with self._lock:
row = self._conn.execute(
"SELECT * FROM holds WHERE id = ?", (hold_id,)).fetchone()
if row is None:
return False, {"code": "not_found",
"error": f"hold {hold_id} not found"}
if row["released_at"] is not None:
return True, {"hold_id": hold_id, "released": True,
"status": "already_released"}
self._conn.execute(
"UPDATE holds SET released_at = ? WHERE id = ?",
(_now_iso(), hold_id))
self._conn.commit()
return True, {"hold_id": hold_id, "released": True,
"status": "released"}
def list_ledger(self, client_id: int, from_: str | None = None,
to: str | None = None) -> tuple:
with self._lock:
if self._client_row(client_id) is None:
return False, {"code": "not_found",
"error": f"client {client_id} not found"}
sql = ("SELECT id, client_id, kind, amount_minor, currency, "
"idempotency_key, ref_type, ref_id, created_at "
"FROM ledger WHERE client_id = ?")
args: list = [client_id]
if from_ is not None:
sql += " AND created_at >= ?"
args.append(from_)
if to is not None:
sql += " AND created_at <= ?"
args.append(to)
sql += " ORDER BY id"
rows = self._conn.execute(sql, args).fetchall()
return True, {"rows": [dict(r) for r in rows]}
def list_clients(self) -> tuple:
with self._lock:
rows = self._conn.execute(
"SELECT * FROM clients ORDER BY id").fetchall()
return True, {"clients": [self._client_record(r, True) for r in rows]}
# --------------------------------------------------------------------------
# App fixtures
# --------------------------------------------------------------------------
try: # exercise the real core when issue #1 lands
from gitea_saas import billing_core as _real_core_module
HAS_REAL_CORE = hasattr(_real_core_module, "BillingCore")
except Exception: # pragma: no cover - depends on unmerged issue #1
_real_core_module = None
HAS_REAL_CORE = False
@pytest.fixture()
def app(tmp_path: Path):
core = ReferenceCore(tmp_path / "test.db")
core = billing_core.BillingCore(tmp_path / "test.db")
application = server.create_app(core, API_KEY)
application.config.update(TESTING=True)
try:
yield application
finally:
core.close()
core._conn.close()
@pytest.fixture()
@ -327,6 +48,30 @@ def _post(client, url, body, key=API_KEY):
return client.post(url, json=body, headers={API_KEY_HEADER: key})
# --------------------------------------------------------------------------
# Minimal doubles for error paths the real core cannot produce on demand
# --------------------------------------------------------------------------
class MissingOpCore:
"""Looks like the core but omits ``credit`` -> drives the 501 path."""
def create_client(self, name, tier="free", status="active", meta=None):
return True, {"client_id": 1, "name": name, "tier": tier,
"status": status, "meta": meta or {},
"created_at": "2026-01-01T00:00:00+00:00"}
def get_client(self, client_id, include_balance=True):
return False, {"error": {"code": "UNKNOWN_CLIENT",
"message": f"no client with id {client_id}"}}
class RaisingCore:
"""Raises where the real core returns a tuple -> drives the 500 path."""
def credit(self, **kwargs):
raise RuntimeError("boom")
# --------------------------------------------------------------------------
# Auth / health
# --------------------------------------------------------------------------
@ -338,20 +83,23 @@ def test_health_is_public(client):
def test_missing_key_is_401(client):
resp = client.get("/api/clients")
resp = client.get("/api/clients/1")
assert resp.status_code == 401
body = resp.get_json()
assert body["code"] == "unauthorized"
def test_wrong_key_is_401(client):
resp = client.get("/api/clients", headers={API_KEY_HEADER: "nope"})
resp = client.get("/api/clients/1", headers={API_KEY_HEADER: "nope"})
assert resp.status_code == 401
assert resp.get_json()["code"] == "unauthorized"
def test_correct_key_is_200(client):
resp = client.get("/api/clients", headers=AUTH)
created = _post(client, "/api/clients", {"name": "AuthOK"})
assert created.status_code == 201
cid = created.get_json()["id"]
resp = client.get(f"/api/clients/{cid}", headers=AUTH)
assert resp.status_code == 200
@ -360,14 +108,17 @@ def test_correct_key_is_200(client):
# --------------------------------------------------------------------------
def test_full_lifecycle_via_api(client):
# create
# create -> response carries the API's id plus a balance
created = _post(client, "/api/clients",
{"name": "Acme", "tier": "enterprise", "meta": {"v": 1}})
assert created.status_code == 201
cid = created.get_json()["id"]
assert created.get_json()["balance"] == 0
assert created.get_json()["name"] == "Acme"
assert created.get_json()["meta"] == {"v": 1}
out = created.get_json()
cid = out["id"]
assert out["balance"] == 0
assert out["name"] == "Acme"
assert out["tier"] == "enterprise"
assert out["meta"] == {"v": 1}
assert out["status"] == "active"
# credit
cred = _post(client, f"/api/clients/{cid}/credits",
@ -375,14 +126,18 @@ def test_full_lifecycle_via_api(client):
assert cred.status_code == 200
assert cred.get_json()["balance"] == 5000
# get
# get -> full client record plus balance
got = client.get(f"/api/clients/{cid}", headers=AUTH)
assert got.status_code == 200
assert got.get_json()["balance"] == 5000
g = got.get_json()
assert g["balance"] == 5000
assert g["name"] == "Acme"
assert g["tier"] == "enterprise"
# hold
held = _post(client, f"/api/clients/{cid}/holds",
{"amount_minor": 2000, "ref_type": "job", "ref_id": "j1"})
{"amount_minor": 2000, "idempotency_key": "h1",
"ref_type": "job", "ref_id": "j1"})
assert held.status_code == 201
hold_id = held.get_json()["hold_id"]
got = client.get(f"/api/clients/{cid}", headers=AUTH)
@ -395,17 +150,18 @@ def test_full_lifecycle_via_api(client):
assert deb.get_json()["balance"] == 0 # 5000 - 3000 - 2000(held)
# release hold -> balance restored
rel = _post(client, f"/api/holds/{hold_id}/release", {})
rel = _post(client, f"/api/holds/{hold_id}/release",
{"idempotency_key": "r1"})
assert rel.status_code == 200
assert rel.get_json()["released"] is True
assert "released_at" in rel.get_json()
got = client.get(f"/api/clients/{cid}", headers=AUTH)
assert got.get_json()["balance"] == 2000
# ledger shows everything
# ledger shows all four kinds (credit, hold, debit, release)
led = client.get(f"/api/clients/{cid}/ledger", headers=AUTH)
assert led.status_code == 200
kinds = [r["kind"] for r in led.get_json()["rows"]]
assert kinds == ["credit", "debit"]
kinds = {r["kind"] for r in led.get_json()["rows"]}
assert kinds == {"credit", "hold", "debit", "release"}
def test_debit_below_zero_is_409_and_no_change(client):
@ -415,7 +171,7 @@ def test_debit_below_zero_is_409_and_no_change(client):
{"amount_minor": 100, "idempotency_key": "d-lose"})
assert resp.status_code == 409
body = resp.get_json()
assert body["code"] == "negative_balance"
assert body["code"] == "insufficient_balance"
assert body["error"]
got = client.get(f"/api/clients/{cid}", headers=AUTH)
assert got.get_json()["balance"] == 0
@ -435,17 +191,22 @@ def test_duplicate_idempotency_key_no_double_apply(client):
{"amount_minor": 777, "idempotency_key": "same"})
assert first.status_code == second.status_code == 200
assert first.get_json()["balance"] == second.get_json()["balance"] == 777
assert second.get_json()["duplicate"] is True
assert second.get_json().get("replayed") is True
got = client.get(f"/api/clients/{cid}", headers=AUTH)
assert got.get_json()["balance"] == 777
# exactly one credit row in the ledger
led = client.get(f"/api/clients/{cid}/ledger", headers=AUTH)
credits = [r for r in led.get_json()["rows"] if r["kind"] == "credit"]
assert len(credits) == 1
def test_hold_insufficient_is_409(client):
created = _post(client, "/api/clients", {"name": "Tiny"})
cid = created.get_json()["id"]
resp = _post(client, f"/api/clients/{cid}/holds", {"amount_minor": 10})
resp = _post(client, f"/api/clients/{cid}/holds",
{"amount_minor": 10, "idempotency_key": "h-x"})
assert resp.status_code == 409
assert resp.get_json()["code"] == "insufficient_funds"
assert resp.get_json()["code"] == "insufficient_balance"
def test_hold_expiry_auto_releases(client, app):
@ -454,7 +215,7 @@ def test_hold_expiry_auto_releases(client, app):
_post(client, f"/api/clients/{cid}/credits",
{"amount_minor": 1000, "idempotency_key": "e1"})
held = _post(client, f"/api/clients/{cid}/holds",
{"amount_minor": 400, "ttl_s": 1})
{"amount_minor": 400, "idempotency_key": "h2", "ttl_s": 60})
hold_id = held.get_json()["hold_id"]
assert client.get(f"/api/clients/{cid}",
headers=AUTH).get_json()["balance"] == 600
@ -471,15 +232,16 @@ def test_hold_expiry_auto_releases(client, app):
def test_release_missing_hold_is_404(client):
resp = _post(client, "/api/holds/999999/release", {})
resp = _post(client, "/api/holds/999999/release",
{"idempotency_key": "r-x"})
assert resp.status_code == 404
assert resp.get_json()["code"] == "not_found"
assert resp.get_json()["code"] == "hold_not_found"
def test_unknown_client_is_404(client):
resp = client.get("/api/clients/424242", headers=AUTH)
assert resp.status_code == 404
assert resp.get_json()["code"] == "not_found"
assert resp.get_json()["code"] == "unknown_client"
def test_ledger_filter_from_to(client):
@ -496,15 +258,9 @@ def test_ledger_filter_from_to(client):
assert led.get_json()["count"] == 2
def test_list_clients_admin(client):
_post(client, "/api/clients", {"name": "A"})
_post(client, "/api/clients", {"name": "B"})
resp = client.get("/api/clients", headers=AUTH)
assert resp.status_code == 200
body = resp.get_json()
assert body["count"] == 2
assert {c["name"] for c in body["clients"]} == {"A", "B"}
# --------------------------------------------------------------------------
# Validation / error shape
# --------------------------------------------------------------------------
def test_errors_are_json_objects(client):
resp = _post(client, "/api/clients", {"amount_minor": "x"}) # bad body
@ -535,26 +291,35 @@ def test_create_client_requires_name(client):
assert resp.status_code == 400
def test_duplicate_client_name_is_409(client):
first = _post(client, "/api/clients", {"name": "DupName"})
assert first.status_code == 201
second = _post(client, "/api/clients", {"name": "DupName"})
assert second.status_code == 409
assert second.get_json()["code"] == "duplicate_client"
# --------------------------------------------------------------------------
# Real-core integration (runs only once issue #1 has landed)
# Error paths the real core cannot produce on demand (fakes only here)
# --------------------------------------------------------------------------
@pytest.mark.skipif(not HAS_REAL_CORE,
reason="billing core (issue #1) not merged onto main yet")
def test_real_core_integration(tmp_path: Path):
core = _real_core_module.BillingCore(tmp_path / "real.db")
application = server.create_app(core, API_KEY)
def test_missing_core_op_is_501():
application = server.create_app(MissingOpCore(), API_KEY)
application.config.update(TESTING=True)
c = application.test_client()
try:
created = c.post("/api/clients", json={"name": "Real"},
headers=AUTH)
assert created.status_code == 201
cid = created.get_json()["id"]
resp = c.post(f"/api/clients/{cid}/credits",
json={"amount_minor": 100, "idempotency_key": "r1"},
headers=AUTH)
assert resp.status_code == 200
assert resp.get_json()["balance"] == 100
finally:
core.close() if hasattr(core, "close") else None
resp = c.post("/api/clients/1/credits",
json={"amount_minor": 100, "idempotency_key": "x"},
headers=AUTH)
assert resp.status_code == 501
assert resp.get_json()["code"] == "not_implemented"
def test_core_raise_is_500():
application = server.create_app(RaisingCore(), API_KEY)
application.config.update(TESTING=True)
c = application.test_client()
resp = c.post("/api/clients/1/credits",
json={"amount_minor": 100, "idempotency_key": "x"},
headers=AUTH)
assert resp.status_code == 500
assert resp.get_json()["code"] == "core_error"