Giorgio Gilestro
411094d7b8
New independent `admin` service (admin.main:app) on the same image, reusing app.db/app.models read-only. Never runs migrations or the scheduler; issues SELECTs only. - Password-gated (ADMIN_CONSOLE_PASSWORD) with a 12h signed cookie; closed by default when the password is empty. - Bound to 127.0.0.1:8091 (SSH-tunnel access); off the intranet/NPM network. - Pages: overview stats, user list + search, per-user history/payment detail, DB usage (information_schema size + row estimates). - Compose: base `admin` service (+prod DB-host override, test mount); Dockerfile bakes admin/ into runtime + test stages. - Tests: tests/test_admin_console.py (auth, queries, page wiring) — 12 passing. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
264 lines
8.6 KiB
Python
264 lines
8.6 KiB
Python
"""Superadmin console — auth gate, read-only queries, and page wiring.
|
|
|
|
Runs against an in-memory-ish sqlite file (same pattern as the other API
|
|
tests): seed a handful of users + related rows, rebind app.db to the test
|
|
engine, then drive both the query layer directly and the FastAPI app via
|
|
TestClient.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import timedelta
|
|
|
|
|
|
def _seed(tmp_path):
|
|
"""Create schema + fixture rows; rebind app.db. Returns the factory."""
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app import db as db_mod
|
|
from app.db import Base, utcnow
|
|
from app.models import (
|
|
EmailSend, PortfolioSync, Referral, StrategicLogFeedback,
|
|
User, UserAcknowledgement,
|
|
)
|
|
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/admin.db")
|
|
factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
db_mod._engine = engine
|
|
db_mod._session_factory = factory
|
|
|
|
now = utcnow()
|
|
|
|
async def _go():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
async with factory() as s:
|
|
# 1: free, referrer
|
|
s.add(User(id=1, email="alice@x", tier="free", created_at=now,
|
|
last_login_at=now, referral_code="ALICE"))
|
|
# 2: paid subscriber via Stripe
|
|
s.add(User(id=2, email="bob@x", tier="paid", created_at=now,
|
|
last_login_at=now, stripe_customer_id="cus_1",
|
|
stripe_subscription_id="sub_1"))
|
|
# 3: free but credit-active, referred by alice, has sync/history
|
|
s.add(User(id=3, email="carol@x", tier="free", created_at=now,
|
|
credit_until=now + timedelta(days=30),
|
|
referred_by_user_id=1))
|
|
await s.flush()
|
|
s.add(PortfolioSync(user_id=3, outer_ciphertext=b"x",
|
|
outer_nonce=b"y", version=1,
|
|
created_at=now, updated_at=now))
|
|
s.add(EmailSend(user_id=3, kind="daily", sent_at=now, status="sent"))
|
|
s.add(UserAcknowledgement(user_id=3, version=2, lang="en",
|
|
accepted_at=now))
|
|
s.add(StrategicLogFeedback(log_id=1, user_id=3, vote="up",
|
|
created_at=now, updated_at=now))
|
|
s.add(Referral(referrer_user_id=1, referred_user_id=3,
|
|
created_at=now, converted_at=now))
|
|
await s.commit()
|
|
|
|
asyncio.run(_go())
|
|
return factory
|
|
|
|
|
|
# --- auth ------------------------------------------------------------------
|
|
|
|
def test_check_password(monkeypatch):
|
|
from app.config import get_settings
|
|
from admin import auth
|
|
|
|
monkeypatch.setenv("ADMIN_CONSOLE_PASSWORD", "s3cret")
|
|
get_settings.cache_clear()
|
|
assert auth.check_password("s3cret") is True
|
|
assert auth.check_password("wrong") is False
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_check_password_empty_denies(monkeypatch):
|
|
from app.config import get_settings
|
|
from admin import auth
|
|
|
|
monkeypatch.setenv("ADMIN_CONSOLE_PASSWORD", "")
|
|
get_settings.cache_clear()
|
|
# Empty password must reject everything — console closed by default.
|
|
assert auth.check_password("") is False
|
|
assert auth.check_password("anything") is False
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_session_round_trip(monkeypatch):
|
|
from app.config import get_settings
|
|
from admin import auth
|
|
|
|
monkeypatch.setenv("ADMIN_CONSOLE_SESSION_SECRET", "unit-test-secret")
|
|
get_settings.cache_clear()
|
|
token = auth.sign_session()
|
|
assert auth.verify_session(token) is True
|
|
assert auth.verify_session("garbage") is False
|
|
get_settings.cache_clear()
|
|
|
|
|
|
# --- queries ---------------------------------------------------------------
|
|
|
|
def test_overview_stats(tmp_path):
|
|
factory = _seed(tmp_path)
|
|
from admin import queries
|
|
|
|
async def _go():
|
|
async with factory() as s:
|
|
return await queries.overview_stats(s)
|
|
|
|
stats = asyncio.run(_go())
|
|
assert stats["total_users"] == 3
|
|
assert stats["free"] == 2
|
|
assert stats["paid"] == 1
|
|
# bob (tier paid) + carol (credit) both count as paid-active.
|
|
assert stats["paid_active"] == 2
|
|
assert stats["credit_active"] == 1
|
|
assert stats["sync_enabled"] == 1
|
|
assert stats["referrals_total"] == 1
|
|
assert stats["referrals_converted"] == 1
|
|
|
|
|
|
def test_list_users(tmp_path):
|
|
factory = _seed(tmp_path)
|
|
from admin import queries
|
|
|
|
async def _go():
|
|
async with factory() as s:
|
|
return await queries.list_users(s)
|
|
|
|
rows, total = asyncio.run(_go())
|
|
assert total == 3
|
|
by_id = {r["id"]: r for r in rows}
|
|
assert by_id[2]["paid_active"] and by_id[2]["paid_source"] == "tier"
|
|
assert by_id[2]["on_stripe"] is True
|
|
assert by_id[3]["paid_active"] and by_id[3]["paid_source"] == "credit"
|
|
assert by_id[3]["has_sync"] is True
|
|
assert by_id[1]["referrals"] == 1
|
|
assert by_id[1]["paid_active"] is False
|
|
|
|
|
|
def test_list_users_search(tmp_path):
|
|
factory = _seed(tmp_path)
|
|
from admin import queries
|
|
|
|
async def _go():
|
|
async with factory() as s:
|
|
return await queries.list_users(s, q="bob")
|
|
|
|
rows, total = asyncio.run(_go())
|
|
assert total == 1 and rows[0]["email"] == "bob@x"
|
|
|
|
|
|
def test_user_detail(tmp_path):
|
|
factory = _seed(tmp_path)
|
|
from admin import queries
|
|
|
|
async def _go():
|
|
async with factory() as s:
|
|
return await queries.user_detail(s, 3)
|
|
|
|
d = asyncio.run(_go())
|
|
assert d is not None
|
|
assert d["user"].email == "carol@x"
|
|
assert d["paid"].active and d["paid"].source == "credit"
|
|
assert d["referred_by"]["email"] == "alice@x"
|
|
assert d["sync"]["enabled"] is True
|
|
assert d["feedback"].get("up") == 1
|
|
assert len(d["emails"]) == 1
|
|
assert len(d["acks"]) == 1
|
|
|
|
|
|
def test_user_detail_missing(tmp_path):
|
|
factory = _seed(tmp_path)
|
|
from admin import queries
|
|
|
|
async def _go():
|
|
async with factory() as s:
|
|
return await queries.user_detail(s, 999)
|
|
|
|
assert asyncio.run(_go()) is None
|
|
|
|
|
|
def test_db_stats_sqlite_fallback(tmp_path):
|
|
factory = _seed(tmp_path)
|
|
from admin import queries
|
|
|
|
async def _go():
|
|
async with factory() as s:
|
|
return await queries.db_stats(s)
|
|
|
|
stats = asyncio.run(_go())
|
|
names = {t["name"] for t in stats["tables"]}
|
|
assert "users" in names
|
|
users_row = next(t for t in stats["tables"] if t["name"] == "users")
|
|
assert users_row["rows"] == 3
|
|
|
|
|
|
# --- app wiring ------------------------------------------------------------
|
|
|
|
def _client(tmp_path, monkeypatch):
|
|
from fastapi.testclient import TestClient
|
|
from app.config import get_settings
|
|
|
|
_seed(tmp_path)
|
|
monkeypatch.setenv("ADMIN_CONSOLE_PASSWORD", "letmein")
|
|
get_settings.cache_clear()
|
|
from admin.main import app
|
|
return TestClient(app, follow_redirects=False)
|
|
|
|
|
|
def test_pages_require_login(tmp_path, monkeypatch):
|
|
client = _client(tmp_path, monkeypatch)
|
|
for path in ("/", "/users", "/users/1", "/db"):
|
|
r = client.get(path)
|
|
assert r.status_code == 303, path
|
|
assert r.headers["location"] == "/login"
|
|
get_settings_clear()
|
|
|
|
|
|
def test_login_flow(tmp_path, monkeypatch):
|
|
client = _client(tmp_path, monkeypatch)
|
|
|
|
# Wrong password → 401, no cookie.
|
|
r = client.post("/login", data={"password": "nope"})
|
|
assert r.status_code == 401
|
|
assert "admin_console_session" not in r.cookies
|
|
|
|
# Right password → 303 to / with a session cookie.
|
|
r = client.post("/login", data={"password": "letmein"})
|
|
assert r.status_code == 303
|
|
assert r.headers["location"] == "/"
|
|
cookie = r.cookies.get("admin_console_session")
|
|
assert cookie
|
|
|
|
# Authenticated pages now load.
|
|
r = client.get("/", cookies={"admin_console_session": cookie})
|
|
assert r.status_code == 200
|
|
assert "Overview" in r.text
|
|
|
|
r = client.get("/users", cookies={"admin_console_session": cookie})
|
|
assert r.status_code == 200
|
|
assert "carol@x" in r.text
|
|
|
|
r = client.get("/users/3", cookies={"admin_console_session": cookie})
|
|
assert r.status_code == 200
|
|
assert "carol@x" in r.text and "alice@x" in r.text
|
|
|
|
r = client.get("/db", cookies={"admin_console_session": cookie})
|
|
assert r.status_code == 200
|
|
assert "users" in r.text
|
|
get_settings_clear()
|
|
|
|
|
|
def test_healthz(tmp_path, monkeypatch):
|
|
client = _client(tmp_path, monkeypatch)
|
|
r = client.get("/healthz")
|
|
assert r.status_code == 200 and r.text == "ok"
|
|
get_settings_clear()
|
|
|
|
|
|
def get_settings_clear():
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|