Files
datatools-dev/server/tests/test_routes.py
Michael bab2c9468c feat(server): mint API + Postgres schema + manual adapter (PR 1)
Source-agnostic license issuance service. FastAPI app fronts a
Postgres `licenses` table; the only currently-wired source is
`manual` (operator mints via /internal/mint). Gumroad webhook
adapter lands in PR 2.

Key design points:

- Signing reuses src/license/crypto.py via a COPY into the image
  (single source of truth — blobs minted server-side verify against
  the same embedded pubkey on the buyer's machine).
- Source adapter Protocol (app/adapters/base.py) is the seam for
  Gumroad / Lemon Squeezy / Stripe in later PRs; Mint API speaks
  only SaleEvent / RefundEvent.
- (source, source_order_id) UNIQUE composite gives idempotent
  webhook retries without double-mint.
- JSONB type uses with_variant(JSON, 'sqlite') so the same models
  drive both Postgres prod and SQLite tests (no testcontainers dep).
- Bearer-token auth on /internal/*; the IP-loopback guard was
  removed after the docker bridge made it fight legitimate prod
  traffic (nginx defense + Bearer remain).
- Secrets resolved via *_FILE env vars pointing at
  /run/secrets/<name>, so passwords never appear in `docker inspect`.

21 unit tests (SQLite in-memory, StaticPool) plus a real-Postgres
docker-compose smoke test in server/scripts/smoke.sh that builds the
image, runs the alembic migration, mints a license, verifies the
signature against the host dev pubkey, and checks the DB row.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 00:46:54 +00:00

131 lines
3.7 KiB
Python

"""HTTP route tests — auth, mint, revoke, list, health."""
from __future__ import annotations
def test_health_is_public(client):
r = client.get("/health")
assert r.status_code == 200
assert r.json()["status"] == "ok"
def test_internal_requires_bearer(client):
r = client.get("/internal/ping")
assert r.status_code == 401
def test_internal_rejects_wrong_bearer(client):
r = client.get("/internal/ping", headers={"Authorization": "Bearer nope"})
assert r.status_code == 401
def test_internal_ping_ok_with_token(client, admin_headers):
r = client.get("/internal/ping", headers=admin_headers)
assert r.status_code == 200
assert r.json() == {"ok": True}
def test_mint_creates_license(client, admin_headers):
r = client.post(
"/internal/mint",
headers=admin_headers,
json={
"name": "Jane Doe",
"email": "jane@example.com",
"tier": "core",
"years": 1,
"source": "manual",
},
)
assert r.status_code == 201, r.text
body = r.json()
assert body["tier"] == "core"
assert body["source"] == "manual"
assert body["blob"].startswith("DTLIC2:")
def test_mint_rejects_non_manual_source(client, admin_headers):
r = client.post(
"/internal/mint",
headers=admin_headers,
json={
"name": "x", "email": "x@example.com", "tier": "core",
"source": "gumroad",
},
)
assert r.status_code == 400
assert "not wired" in r.json()["detail"]
def test_mint_rejects_bad_email(client, admin_headers):
r = client.post(
"/internal/mint",
headers=admin_headers,
json={"name": "x", "email": "not-an-email", "tier": "core"},
)
assert r.status_code == 422
def test_mint_rejects_unknown_tier(client, admin_headers):
r = client.post(
"/internal/mint",
headers=admin_headers,
json={"name": "x", "email": "x@example.com", "tier": "platinum"},
)
assert r.status_code == 422
def test_list_licenses_filters_email_case_insensitive(client, admin_headers):
# Pydantic EmailStr normalizes the domain to lowercase per RFC.
for email in ("alice@example.com", "Bob@Example.com", "carol@other.test"):
client.post(
"/internal/mint",
headers=admin_headers,
json={"name": "User", "email": email, "tier": "core"},
)
r = client.get(
"/internal/licenses?email=example.com",
headers=admin_headers,
)
assert r.status_code == 200
emails = {row["email"].lower() for row in r.json()}
assert "alice@example.com" in emails
assert "bob@example.com" in emails
assert "carol@other.test" not in emails
def test_revoke_then_excluded_by_default(client, admin_headers):
r = client.post(
"/internal/mint",
headers=admin_headers,
json={"name": "x", "email": "x@example.com", "tier": "lite"},
)
key = r.json()["license_key"]
r2 = client.post(
"/internal/revoke",
headers=admin_headers,
json={"license_key": key, "reason": "refund"},
)
assert r2.status_code == 200
assert r2.json()["revoked_at"] is not None
listed = client.get("/internal/licenses", headers=admin_headers).json()
assert all(row["license_key"] != key for row in listed)
listed_all = client.get(
"/internal/licenses?include_revoked=true",
headers=admin_headers,
).json()
assert any(row["license_key"] == key for row in listed_all)
def test_revoke_unknown_returns_404(client, admin_headers):
r = client.post(
"/internal/revoke",
headers=admin_headers,
json={"license_key": "DT1-CORE-doesnot-exist"},
)
assert r.status_code == 404