99admandClaude Opus 5.5
35985fe66d
Statt des zufälligen Links kann der Admin einen sprechenden Link wählen (z. B. /z/sommerfest2026, mind. 6 Zeichen: A–Z, a–z, 0–9, - und _). Alte Zugänge werden dabei ungültig; der QR-Code passt sich an. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
252 lines
12 KiB
Python
252 lines
12 KiB
Python
"""Tests für den Admin-Bereich: Login, Sitzung, CSRF und alle Steuerfunktionen."""
|
||
|
||
import pytest
|
||
from argon2 import PasswordHasher
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app import admin, config, db, gaeste
|
||
from app.spotify import spotify
|
||
from tests.test_gaeste import URI, FakeSpotify, sql
|
||
|
||
PASSWORT = "richtig-langes-passwort"
|
||
|
||
|
||
@pytest.fixture
|
||
def client(tmp_path, monkeypatch):
|
||
env = tmp_path / ".env"
|
||
env.write_text(f"DB_PATH={tmp_path / 'playwish.db'}\nSESSION_SECRET=geheim\nCOOKIE_SECURE=0\n"
|
||
f"PUBLIC_URL=https://musik.example.de\nADMIN_PASSWORD_HASH={PasswordHasher().hash(PASSWORT)}\n")
|
||
monkeypatch.setenv("PLAYWISH_ENV", str(env))
|
||
config.konfig.cache_clear()
|
||
for name in ("suche_drossel", "wunsch_drossel", "einladung_drossel"):
|
||
monkeypatch.setattr(gaeste, name, gaeste.Drossel(1000, 60))
|
||
monkeypatch.setattr(admin, "login_drossel", gaeste.Drossel(5, 900))
|
||
from app.main import app
|
||
app.dependency_overrides[spotify] = FakeSpotify
|
||
with TestClient(app) as c:
|
||
yield c
|
||
app.dependency_overrides.clear()
|
||
config.konfig.cache_clear()
|
||
|
||
|
||
def anmelden(client) -> dict:
|
||
assert client.post("/admin/login", json={"password": PASSWORT}).status_code == 200
|
||
return {"X-CSRF-Token": client.get("/admin/api/state").json()["csrf"]}
|
||
|
||
|
||
def einstellung(key):
|
||
conn = db.verbinden()
|
||
wert = db.einstellungen(conn)[key]
|
||
conn.close()
|
||
return wert
|
||
|
||
|
||
def wunsch(status="queued", position=1.0, priority=0, guest_id=None, guest_ip=None, k="A"):
|
||
return sql("INSERT INTO wishes (track_uri, title, artist, status, priority, position, guest_id, guest_ip, created_at) "
|
||
"VALUES (?, ?, 'X', ?, ?, ?, ?, ?, ?) RETURNING id",
|
||
URI[k], f"Lied {k}", status, priority, position, guest_id, guest_ip, db.jetzt())[0]["id"]
|
||
|
||
|
||
# --- Zugang ----------------------------------------------------------------------
|
||
|
||
def test_ohne_login_gesperrt(client):
|
||
assert client.get("/admin", follow_redirects=False).headers["location"] == "/admin/login"
|
||
assert client.get("/admin/api/state").status_code == 401
|
||
assert client.post("/admin/api/command", json={"type": "skip"}).status_code == 401
|
||
|
||
|
||
def test_login_falsch_und_drossel(client):
|
||
for _ in range(5):
|
||
assert client.post("/admin/login", json={"password": "falsch"}).status_code == 401
|
||
assert client.post("/admin/login", json={"password": PASSWORT}).status_code == 429 # 6. Versuch
|
||
|
||
|
||
def test_login_cookie_und_seite(client):
|
||
r = client.post("/admin/login", json={"password": PASSWORT})
|
||
cookie = r.headers["set-cookie"].lower()
|
||
assert "httponly" in cookie and "samesite=strict" in cookie
|
||
assert client.get("/admin").status_code == 200
|
||
assert client.get("/admin/login", follow_redirects=False).headers["location"] == "/admin"
|
||
# in der Datenbank steht nur der Hash
|
||
wert = client.cookies.get(admin.ADMIN_COOKIE)
|
||
assert not sql("SELECT 1 FROM admin_sessions WHERE id = ?", wert)
|
||
|
||
|
||
def test_csrf_pflicht(client):
|
||
anmelden(client)
|
||
assert client.post("/admin/api/command", json={"type": "skip"}).status_code == 403
|
||
assert client.post("/admin/api/command", json={"type": "skip"},
|
||
headers={"X-CSRF-Token": "falsch"}).status_code == 403
|
||
|
||
|
||
def test_logout(client):
|
||
h = anmelden(client)
|
||
assert client.post("/admin/api/logout", headers=h).status_code == 200
|
||
assert client.get("/admin/api/state").status_code == 401
|
||
|
||
|
||
def test_kein_passwort_gesetzt(client, tmp_path, monkeypatch):
|
||
env = tmp_path / ".env"
|
||
env.write_text(env.read_text().replace("ADMIN_PASSWORD_HASH", "#"))
|
||
config.konfig.cache_clear()
|
||
assert client.post("/admin/login", json={"password": "egal"}).status_code == 503
|
||
|
||
|
||
# --- Wiedergabe ---------------------------------------------------------------------
|
||
|
||
def test_befehle_gehen_an_den_pi(client):
|
||
h = anmelden(client)
|
||
assert client.post("/admin/api/command", json={"type": "skip"}, headers=h).status_code == 200
|
||
assert client.post("/admin/api/command", json={"type": "volume", "percent": 40}, headers=h).status_code == 200
|
||
assert client.post("/admin/api/command", json={"type": "volume"}, headers=h).status_code == 422
|
||
assert client.post("/admin/api/command", json={"type": "rm -rf"}, headers=h).status_code == 422
|
||
befehle = sql("SELECT type, payload FROM commands ORDER BY id")
|
||
assert [(b["type"], b["payload"]) for b in befehle] == [("skip", "{}"), ("volume", '{"percent": 40}')]
|
||
assert [b["status"] for b in client.get("/admin/api/state").json()["commands"]] == ["wartet", "wartet"]
|
||
|
||
|
||
def test_not_aus(client):
|
||
h = anmelden(client)
|
||
client.post("/admin/api/emergency", json={"stop": True}, headers=h)
|
||
assert einstellung("emergency_stop") is True
|
||
client.post("/admin/api/emergency", json={"stop": False}, headers=h)
|
||
assert einstellung("emergency_stop") is False
|
||
assert [r["type"] for r in sql("SELECT type FROM commands ORDER BY id")] == ["pause", "play"]
|
||
|
||
|
||
def test_als_naechstes_vor_allen_wuenschen(client):
|
||
h = anmelden(client)
|
||
wunsch(position=1, k="A")
|
||
assert client.post("/admin/api/play-next", json={"track_uri": URI["B"]}, headers=h).status_code == 200
|
||
queue = client.get("/admin/api/state").json()["queue"]
|
||
assert [(w["title"], w["priority"]) for w in queue] == [("Lied B", 1), ("Lied A", 0)]
|
||
|
||
|
||
# --- Wünsche ------------------------------------------------------------------------
|
||
|
||
def test_freigabe_ablehnen_loeschen(client):
|
||
h = anmelden(client)
|
||
a, b, c = wunsch("pending", k="A"), wunsch("pending", k="B"), wunsch("queued", k="C")
|
||
client.post(f"/admin/api/wish/{a}/approve", headers=h)
|
||
client.post(f"/admin/api/wish/{b}/reject", headers=h)
|
||
client.post(f"/admin/api/wish/{c}/delete", headers=h)
|
||
status = {r["id"]: r["status"] for r in sql("SELECT id, status FROM wishes")}
|
||
assert status == {a: "queued", b: "rejected", c: "deleted"}
|
||
assert client.post("/admin/api/wish/999/approve", headers=h).status_code == 404
|
||
|
||
|
||
def test_umsortieren(client):
|
||
h = anmelden(client)
|
||
a, b, c = wunsch(position=1, k="A"), wunsch(position=2, k="B"), wunsch(position=3, k="C")
|
||
client.post(f"/admin/api/wish/{c}/up", headers=h)
|
||
client.post(f"/admin/api/wish/{a}/down", headers=h)
|
||
reihenfolge = [w["id"] for w in client.get("/admin/api/state").json()["queue"]]
|
||
assert reihenfolge == [c, a, b]
|
||
client.post(f"/admin/api/wish/{c}/up", headers=h) # ganz oben – bleibt
|
||
assert [w["id"] for w in client.get("/admin/api/state").json()["queue"]] == [c, a, b]
|
||
|
||
|
||
def test_gast_sperren(client):
|
||
h = anmelden(client)
|
||
a = wunsch(guest_id="gast1", guest_ip="hash1", k="A")
|
||
b = wunsch(guest_id="gast1", guest_ip="hash1", k="B", position=2)
|
||
andere = wunsch(guest_id="gast2", guest_ip="hash2", k="C", position=3)
|
||
client.post(f"/admin/api/wish/{a}/block-guest", headers=h)
|
||
sperren = {(r["kind"], r["value"]) for r in sql("SELECT kind, value FROM blocklist")}
|
||
assert sperren == {("guest", "gast1"), ("ip", "hash1")}
|
||
status = {r["id"]: r["status"] for r in sql("SELECT id, status FROM wishes")}
|
||
assert status == {a: "deleted", b: "deleted", andere: "queued"}
|
||
|
||
|
||
# --- Sperrliste & Einstellungen -------------------------------------------------------
|
||
|
||
def test_sperrliste(client):
|
||
h = anmelden(client)
|
||
assert client.post("/admin/api/blocklist", headers=h,
|
||
json={"kind": "artist", "value": "spotify:artist:" + "Z" * 22, "label": "Band Z"}).status_code == 200
|
||
assert client.post("/admin/api/blocklist", headers=h,
|
||
json={"kind": "track", "value": "spotify:artist:" + "Z" * 22, "label": "x"}).status_code == 422
|
||
eintraege = client.get("/admin/api/blocklist").json()["items"]
|
||
assert [(e["kind"], e["label"]) for e in eintraege] == [("artist", "Band Z")]
|
||
client.post(f"/admin/api/blocklist/{eintraege[0]['id']}/delete", headers=h)
|
||
assert client.get("/admin/api/blocklist").json()["items"] == []
|
||
|
||
|
||
def test_einstellungen_validiert(client):
|
||
h = anmelden(client)
|
||
assert client.post("/admin/api/settings", json={"wish_every_n": 4, "moderation": True}, headers=h).status_code == 200
|
||
assert (einstellung("wish_every_n"), einstellung("moderation")) == (4, True)
|
||
assert client.post("/admin/api/settings", json={"playlist_repeat_percent": 80}, headers=h).status_code == 200
|
||
assert einstellung("playlist_repeat_percent") == 80
|
||
for falsch in ({"wish_every_n": 0}, {"wish_every_n": "4"}, {"moderation": 1}, {"access_token": "x"},
|
||
{"emergency_stop": True}, {"playlist_repeat_percent": 101}, {"playlist_repeat_percent": -1}):
|
||
assert client.post("/admin/api/settings", json=falsch, headers=h).status_code == 422, falsch
|
||
|
||
|
||
def test_playlist_wechseln(client):
|
||
h = anmelden(client)
|
||
link = "https://open.spotify.com/playlist/37i9dQZF1DXcBWIGoYBM5M?si=abc"
|
||
r = client.post("/admin/api/playlist", json={"link": link}, headers=h)
|
||
assert r.json()["playlist_id"] == "37i9dQZF1DXcBWIGoYBM5M"
|
||
assert einstellung("reshuffle_token") == 1
|
||
assert client.post("/admin/api/playlist", json={"link": "quatsch"}, headers=h).status_code == 422
|
||
client.post("/admin/api/reshuffle", headers=h)
|
||
assert einstellung("reshuffle_token") == 2
|
||
|
||
|
||
# --- Gastzugang -----------------------------------------------------------------------
|
||
|
||
def test_gastzugang_und_qr(client):
|
||
h = anmelden(client)
|
||
z = client.get("/admin/api/access").json()
|
||
assert z["enabled"] is False and z["link"].startswith("https://musik.example.de/z/")
|
||
client.post("/admin/api/access", json={"enabled": True}, headers=h)
|
||
assert einstellung("access_enabled") is True
|
||
alt = z["link"]
|
||
client.post("/admin/api/access/renew", headers=h)
|
||
assert client.get("/admin/api/access").json()["link"] != alt
|
||
svg = client.get("/admin/api/access/qr")
|
||
assert svg.headers["content-type"].startswith("image/svg+xml") and b"<svg" in svg.content
|
||
png = client.get("/admin/api/access/qr?format=png")
|
||
assert png.content[:8] == b"\x89PNG\r\n\x1a\n" and "attachment" in png.headers["content-disposition"]
|
||
|
||
|
||
def test_eigener_einladungslink(client):
|
||
h = anmelden(client)
|
||
assert client.post("/admin/api/access/token", json={"token": "sommerfest2026"}, headers=h).status_code == 200
|
||
z = client.get("/admin/api/access").json()
|
||
assert z["link"] == "https://musik.example.de/z/sommerfest2026" and z["token"] == "sommerfest2026"
|
||
assert z["base"] == "https://musik.example.de/z/"
|
||
for falsch in ("kurz", "mit leerzeichen", "../../etc", "ümlaut1", "x" * 65, ""):
|
||
assert client.post("/admin/api/access/token", json={"token": falsch}, headers=h).status_code == 422, falsch
|
||
assert client.post("/admin/api/access/token", json={"token": "sommerfest2026"}).status_code == 403 # ohne CSRF
|
||
# Gäste kommen mit dem eigenen Link hinein, mit dem alten zufälligen nicht mehr
|
||
client.post("/admin/api/access", json={"enabled": True}, headers=h)
|
||
gast = TestClient(client.app)
|
||
assert gast.get("/z/sommerfest2026", follow_redirects=False).status_code == 303
|
||
assert gast.get("/api/status").status_code == 200
|
||
assert gast.get("/z/Sommerfest2026", follow_redirects=False).status_code == 403 # Groß/klein zählt
|
||
|
||
|
||
def test_verlauf(client):
|
||
anmelden(client)
|
||
w = wunsch(k="A")
|
||
sql("UPDATE wishes SET guest_name = 'Erna' WHERE id = ?", w)
|
||
sql("INSERT INTO history (track_uri, title, artist, source, wish_id, started_at) VALUES (?, 'Lied A', 'X', 'wish', ?, ?)",
|
||
URI["A"], w, db.jetzt())
|
||
eintrag = client.get("/admin/api/history").json()["items"][0]
|
||
assert (eintrag["title"], eintrag["source"], eintrag["guest_name"]) == ("Lied A", "wish", "Erna")
|
||
|
||
|
||
def test_lautsprecher_befehle(client):
|
||
h = anmelden(client)
|
||
assert client.post("/admin/api/command", json={"type": "bt_scan"}, headers=h).status_code == 200
|
||
assert client.post("/admin/api/command", json={"type": "bt_connect"}, headers=h).status_code == 422
|
||
assert client.post("/admin/api/command", json={"type": "bt_connect", "mac": "falsch"}, headers=h).status_code == 422
|
||
assert client.post("/admin/api/command", json={"type": "bt_connect", "mac": "AA:BB:CC:DD:EE:FF"},
|
||
headers=h).status_code == 200
|
||
befehle = sql("SELECT type, payload FROM commands ORDER BY id")
|
||
assert [(b["type"], b["payload"]) for b in befehle] == [("bt_scan", "{}"),
|
||
("bt_connect", '{"mac": "AA:BB:CC:DD:EE:FF"}')]
|
||
assert "bluetooth" in client.get("/admin/api/state").json()["player"]
|