d7d462de14
Implements gitea-saas-core#2: Flask app over the issue #1 billing core. - create_app(core, api_key) factory; X-API-Key auth on every route except /health (missing/wrong key -> 401), hmac.compare_digest, no key configured -> deny all (fail-closed). - Endpoints: POST /api/clients, GET /api/clients (admin list), GET /api/clients/{id}, POST .../credits, POST .../debits (409 on negative balance), POST .../holds (409 on insufficient funds), POST /api/holds/{id}/release, GET .../ledger?from=&to=. - JSON in/out, errors always {"error", "code"}; core (ok, details) contract enforced; missing core op -> 501, unexpected core failure -> 500 JSON (never HTML, never leaked internals). - config.py reads SAAS_API_KEY/SAAS_PORT/SAAS_DB_PATH with sane defaults; never reads Stripe keys. CLI (python -m gitea_saas.server) refuses to start without SAAS_API_KEY. - tests use temp SQLite + Flask test client, no network. The billing core (#1) is not merged onto main yet, so tests carry a reference implementation of the documented core contract; a skipped-until-then integration test runs against the real core once it lands. Verification: python -m pytest tests/test_server.py -q -> 17 passed, 1 skipped.
561 lines
22 KiB
Python
561 lines
22 KiB
Python
"""Tests for gitea_saas.server against the issue #1 core contract.
|
|
|
|
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.
|
|
"""
|
|
|
|
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.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)
|
|
# --------------------------------------------------------------------------
|
|
|
|
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")
|
|
application = server.create_app(core, API_KEY)
|
|
application.config.update(TESTING=True)
|
|
try:
|
|
yield application
|
|
finally:
|
|
core.close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(app):
|
|
return app.test_client()
|
|
|
|
|
|
def _post(client, url, body, key=API_KEY):
|
|
return client.post(url, json=body, headers={API_KEY_HEADER: key})
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Auth / health
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_health_is_public(client):
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"ok": True}
|
|
|
|
|
|
def test_missing_key_is_401(client):
|
|
resp = client.get("/api/clients")
|
|
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"})
|
|
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)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Client lifecycle
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_full_lifecycle_via_api(client):
|
|
# create
|
|
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}
|
|
|
|
# credit
|
|
cred = _post(client, f"/api/clients/{cid}/credits",
|
|
{"amount_minor": 5000, "idempotency_key": "c1"})
|
|
assert cred.status_code == 200
|
|
assert cred.get_json()["balance"] == 5000
|
|
|
|
# get
|
|
got = client.get(f"/api/clients/{cid}", headers=AUTH)
|
|
assert got.status_code == 200
|
|
assert got.get_json()["balance"] == 5000
|
|
|
|
# hold
|
|
held = _post(client, f"/api/clients/{cid}/holds",
|
|
{"amount_minor": 2000, "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)
|
|
assert got.get_json()["balance"] == 3000 # 5000 - 2000 held
|
|
|
|
# debit (up to held-away balance)
|
|
deb = _post(client, f"/api/clients/{cid}/debits",
|
|
{"amount_minor": 3000, "idempotency_key": "d1"})
|
|
assert deb.status_code == 200
|
|
assert deb.get_json()["balance"] == 0 # 5000 - 3000 - 2000(held)
|
|
|
|
# release hold -> balance restored
|
|
rel = _post(client, f"/api/holds/{hold_id}/release", {})
|
|
assert rel.status_code == 200
|
|
assert rel.get_json()["released"] is True
|
|
got = client.get(f"/api/clients/{cid}", headers=AUTH)
|
|
assert got.get_json()["balance"] == 2000
|
|
|
|
# ledger shows everything
|
|
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"]
|
|
|
|
|
|
def test_debit_below_zero_is_409_and_no_change(client):
|
|
created = _post(client, "/api/clients", {"name": "Broke"})
|
|
cid = created.get_json()["id"]
|
|
resp = _post(client, f"/api/clients/{cid}/debits",
|
|
{"amount_minor": 100, "idempotency_key": "d-lose"})
|
|
assert resp.status_code == 409
|
|
body = resp.get_json()
|
|
assert body["code"] == "negative_balance"
|
|
assert body["error"]
|
|
got = client.get(f"/api/clients/{cid}", headers=AUTH)
|
|
assert got.get_json()["balance"] == 0
|
|
# idempotency key must not be burned by a rejected debit
|
|
resp2 = _post(client, f"/api/clients/{cid}/credits",
|
|
{"amount_minor": 50, "idempotency_key": "d-lose"})
|
|
assert resp2.status_code == 200
|
|
assert resp2.get_json()["balance"] == 50
|
|
|
|
|
|
def test_duplicate_idempotency_key_no_double_apply(client):
|
|
created = _post(client, "/api/clients", {"name": "Dup"})
|
|
cid = created.get_json()["id"]
|
|
first = _post(client, f"/api/clients/{cid}/credits",
|
|
{"amount_minor": 777, "idempotency_key": "same"})
|
|
second = _post(client, f"/api/clients/{cid}/credits",
|
|
{"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
|
|
got = client.get(f"/api/clients/{cid}", headers=AUTH)
|
|
assert got.get_json()["balance"] == 777
|
|
|
|
|
|
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})
|
|
assert resp.status_code == 409
|
|
assert resp.get_json()["code"] == "insufficient_funds"
|
|
|
|
|
|
def test_hold_expiry_auto_releases(client, app):
|
|
created = _post(client, "/api/clients", {"name": "Expiry"})
|
|
cid = created.get_json()["id"]
|
|
_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})
|
|
hold_id = held.get_json()["hold_id"]
|
|
assert client.get(f"/api/clients/{cid}",
|
|
headers=AUTH).get_json()["balance"] == 600
|
|
# force expiry by rewriting timestamps back in time
|
|
core = app.config["CORE"]
|
|
past = (datetime.datetime.now(datetime.timezone.utc)
|
|
- datetime.timedelta(seconds=5)).isoformat()
|
|
with core._lock:
|
|
core._conn.execute("UPDATE holds SET expires_at = ? WHERE id = ?",
|
|
(past, hold_id))
|
|
core._conn.commit()
|
|
got = client.get(f"/api/clients/{cid}", headers=AUTH)
|
|
assert got.get_json()["balance"] == 1000 # auto-released on read
|
|
|
|
|
|
def test_release_missing_hold_is_404(client):
|
|
resp = _post(client, "/api/holds/999999/release", {})
|
|
assert resp.status_code == 404
|
|
assert resp.get_json()["code"] == "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"
|
|
|
|
|
|
def test_ledger_filter_from_to(client):
|
|
created = _post(client, "/api/clients", {"name": "Filter"})
|
|
cid = created.get_json()["id"]
|
|
_post(client, f"/api/clients/{cid}/credits",
|
|
{"amount_minor": 10, "idempotency_key": "f1"})
|
|
_post(client, f"/api/clients/{cid}/credits",
|
|
{"amount_minor": 20, "idempotency_key": "f2"})
|
|
# from=epoch start -> both rows; to=something future -> both rows
|
|
led = client.get(f"/api/clients/{cid}/ledger?from=2000-01-01T00:00:00",
|
|
headers=AUTH)
|
|
assert led.status_code == 200
|
|
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"}
|
|
|
|
|
|
def test_errors_are_json_objects(client):
|
|
resp = _post(client, "/api/clients", {"amount_minor": "x"}) # bad body
|
|
assert resp.status_code == 400
|
|
body = resp.get_json()
|
|
assert set(body) == {"error", "code"}
|
|
assert isinstance(body["error"], str)
|
|
|
|
|
|
def test_invalid_amount_is_400(client):
|
|
created = _post(client, "/api/clients", {"name": "Bad"})
|
|
cid = created.get_json()["id"]
|
|
resp = _post(client, f"/api/clients/{cid}/credits",
|
|
{"amount_minor": -5, "idempotency_key": "neg"})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()["code"] == "invalid_request"
|
|
|
|
|
|
def test_missing_idempotency_key_is_400(client):
|
|
created = _post(client, "/api/clients", {"name": "NoKey"})
|
|
cid = created.get_json()["id"]
|
|
resp = _post(client, f"/api/clients/{cid}/credits", {"amount_minor": 1})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_create_client_requires_name(client):
|
|
resp = _post(client, "/api/clients", {"tier": "standard"})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Real-core integration (runs only once issue #1 has landed)
|
|
# --------------------------------------------------------------------------
|
|
|
|
@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)
|
|
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
|