ProfessorNovaandClaude Opus 5
784b76baf7
Self-hosted knowledge management for SMEs: a split-screen Markdown editor whose sections an LLM refines while you write, and RAG question answering over the documents that result. FastAPI + Postgres/pgvector on the back, SvelteKit on the front, everything OpenAI-compatible and self-hostable. Squashed into a single commit; the development history stays local. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CA43ZJda8Rbp2hKXNy8f6b
340 lines
11 KiB
Python
340 lines
11 KiB
Python
from httpx import AsyncClient
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import (
|
|
AuthSession,
|
|
Conversation,
|
|
ConversationMode,
|
|
Document,
|
|
DocumentStatus,
|
|
Message,
|
|
MessageRole,
|
|
User,
|
|
)
|
|
|
|
|
|
async def _login(client: AsyncClient, email: str) -> None:
|
|
response = await client.post(
|
|
"/api/auth/login", json={"email": email, "password": "secret123"}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
async def test_users_crud_requires_admin(
|
|
client: AsyncClient, seeded_user: User
|
|
) -> None:
|
|
await _login(client, "pablo@test.dev")
|
|
assert (await client.get("/api/admin/users")).status_code == 403
|
|
|
|
|
|
async def test_full_user_lifecycle(
|
|
client: AsyncClient, db: AsyncSession, seeded_admin: User
|
|
) -> None:
|
|
await _login(client, "florian@test.dev")
|
|
|
|
department = (
|
|
await client.post("/api/admin/departments", json={"name": "QS"})
|
|
).json()
|
|
created = await client.post(
|
|
"/api/admin/users",
|
|
json={
|
|
"email": "quinn@test.dev",
|
|
"name": "Quinn Test",
|
|
"role": "member",
|
|
"department_id": department["id"],
|
|
"password": "secret123",
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
user_id = created.json()["id"]
|
|
|
|
# Duplicate email → 409.
|
|
duplicate = await client.post(
|
|
"/api/admin/users",
|
|
json={"email": "quinn@test.dev", "name": "X", "password": "secret123"},
|
|
)
|
|
assert duplicate.status_code == 409
|
|
assert duplicate.json()["code"] == "email_taken"
|
|
|
|
# The new user can log in.
|
|
await client.post("/api/auth/logout")
|
|
await _login(client, "quinn@test.dev")
|
|
me = (await client.get("/api/auth/me")).json()
|
|
assert me["department_id"] == department["id"]
|
|
|
|
# Password reset by the admin invalidates the old password.
|
|
await client.post("/api/auth/logout")
|
|
await _login(client, "florian@test.dev")
|
|
reset = await client.patch(
|
|
f"/api/admin/users/{user_id}", json={"password": "new-secret-1"}
|
|
)
|
|
assert reset.status_code == 200
|
|
await client.post("/api/auth/logout")
|
|
old_login = await client.post(
|
|
"/api/auth/login", json={"email": "quinn@test.dev", "password": "secret123"}
|
|
)
|
|
assert old_login.status_code == 401
|
|
new_login = await client.post(
|
|
"/api/auth/login", json={"email": "quinn@test.dev", "password": "new-secret-1"}
|
|
)
|
|
assert new_login.status_code == 200
|
|
|
|
# Deleting the user keeps their documents, authorless.
|
|
document = Document(
|
|
title="Bleibt",
|
|
status=DocumentStatus.published,
|
|
visibility="public",
|
|
content_md="# Bleibt",
|
|
author_id=user_id,
|
|
)
|
|
db.add(document)
|
|
await db.commit()
|
|
|
|
await client.post("/api/auth/logout")
|
|
await _login(client, "florian@test.dev")
|
|
assert (await client.delete(f"/api/admin/users/{user_id}")).status_code == 204
|
|
db.expire_all()
|
|
survivor = (
|
|
await db.execute(select(Document).where(Document.title == "Bleibt"))
|
|
).scalar_one()
|
|
assert survivor.author_id is None
|
|
|
|
|
|
async def test_admin_cannot_delete_or_demote_self(
|
|
client: AsyncClient, seeded_admin: User
|
|
) -> None:
|
|
await _login(client, "florian@test.dev")
|
|
me = (await client.get("/api/auth/me")).json()
|
|
delete = await client.delete(f"/api/admin/users/{me['id']}")
|
|
assert delete.status_code == 409
|
|
assert delete.json()["code"] == "self_modification"
|
|
demote = await client.patch(f"/api/admin/users/{me['id']}", json={"role": "member"})
|
|
assert demote.status_code == 409
|
|
|
|
|
|
async def test_department_crud_and_conflicts(
|
|
client: AsyncClient, seeded_admin: User
|
|
) -> None:
|
|
await _login(client, "florian@test.dev")
|
|
created = (
|
|
await client.post("/api/admin/departments", json={"name": "Montage"})
|
|
).json()
|
|
|
|
conflict = await client.post("/api/admin/departments", json={"name": "Montage"})
|
|
assert conflict.status_code == 409
|
|
assert conflict.json()["code"] == "name_taken"
|
|
|
|
renamed = await client.patch(
|
|
f"/api/admin/departments/{created['id']}", json={"name": "Endmontage"}
|
|
)
|
|
assert renamed.json()["name"] == "Endmontage"
|
|
|
|
# Any authenticated user can list departments for pickers.
|
|
listing = (await client.get("/api/departments")).json()
|
|
assert "Endmontage" in [d["name"] for d in listing]
|
|
|
|
assert (
|
|
await client.delete(f"/api/admin/departments/{created['id']}")
|
|
).status_code == 204
|
|
|
|
|
|
async def test_deleting_a_department_in_use_needs_confirmation(
|
|
client: AsyncClient, seeded_admin: User
|
|
) -> None:
|
|
"""A department with members/documents/grants cannot be silently deleted:
|
|
the CASCADE would drop its shared-access grants unseen, so it takes an
|
|
explicit confirm."""
|
|
await _login(client, "florian@test.dev")
|
|
dept = (
|
|
await client.post("/api/admin/departments", json={"name": "Vertrieb"})
|
|
).json()
|
|
await client.post(
|
|
"/api/admin/users",
|
|
json={
|
|
"email": "neu@test.dev",
|
|
"name": "Neu",
|
|
"role": "member",
|
|
"department_id": dept["id"],
|
|
"password": "secret123",
|
|
},
|
|
)
|
|
|
|
blocked = await client.delete(f"/api/admin/departments/{dept['id']}")
|
|
assert blocked.status_code == 409
|
|
assert blocked.json()["code"] == "department_in_use"
|
|
|
|
confirmed = await client.delete(
|
|
f"/api/admin/departments/{dept['id']}", params={"confirm": "true"}
|
|
)
|
|
assert confirmed.status_code == 204
|
|
|
|
|
|
async def test_password_reset_revokes_all_sessions(
|
|
client: AsyncClient, db: AsyncSession, seeded_user: User, seeded_admin: User
|
|
) -> None:
|
|
# Pablo logs in — her session row exists and her cookie is captured.
|
|
await _login(client, "pablo@test.dev")
|
|
anna_cookie = client.cookies.get("pablan_session")
|
|
assert anna_cookie is not None
|
|
|
|
# Admin (same client, new cookie) resets Pablo's password.
|
|
await _login(client, "florian@test.dev")
|
|
reset = await client.patch(
|
|
f"/api/admin/users/{seeded_user.id}", json={"password": "brand-new-pw1"}
|
|
)
|
|
assert reset.status_code == 200
|
|
|
|
# Pablo's old session is dead ...
|
|
remaining = (
|
|
await db.execute(
|
|
select(func.count(AuthSession.id)).where(
|
|
AuthSession.user_id == seeded_user.id
|
|
)
|
|
)
|
|
).scalar_one()
|
|
assert remaining == 0
|
|
client.cookies.clear()
|
|
client.cookies.set("pablan_session", anna_cookie)
|
|
assert (await client.get("/api/auth/me")).status_code == 401
|
|
|
|
# ... while the acting admin's session survived.
|
|
await _login(client, "florian@test.dev")
|
|
assert (await client.get("/api/auth/me")).status_code == 200
|
|
|
|
|
|
async def test_admin_changing_own_password_logs_themselves_out(
|
|
client: AsyncClient, seeded_admin: User
|
|
) -> None:
|
|
await _login(client, "florian@test.dev")
|
|
me = (await client.get("/api/auth/me")).json()
|
|
reset = await client.patch(
|
|
f"/api/admin/users/{me['id']}", json={"password": "next-password-1"}
|
|
)
|
|
assert reset.status_code == 200
|
|
assert (await client.get("/api/auth/me")).status_code == 401
|
|
|
|
|
|
async def test_user_delete_cascades_conversations_but_not_documents(
|
|
client: AsyncClient, db: AsyncSession, seeded_user: User, seeded_admin: User
|
|
) -> None:
|
|
"""GDPR offboarding: transcripts are personal data and go with the user;
|
|
the approved document is the legitimate artifact and stays."""
|
|
conversation = Conversation(mode=ConversationMode.query, user_id=seeded_user.id)
|
|
db.add(conversation)
|
|
await db.flush()
|
|
db.add(
|
|
Message(
|
|
conversation_id=conversation.id,
|
|
role=MessageRole.user,
|
|
content="personal transcript",
|
|
)
|
|
)
|
|
document = Document(
|
|
title="Artefakt",
|
|
status=DocumentStatus.published,
|
|
visibility="public",
|
|
content_md="# Artefakt",
|
|
author_id=seeded_user.id,
|
|
)
|
|
db.add(document)
|
|
await db.commit()
|
|
|
|
await _login(client, "florian@test.dev")
|
|
assert (
|
|
await client.delete(f"/api/admin/users/{seeded_user.id}")
|
|
).status_code == 204
|
|
|
|
db.expire_all()
|
|
assert (await db.execute(select(func.count(Conversation.id)))).scalar_one() == 0
|
|
assert (await db.execute(select(func.count(Message.id)))).scalar_one() == 0
|
|
survivor = (
|
|
await db.execute(select(Document).where(Document.title == "Artefakt"))
|
|
).scalar_one()
|
|
assert survivor.author_id is None
|
|
|
|
|
|
async def test_admin_can_correct_a_users_email(
|
|
client: AsyncClient, db: AsyncSession, seeded_admin: User
|
|
) -> None:
|
|
"""A typo in an address used to mean deleting the person and losing
|
|
everything hanging off their id."""
|
|
await _login(client, "florian@test.dev")
|
|
created = (
|
|
await client.post(
|
|
"/api/admin/users",
|
|
json={
|
|
"email": "tpyo@pablan.dev",
|
|
"name": "Typo",
|
|
"role": "member",
|
|
"password": "secret123",
|
|
},
|
|
)
|
|
).json()
|
|
|
|
patched = await client.patch(
|
|
f"/api/admin/users/{created['id']}",
|
|
json={"email": " Fixed@Pablan.DEV ", "name": "Fixed"},
|
|
)
|
|
assert patched.status_code == 200
|
|
# Normalised exactly as creation does, or login would stop finding them.
|
|
assert patched.json()["email"] == "fixed@pablan.dev"
|
|
assert patched.json()["name"] == "Fixed"
|
|
|
|
login = await client.post(
|
|
"/api/auth/login", json={"email": "fixed@pablan.dev", "password": "secret123"}
|
|
)
|
|
assert login.status_code == 200
|
|
|
|
|
|
async def test_an_email_already_in_use_is_refused(
|
|
client: AsyncClient, db: AsyncSession, seeded_admin: User, seeded_user: User
|
|
) -> None:
|
|
await _login(client, "florian@test.dev")
|
|
users = (await client.get("/api/admin/users")).json()["items"]
|
|
target = next(u for u in users if u["email"] == "pablo@test.dev")
|
|
|
|
clash = await client.patch(
|
|
f"/api/admin/users/{target['id']}", json={"email": "florian@test.dev"}
|
|
)
|
|
assert clash.status_code == 409
|
|
assert clash.json()["code"] == "email_taken"
|
|
|
|
|
|
async def test_the_user_list_pages_and_searches(
|
|
client: AsyncClient, db: AsyncSession, seeded_admin: User, seeded_user: User
|
|
) -> None:
|
|
"""The admin screen is the one place that scales with headcount."""
|
|
await _login(client, "florian@test.dev")
|
|
for index in range(6):
|
|
await client.post(
|
|
"/api/admin/users",
|
|
json={
|
|
"email": f"kolleg{index}@pablan.dev",
|
|
"name": f"Kollege {index}",
|
|
"role": "member",
|
|
"password": "secret123",
|
|
},
|
|
)
|
|
|
|
first = (await client.get("/api/admin/users", params={"per_page": 3})).json()
|
|
assert len(first["items"]) == 3
|
|
assert first["total"] >= 8
|
|
second = (
|
|
await client.get("/api/admin/users", params={"per_page": 3, "page": 2})
|
|
).json()
|
|
assert {u["id"] for u in first["items"]}.isdisjoint(
|
|
u["id"] for u in second["items"]
|
|
)
|
|
|
|
# Search covers both columns, because an admin looking for someone knows
|
|
# one of the two and rarely which.
|
|
by_name = (
|
|
await client.get("/api/admin/users", params={"search": "Kollege 4"})
|
|
).json()
|
|
assert [u["email"] for u in by_name["items"]] == ["kolleg4@pablan.dev"]
|
|
by_email = (
|
|
await client.get("/api/admin/users", params={"search": "kolleg4@"})
|
|
).json()
|
|
assert by_email["total"] == 1
|