139 lines
5.7 KiB
Python
139 lines
5.7 KiB
Python
"""
|
|
Drive subscription enforcement against a running application, over HTTP.
|
|
|
|
The entitlement layer has only ever been exercised through a TestClient. This
|
|
runs it the way production would: a real uvicorn process, real middleware
|
|
ordering, `SUBSCRIPTION_ENFORCEMENT_ENABLED` on, and a tenant whose
|
|
subscription state is changed underneath it between calls.
|
|
|
|
APP_ENV=verify python scripts/smoke_billing.py --port 20077
|
|
|
|
What matters here is the *shape* of the degradation, not just that a code is
|
|
returned. A lapsed subscription must leave the tenant able to read their own
|
|
data and unable to add more — anything else is either a lock-out (support
|
|
ticket) or a free tier nobody sold.
|
|
|
|
Reports rather than asserts, for the same reason `smoke_org.py` does.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("APP_ENV", "verify")
|
|
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
|
sys.path.insert(0, str(Path.cwd()))
|
|
|
|
import requests # noqa: E402
|
|
from sqlalchemy import create_engine, text # noqa: E402
|
|
|
|
PASS = " ok "
|
|
FAIL = " FAIL "
|
|
results = []
|
|
|
|
|
|
def check(label: str, ok: bool, detail: str = "") -> None:
|
|
results.append(ok)
|
|
print(f"[{PASS if ok else FAIL}] {label}" + (f" — {detail}" if detail else ""))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--port", type=int, default=20077)
|
|
args = parser.parse_args()
|
|
base = f"http://127.0.0.1:{args.port}"
|
|
|
|
from app.core.security import create_access_token
|
|
from app.core.settings import DATABASE_URL
|
|
|
|
engine = create_engine(DATABASE_URL, future=True)
|
|
|
|
with engine.connect() as conn:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT u.id, u.tenant_id FROM users u "
|
|
"WHERE u.tenant_id IS NOT NULL AND u.is_superadmin = false "
|
|
"ORDER BY u.id LIMIT 1"
|
|
)
|
|
).first()
|
|
if row is None:
|
|
print("No tenant user found. Run scripts/seed_smoke.py first.")
|
|
return 1
|
|
user_id, tenant_id = row
|
|
|
|
plan_id = conn.execute(
|
|
text("SELECT id FROM plans WHERE code = 'starter'")
|
|
).scalar()
|
|
|
|
token = create_access_token({"sub": str(user_id), "tenant_id": str(tenant_id)})
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
def set_subscription(status: str | None) -> None:
|
|
"""Put the tenant into a subscription state, from outside the app."""
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text("DELETE FROM tenant_subscriptions WHERE tenant_id = :t"),
|
|
{"t": tenant_id},
|
|
)
|
|
if status is not None:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO tenant_subscriptions "
|
|
"(id, tenant_id, plan_id, status, started_at) "
|
|
"VALUES (gen_random_uuid(), :t, :p, :s, now())"
|
|
),
|
|
{"t": tenant_id, "p": plan_id, "s": status},
|
|
)
|
|
|
|
print("\n--- the public plan list ---")
|
|
r = requests.get(f"{base}/api/billing/plans", timeout=30)
|
|
check("plans are readable without a token", r.status_code == 200, f"HTTP {r.status_code}")
|
|
plans = r.json() if r.status_code == 200 else []
|
|
check("every public plan carries its limits", bool(plans) and all(p.get("limits") for p in plans), f"{len(plans)} plans")
|
|
check(
|
|
"no internal ids beyond the plan id leak into the public list",
|
|
all(set(p) <= {"id", "code", "name", "description", "price_amount", "currency", "interval", "grace_period_days", "notify_days_before_expiry", "is_public", "is_popular", "sort_order", "limits"} for p in plans), )
|
|
|
|
print("\n--- an active subscription ---")
|
|
set_subscription("active")
|
|
r = requests.get(f"{base}/api/billing/me", headers=headers, timeout=30)
|
|
check("the tenant can read its own entitlement", r.status_code == 200, f"HTTP {r.status_code}")
|
|
if r.status_code == 200:
|
|
body = r.json()
|
|
check("it reports full access", body.get("access_level") == "full", str(body.get("access_level")))
|
|
|
|
r = requests.post(f"{base}/api/me/projects/", json={"name": "Smoke project"}, headers=headers, timeout=30)
|
|
check("a write is allowed", r.status_code not in (402,), f"HTTP {r.status_code}")
|
|
|
|
print("\n--- a lapsed subscription ---")
|
|
set_subscription("past_due")
|
|
r = requests.get(f"{base}/api/me/projects", headers=headers, timeout=30)
|
|
check("reads still work — the tenant is not locked out", r.status_code == 200, f"HTTP {r.status_code}")
|
|
|
|
r = requests.post(f"{base}/api/me/projects/", json={"name": "Should be refused"}, headers=headers, timeout=30)
|
|
check("a write is refused with 402, not 403 or 500", r.status_code == 402, f"HTTP {r.status_code} {r.text[:120]}")
|
|
|
|
r = requests.get(f"{base}/api/billing/me", headers=headers, timeout=30)
|
|
if r.status_code == 200:
|
|
check("entitlement reports read_only", r.json().get("access_level") == "read_only", str(r.json().get("access_level")))
|
|
|
|
print("\n--- no subscription at all ---")
|
|
set_subscription(None)
|
|
r = requests.get(f"{base}/api/me/projects", headers=headers, timeout=30)
|
|
check("an unmapped tenant is not locked out of reads", r.status_code == 200, f"HTTP {r.status_code}")
|
|
|
|
print("\n--- restored ---")
|
|
set_subscription("active")
|
|
r = requests.post(f"{base}/api/me/projects/", json={"name": "Restored"}, headers=headers, timeout=30)
|
|
check("writes resume once the subscription is active", r.status_code not in (402,), f"HTTP {r.status_code}")
|
|
|
|
passed = sum(1 for ok in results if ok)
|
|
print("\n" + "=" * 62)
|
|
print(f"{passed}/{len(results)} checks passed")
|
|
return 0 if passed == len(results) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|