Chris Farhood
c6c602fd20
Drops Authentik OIDC + the groups claim. Public PKCE client (Better Auth hashes confidential secrets), scopes openid/email/profile, admin determined by ADMIN_EMAILS.
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""Route tests via TestClient (OIDC bypassed with a dependency override)."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from intervalsicu_mcp_ui import app as appmod
|
|
from intervalsicu_mcp_ui import crypto, db
|
|
from intervalsicu_mcp_ui.config import Config
|
|
|
|
KEY = crypto.generate_key_b64()
|
|
USER = {"sub": "sub-user", "email": "user@x.com", "name": "User", "is_admin": False}
|
|
ADMIN = {"sub": "sub-admin", "email": "admin@x.com", "name": "Admin", "is_admin": True}
|
|
|
|
|
|
@pytest.fixture
|
|
def app_url(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("INTERVALS_ENC_KEY", KEY)
|
|
url = f"sqlite+aiosqlite:///{tmp_path / 'ui.db'}"
|
|
|
|
async def init():
|
|
engine = create_async_engine(url)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(db.Base.metadata.create_all)
|
|
maker = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with maker() as s:
|
|
await db.upsert_login(s, "sub-user", "user@x.com", "User")
|
|
await db.upsert_login(s, "sub-admin", "admin@x.com", "Admin")
|
|
await engine.dispose()
|
|
|
|
asyncio.run(init())
|
|
cfg = Config(
|
|
database_url=url,
|
|
oidc_issuer="https://auth.example/application/o/x/",
|
|
oidc_client_id="cid",
|
|
oidc_client_secret="sec",
|
|
oidc_redirect_url="https://ui.example/auth/callback",
|
|
session_secret="x" * 32,
|
|
admin_emails=frozenset({"admin@x.com"}),
|
|
intervals_api_base="https://intervals.icu/api/v1",
|
|
oidc_scopes="openid email profile",
|
|
root_path="",
|
|
mcp_url="https://intervalsicu.farhoodlabs.com/mcp",
|
|
)
|
|
return appmod.create_app(cfg), url
|
|
|
|
|
|
def _as(app, user):
|
|
app.dependency_overrides[appmod.current_user] = lambda: user
|
|
|
|
|
|
def _fetch_user(url, sub):
|
|
async def go():
|
|
engine = create_async_engine(url)
|
|
async with async_sessionmaker(engine, expire_on_commit=False)() as s:
|
|
u = await db.get_user(s, sub)
|
|
# detach copy of the bits we assert on
|
|
data = None if u is None else {"athlete_id": u.athlete_id, "enc": u.api_key_enc, "enabled": u.enabled}
|
|
await engine.dispose()
|
|
return data
|
|
|
|
return asyncio.run(go())
|
|
|
|
|
|
def test_health(app_url):
|
|
app, _ = app_url
|
|
assert TestClient(app).get("/healthz").json() == {"status": "ok"}
|
|
|
|
|
|
def test_unauthenticated_account_redirects_to_login(app_url):
|
|
app, _ = app_url
|
|
_as(app, None)
|
|
r = TestClient(app).get("/account", follow_redirects=False)
|
|
assert r.status_code in (302, 307) and r.headers["location"] == "/login"
|
|
|
|
|
|
def test_account_shows_pending_for_disabled_user(app_url):
|
|
app, _ = app_url
|
|
_as(app, USER)
|
|
assert "pending admin approval" in TestClient(app).get("/account").text.lower()
|
|
|
|
|
|
def test_save_valid_credentials_stores_encrypted(app_url, monkeypatch):
|
|
app, url = app_url
|
|
_as(app, USER)
|
|
|
|
async def ok(_base, _aid, _key):
|
|
return True, "Connected as Chris."
|
|
|
|
monkeypatch.setattr(appmod, "validate_credentials", ok)
|
|
r = TestClient(app).post("/account", data={"athlete_id": "i123", "api_key": "secret"})
|
|
assert "Connected as Chris" in r.text
|
|
rec = _fetch_user(url, "sub-user")
|
|
assert rec["athlete_id"] == "i123"
|
|
assert b"secret" not in rec["enc"] and crypto.decrypt(rec["enc"]) == "secret"
|
|
|
|
|
|
def test_save_invalid_credentials_not_stored(app_url, monkeypatch):
|
|
app, url = app_url
|
|
_as(app, USER)
|
|
|
|
async def bad(_base, _aid, _key):
|
|
return False, "Invalid API key, or it doesn't have access to that athlete."
|
|
|
|
monkeypatch.setattr(appmod, "validate_credentials", bad)
|
|
r = TestClient(app).post("/account", data={"athlete_id": "i123", "api_key": "bad"})
|
|
assert "Invalid API key" in r.text
|
|
assert _fetch_user(url, "sub-user")["enc"] is None # nothing stored
|
|
|
|
|
|
def test_admin_page_forbidden_for_non_admin(app_url):
|
|
app, _ = app_url
|
|
_as(app, USER)
|
|
r = TestClient(app).get("/admin", follow_redirects=False)
|
|
assert r.headers["location"] == "/account"
|
|
|
|
|
|
def test_admin_lists_and_approves(app_url):
|
|
app, url = app_url
|
|
_as(app, ADMIN)
|
|
client = TestClient(app)
|
|
listing = client.get("/admin").text
|
|
assert "user@x.com" in listing and "admin@x.com" in listing
|
|
r = client.post("/admin/sub-user/enable", follow_redirects=False)
|
|
assert r.status_code == 303
|
|
assert _fetch_user(url, "sub-user")["enabled"] is True
|
|
|
|
|
|
def test_admin_delete(app_url):
|
|
app, url = app_url
|
|
_as(app, ADMIN)
|
|
TestClient(app).post("/admin/sub-user/delete", follow_redirects=False)
|
|
assert _fetch_user(url, "sub-user") is None
|