92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
"""Testumgebung: eigene Datenbank, kein Netz.
|
|
|
|
Die Umgebung steht, bevor app.* importiert wird — config.py liest DB_PATH
|
|
beim Import. Die Schlüssel prüft config.py nur auf Vorhandensein.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
_ORDNER = Path(tempfile.mkdtemp(prefix="filmabend-tests-"))
|
|
os.environ["DB_PATH"] = str(_ORDNER / "test.db")
|
|
os.environ.setdefault("TMDB_TOKEN", "test")
|
|
os.environ.setdefault("MINIMAX_KEY", "test")
|
|
|
|
import pytest # noqa: E402
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
from app import db, features, main, suggestions, users, watched, wishlist # noqa: E402
|
|
|
|
# Zählt, ob der Startup doch einen Sync anstoßen wollte
|
|
_SYNC_AUFRUFE = []
|
|
|
|
|
|
async def kein_sync():
|
|
_SYNC_AUFRUFE.append(True)
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def app_start():
|
|
"""Startup einmal je Lauf. last_sync steht vorher, damit main.startup
|
|
keinen TMDB-Sync startet; guarded_sync ist zur Sicherheit ersetzt."""
|
|
db.init()
|
|
with db.connect() as conn:
|
|
db.set_meta(conn, "last_sync", "test")
|
|
main.guarded_sync = kein_sync
|
|
with TestClient(main.app):
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def sync_aufrufe():
|
|
return _SYNC_AUFRUFE
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def frische_db(app_start):
|
|
"""Jeder Test beginnt mit leeren Tabellen und ohne laufende Ziehung."""
|
|
db.DB_PATH.unlink(missing_ok=True)
|
|
for modul in (db, users, suggestions, watched, wishlist, features):
|
|
modul.init()
|
|
main.letzter_dreh.update(ergebnis=None, start=0.0)
|
|
|
|
|
|
# Ein Client je Person — jeder hält sein eigenes Cookie
|
|
@pytest.fixture
|
|
def client():
|
|
return TestClient(main.app)
|
|
|
|
|
|
@pytest.fixture
|
|
def client2():
|
|
return TestClient(main.app)
|
|
|
|
|
|
@pytest.fixture
|
|
def film():
|
|
"""Testfilm direkt in der Datenbank, ohne TMDB. Gibt die id zurück."""
|
|
|
|
def anlegen(movie_id, title=None):
|
|
with db.connect() as conn:
|
|
db.upsert(conn, {"id": movie_id, "title": title or f"Film {movie_id}"})
|
|
return movie_id
|
|
|
|
return anlegen
|
|
|
|
|
|
@pytest.fixture
|
|
def anmelden():
|
|
"""Client wählt einen Namen; legt ihn an, falls es ihn noch nicht gibt."""
|
|
|
|
def waehlen(client, name):
|
|
antwort = client.post("/api/users", json={"name": name})
|
|
if antwort.status_code == 200:
|
|
user_id = antwort.json()["id"]
|
|
else:
|
|
runde = client.get("/api/users").json()["items"]
|
|
user_id = next(u["id"] for u in runde if u["name"] == name)
|
|
return client.post("/api/users/waehlen", json={"user_id": user_id}).json()["mine"]
|
|
|
|
return waehlen
|