Chris Farhood
bb5e9d518c
- 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>
61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
"""Store operations against in-memory SQLite (mirrors the MCP server's schema)."""
|
|
|
|
import asyncio
|
|
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from intervalsicu_mcp_ui import crypto, db
|
|
|
|
KEY = crypto.generate_key_b64()
|
|
|
|
|
|
def _run_db(monkeypatch, body):
|
|
async def go():
|
|
monkeypatch.setenv("INTERVALS_ENC_KEY", KEY)
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
poolclass=StaticPool,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(db.Base.metadata.create_all)
|
|
maker = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
await body(maker)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
asyncio.run(go())
|
|
|
|
|
|
def test_first_login_disabled_then_credentials_encrypted(monkeypatch):
|
|
async def body(maker):
|
|
async with maker() as s:
|
|
user = await db.upsert_login(s, "sub1", "a@b.com", "Alice")
|
|
assert user.enabled is False and user.has_credentials is False
|
|
assert await db.set_credentials(s, "sub1", "i123", "SECRET") is True
|
|
again = await db.get_user(s, "sub1")
|
|
assert again.athlete_id == "i123"
|
|
assert b"SECRET" not in again.api_key_enc
|
|
assert crypto.decrypt(again.api_key_enc) == "SECRET"
|
|
assert again.has_credentials is True
|
|
|
|
_run_db(monkeypatch, body)
|
|
|
|
|
|
def test_enable_disable_delete_and_list(monkeypatch):
|
|
async def body(maker):
|
|
async with maker() as s:
|
|
await db.upsert_login(s, "sub1", "a@b.com", None)
|
|
await db.upsert_login(s, "sub2", "b@b.com", None)
|
|
assert len(await db.list_users(s)) == 2
|
|
assert await db.set_enabled(s, "sub1", True) is True
|
|
assert (await db.get_user(s, "sub1")).enabled is True
|
|
assert await db.clear_credentials(s, "sub1") is True
|
|
assert await db.delete_user(s, "sub2") is True
|
|
assert await db.get_user(s, "sub2") is None
|
|
assert await db.delete_user(s, "ghost") is False
|
|
|
|
_run_db(monkeypatch, body)
|