feat: Intervals.icu MCP portal (FastAPI + Authentik OIDC)
- OIDC login via Authentik; first login creates a disabled user (admin approval). - Users set/replace their Intervals.icu athlete ID + API key; the key is validated against Intervals.icu and stored AES-256-GCM encrypted (shared key with the MCP server). Same users table (schema owned by the MCP server's migrations). - Admin page (gated on the intervalsicu-mcp-admins group claim): list, approve, disable, delete users. - 29 tests @ 93% (OIDC routes integration-only); Dockerfile asserts templates are packaged; .gitea CI test-gates the image build. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,132 @@
|
||||
"""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_group="intervalsicu-mcp-admins",
|
||||
intervals_api_base="https://intervals.icu/api/v1",
|
||||
)
|
||||
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
|
||||
Reference in New Issue
Block a user