173 lines
7.2 KiB
Python
173 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "test.db"
|
|
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{db_path}")
|
|
monkeypatch.setenv("SESSION_SECRET", "test-secret")
|
|
monkeypatch.setenv("DEV_MODE", "true")
|
|
monkeypatch.setenv("UPLOAD_DIR", str(tmp_path / "uploads"))
|
|
monkeypatch.setenv("FRONTEND_ORIGIN", "http://testserver")
|
|
monkeypatch.setenv("SERVER_ORIGIN", "http://testserver")
|
|
monkeypatch.setenv("API_BASE_URL", "http://testserver/api")
|
|
|
|
from app.core.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
import app.db.base as base
|
|
from app.core.config import get_settings as refreshed_settings
|
|
from app.db.seed import seed
|
|
from app.main import app
|
|
|
|
settings = refreshed_settings()
|
|
base.settings = settings
|
|
base.engine.dispose()
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
base.engine = create_engine(settings.database_url, connect_args={"check_same_thread": False}, future=True)
|
|
base.SessionLocal = sessionmaker(bind=base.engine, autoflush=False, autocommit=False, expire_on_commit=False, future=True)
|
|
seed()
|
|
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
|
|
|
|
def test_invite_is_hashed_and_claim_creates_session(client):
|
|
from app.core.security import hash_token
|
|
from app.db.base import SessionLocal
|
|
from app.db.seed import DEMO_INVITE_TOKEN
|
|
from app.models import MemberInvite
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
invite = db.scalar(select(MemberInvite).where(MemberInvite.token_hash == hash_token(DEMO_INVITE_TOKEN)))
|
|
assert invite is not None
|
|
assert invite.token_hash != DEMO_INVITE_TOKEN
|
|
finally:
|
|
db.close()
|
|
|
|
preview = client.get(f"/api/join/{DEMO_INVITE_TOKEN}/preview")
|
|
assert preview.status_code == 200
|
|
assert preview.json()["group"]["name"] == "FC Kreuzberg U12 Parents"
|
|
|
|
claimed = client.post(f"/api/auth/invite/{DEMO_INVITE_TOKEN}/claim", json={"display_name": "New Parent", "device_label": "Pytest browser"})
|
|
assert claimed.status_code == 200
|
|
assert "grouphome_session" in claimed.cookies
|
|
assert claimed.json()["member"]["status"] == "joined"
|
|
|
|
|
|
def test_limited_invite_blocks_after_max_use(client):
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
groups = client.get("/api/groups").json()["groups"]
|
|
group_id = groups[0]["group"]["id"]
|
|
created = client.post(f"/api/groups/{group_id}/invites", json={"label": "Once", "max_uses": 1})
|
|
token = created.json()["token_display_once"]
|
|
|
|
first = TestClient(client.app)
|
|
assert first.post(f"/api/auth/invite/{token}/claim", json={"display_name": "First", "device_label": "Browser"}).status_code == 200
|
|
second = TestClient(client.app)
|
|
assert second.post(f"/api/auth/invite/{token}/claim", json={"display_name": "Second", "device_label": "Browser"}).status_code == 404
|
|
|
|
|
|
def test_home_dashboard_contains_required_sections_and_actions(client):
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
response = client.get("/api/home")
|
|
assert response.status_code == 200
|
|
sections = response.json()["sections"]
|
|
assert set(sections) == {"needs_me", "today", "changed", "official_updates", "catch_up"}
|
|
types = {item["type"] for item in sections["needs_me"]}
|
|
assert {"rsvp_required", "task_assigned", "vote_required"}.issubset(types)
|
|
|
|
|
|
def test_permissions_and_rsvp_flow(client):
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
groups = client.get("/api/groups").json()["groups"]
|
|
school = next(item for item in groups if item["group"]["name"] == "Class 4B Parents")
|
|
group_id = school["group"]["id"]
|
|
denied = client.post(f"/api/groups/{group_id}/announcements", json={"title": "Official", "body": "No", "official": True})
|
|
assert denied.status_code == 403
|
|
|
|
fc = next(item for item in groups if item["group"]["name"] == "FC Kreuzberg U12 Parents")
|
|
event = next(item for item in fc["dashboard"]["upcoming"] if item["rsvp_required"])
|
|
rsvp = client.post(f"/api/events/{event['id']}/rsvp", json={"status": "yes"})
|
|
assert rsvp.status_code == 200
|
|
home = client.get("/api/home").json()
|
|
assert event["id"] not in [item["object_id"] for item in home["sections"]["needs_me"] if item["type"] == "rsvp_required"]
|
|
|
|
|
|
def test_device_link_flow(client):
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
start = client.post("/api/auth/device-link/start", json={"device_label": "Laptop"})
|
|
code = start.json()["code"]
|
|
assert client.post("/api/auth/device-link/approve", json={"code": code}).status_code == 200
|
|
other = TestClient(client.app)
|
|
complete = other.post("/api/auth/device-link/complete", json={"code": code, "device_label": "Laptop"})
|
|
assert complete.status_code == 200
|
|
devices = client.get("/api/me/devices").json()["devices"]
|
|
assert len(devices) >= 2
|
|
|
|
|
|
def test_recovery_flow(client):
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
requested = client.post("/api/auth/recovery/request", json={"email": "anna@example.test"})
|
|
code = requested.json()["dev_code"]
|
|
other = TestClient(client.app)
|
|
consumed = other.post("/api/auth/recovery/consume", json={"recovery_code": code, "device_label": "Recovered"})
|
|
assert consumed.status_code == 200
|
|
|
|
|
|
def test_remote_manifest_token_and_sync(client):
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
manifest = client.get("/.well-known/group-platform.json")
|
|
assert manifest.status_code == 200
|
|
assert manifest.json()["capabilities"]["federation"] is False
|
|
|
|
token_response = client.post("/api/connection-tokens", json={"label": "Test sync", "scopes": ["sync:read"]})
|
|
raw = token_response.json()["connection_code_display_once"]
|
|
denied = client.get("/api/sync")
|
|
assert denied.status_code == 401
|
|
synced = client.get("/api/sync", headers={"Authorization": f"Bearer {raw}"})
|
|
assert synced.status_code == 200
|
|
assert synced.json()["events"]
|
|
|
|
|
|
def test_chat_home_and_message_metadata(client):
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
chat = client.get("/api/chat")
|
|
assert chat.status_code == 200
|
|
payload = chat.json()
|
|
assert payload["groups"]
|
|
assert payload["threads"]
|
|
thread_id = payload["active_thread"]["id"]
|
|
|
|
sent = client.post(f"/api/chat/threads/{thread_id}/messages", json={"body": "ok"})
|
|
assert sent.status_code == 200
|
|
message = sent.json()["message"]
|
|
assert message["author_name"] == "Anna Müller"
|
|
assert message["mine"] is True
|
|
assert message["low_signal"] is True
|
|
|
|
|
|
def test_file_upload_requires_auth_and_sanitizes_filename(client, tmp_path):
|
|
no_auth = TestClient(client.app)
|
|
assert no_auth.post("/api/groups/not-real/files", files={"upload": ("bad.txt", b"x", "text/plain")}).status_code == 401
|
|
|
|
client.post("/api/auth/dev/demo-session", json={})
|
|
group_id = client.get("/api/groups").json()["groups"][0]["group"]["id"]
|
|
uploaded = client.post(
|
|
f"/api/groups/{group_id}/files",
|
|
data={"description": "test"},
|
|
files={"upload": ("../../unsafe name.txt", b"hello", "text/plain")},
|
|
)
|
|
assert uploaded.status_code == 200
|
|
assert uploaded.json()["file"]["filename_original"] == "unsafe name.txt"
|