Audiofy – Musikwünsche für Partys
Server (Website, Admin-Bereich, Player-API) und Player (raspotify + Controller) mit Install-Skripten für Debian-Systeme. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,232 @@
|
||||
"""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 / 'audiofy.db'}\nSESSION_SECRET=geheim\nCOOKIE_SECURE=0\n"
|
||||
f"PUBLIC_URL=https://musik.example.de\nADMIN_PASSWORD_HASH={PasswordHasher().hash(PASSWORT)}\n")
|
||||
monkeypatch.setenv("AUDIOFY_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)
|
||||
for falsch in ({"wish_every_n": 0}, {"wish_every_n": "4"}, {"moderation": 1}, {"access_token": "x"},
|
||||
{"emergency_stop": True}):
|
||||
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_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"]
|
||||
@@ -0,0 +1,257 @@
|
||||
"""Tests für die Gastseite: Einladungszugang, Suche, Wünschen mit Schutzregeln, Status."""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app import config, db, gaeste
|
||||
from app.spotify import SpotifyFehler, spotify
|
||||
|
||||
URI = {k: "spotify:track:" + k * 22 for k in "ABCDE"}
|
||||
KUENSTLER_GESPERRT = "spotify:artist:" + "Z" * 22
|
||||
|
||||
|
||||
def titel(k, explicit=False, kuenstler="spotify:artist:" + "X" * 22):
|
||||
return {"track_uri": URI[k], "title": f"Lied {k}", "artist": f"Band {k}", "artist_uris": [kuenstler],
|
||||
"album": "Album", "cover_url": "https://i.scdn.co/image/x", "duration_ms": 200_000, "explicit": explicit}
|
||||
|
||||
|
||||
class FakeSpotify:
|
||||
def __init__(self):
|
||||
self.titel_db = {URI["A"]: titel("A"), URI["B"]: titel("B"), URI["C"]: titel("C", explicit=True),
|
||||
URI["D"]: titel("D", kuenstler=KUENSTLER_GESPERRT), URI["E"]: titel("E")}
|
||||
|
||||
def verfuegbar(self):
|
||||
return True
|
||||
|
||||
def suche(self, q):
|
||||
return list(self.titel_db.values())
|
||||
|
||||
def titel(self, uri):
|
||||
if uri not in self.titel_db:
|
||||
raise SpotifyFehler(404)
|
||||
return dict(self.titel_db[uri])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path, monkeypatch):
|
||||
env = tmp_path / ".env"
|
||||
env.write_text(f"DB_PATH={tmp_path / 'audiofy.db'}\nSESSION_SECRET=geheim\nCOOKIE_SECURE=0\n")
|
||||
monkeypatch.setenv("AUDIOFY_ENV", str(env))
|
||||
config.konfig.cache_clear()
|
||||
for name in ("suche_drossel", "wunsch_drossel", "einladung_drossel"):
|
||||
monkeypatch.setattr(gaeste, name, gaeste.Drossel(1000, 60))
|
||||
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 setzen(**werte):
|
||||
conn = db.verbinden()
|
||||
for k, v in werte.items():
|
||||
db.einstellung_setzen(conn, k, v)
|
||||
conn.close()
|
||||
|
||||
|
||||
def token():
|
||||
conn = db.verbinden()
|
||||
t = db.einstellungen(conn)["access_token"]
|
||||
conn.close()
|
||||
return t
|
||||
|
||||
|
||||
def eintreten(client):
|
||||
"""Zugang einschalten und Einladungslink öffnen."""
|
||||
setzen(access_enabled=True)
|
||||
r = client.get(f"/z/{token()}", follow_redirects=False)
|
||||
assert r.status_code == 303 and r.headers["location"] == "/"
|
||||
return client
|
||||
|
||||
|
||||
def sql(abfrage, *params):
|
||||
conn = db.verbinden()
|
||||
with conn:
|
||||
zeilen = conn.execute(abfrage, params).fetchall()
|
||||
conn.close()
|
||||
return zeilen
|
||||
|
||||
|
||||
# --- Einladungszugang ------------------------------------------------------
|
||||
|
||||
def test_ohne_einladung_kein_zugang(client):
|
||||
r = client.get("/")
|
||||
assert r.status_code == 403 and "Nur für Gäste" in r.text
|
||||
assert client.get("/api/status").status_code == 401
|
||||
assert client.get("/api/search?q=ab").status_code == 401
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 401
|
||||
|
||||
|
||||
def test_token_vorhanden_und_zugang_anfangs_aus(client):
|
||||
assert len(token()) >= 12
|
||||
assert client.get(f"/z/{token()}", follow_redirects=False).status_code == 403 # access_enabled = False
|
||||
|
||||
|
||||
def test_einladung_gibt_zugang(client):
|
||||
eintreten(client)
|
||||
r = client.get("/")
|
||||
assert r.status_code == 200 and "Song wünschen" in r.text
|
||||
assert client.get("/api/status").status_code == 200
|
||||
|
||||
|
||||
def test_falscher_link(client):
|
||||
setzen(access_enabled=True)
|
||||
assert client.get("/z/falsch", follow_redirects=False).status_code == 403
|
||||
assert client.get("/api/status").status_code == 401
|
||||
|
||||
|
||||
def test_zugang_aus_sperrt_alle(client):
|
||||
eintreten(client)
|
||||
setzen(access_enabled=False)
|
||||
assert client.get("/api/status").status_code == 401
|
||||
assert client.get("/").status_code == 403
|
||||
|
||||
|
||||
def test_neuer_link_macht_alte_zugaenge_ungueltig(client):
|
||||
eintreten(client)
|
||||
alt = token()
|
||||
setzen(access_token=db.neuer_zugangs_token())
|
||||
assert client.get("/api/status").status_code == 401
|
||||
assert client.get(f"/z/{alt}", follow_redirects=False).status_code == 403
|
||||
|
||||
|
||||
# --- Suche -------------------------------------------------------------------
|
||||
|
||||
def test_suche_markiert_gesperrte(client):
|
||||
eintreten(client)
|
||||
setzen(block_explicit=True)
|
||||
sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('artist', ?, 'x')", KUENSTLER_GESPERRT)
|
||||
client.post("/api/wish", json={"track_uri": URI["B"]})
|
||||
treffer = {t["track_uri"][-1]: t["blocked"] for t in client.get("/api/search?q=lied").json()["results"]}
|
||||
assert treffer["A"] is None
|
||||
assert "Wunschliste" in treffer["B"]
|
||||
assert "explizit" in treffer["C"]
|
||||
assert "nicht gewünscht" in treffer["D"]
|
||||
assert "artist_uris" not in client.get("/api/search?q=lied").json()["results"][0]
|
||||
|
||||
|
||||
def test_suche_mindestlaenge(client):
|
||||
eintreten(client)
|
||||
assert client.get("/api/search?q=a").status_code == 422
|
||||
|
||||
|
||||
# --- Wünschen ----------------------------------------------------------------
|
||||
|
||||
def test_wunsch_landet_in_warteschlange(client):
|
||||
eintreten(client)
|
||||
r = client.post("/api/wish", json={"track_uri": URI["A"], "name": " Tante Erna "})
|
||||
assert r.status_code == 200 and r.json()["status"] == "queued"
|
||||
w = sql("SELECT title, guest_name, guest_ip, guest_id FROM wishes")[0]
|
||||
assert (w["title"], w["guest_name"]) == ("Lied A", "Tante Erna")
|
||||
assert w["guest_ip"] != "testclient" and len(w["guest_ip"]) == 24 # nur Hash
|
||||
s = client.get("/api/status").json()
|
||||
assert [q["title"] for q in s["queue"]] == ["Lied A"] and s["queue"][0]["mine"] is True
|
||||
assert s["mine"][0]["status"] == "queued" and s["limit"]["remaining"] == 1
|
||||
|
||||
|
||||
def test_titelinfos_kommen_vom_server(client):
|
||||
eintreten(client)
|
||||
client.post("/api/wish", json={"track_uri": URI["A"], "title": "Gefälscht"})
|
||||
assert sql("SELECT title FROM wishes")[0]["title"] == "Lied A"
|
||||
|
||||
|
||||
def test_moderation(client):
|
||||
eintreten(client)
|
||||
setzen(moderation=True)
|
||||
r = client.post("/api/wish", json={"track_uri": URI["A"]})
|
||||
assert r.json()["status"] == "pending"
|
||||
s = client.get("/api/status").json()
|
||||
assert s["queue"] == [] and s["pending_count"] == 1
|
||||
|
||||
|
||||
def test_pausiert(client):
|
||||
eintreten(client)
|
||||
setzen(wishes_paused=True)
|
||||
r = client.post("/api/wish", json={"track_uri": URI["A"]})
|
||||
assert r.status_code == 403 and "pausiert" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_limit_pro_gast(client):
|
||||
eintreten(client)
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 200
|
||||
assert client.post("/api/wish", json={"track_uri": URI["B"]}).status_code == 200
|
||||
r = client.post("/api/wish", json={"track_uri": URI["E"]})
|
||||
assert r.status_code == 429 and "Limit" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_keine_duplikate(client):
|
||||
eintreten(client)
|
||||
setzen(wish_limit_count=10)
|
||||
client.post("/api/wish", json={"track_uri": URI["A"]})
|
||||
r = client.post("/api/wish", json={"track_uri": URI["A"]})
|
||||
assert r.status_code == 409 and "Wunschliste" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_kuerzlich_gespielt_und_laeuft_gerade(client):
|
||||
eintreten(client)
|
||||
setzen(wish_limit_count=10)
|
||||
sql("INSERT INTO history (track_uri, title, artist, started_at) VALUES (?, 'x', 'y', ?)", URI["A"], db.jetzt())
|
||||
sql("UPDATE pi_state SET track_uri = ? WHERE id = 1", URI["B"])
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 409
|
||||
assert client.post("/api/wish", json={"track_uri": URI["B"]}).status_code == 409
|
||||
alt = "2020-01-01T00:00:00+00:00" # außerhalb der Sperrzeit → wieder erlaubt
|
||||
sql("UPDATE history SET started_at = ?", alt)
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 200
|
||||
|
||||
|
||||
def test_explizit_und_sperrliste(client):
|
||||
eintreten(client)
|
||||
setzen(block_explicit=True, wish_limit_count=10)
|
||||
assert client.post("/api/wish", json={"track_uri": URI["C"]}).status_code == 409
|
||||
sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('artist', ?, 'x')", KUENSTLER_GESPERRT)
|
||||
assert client.post("/api/wish", json={"track_uri": URI["D"]}).status_code == 409
|
||||
sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('track', ?, 'x')", URI["E"])
|
||||
assert client.post("/api/wish", json={"track_uri": URI["E"]}).status_code == 409
|
||||
|
||||
|
||||
def test_gesperrter_gast(client):
|
||||
eintreten(client)
|
||||
gid = client.cookies.get(gaeste.GAST_COOKIE)
|
||||
sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('guest', ?, 'x')", gid)
|
||||
r = client.post("/api/wish", json={"track_uri": URI["A"]})
|
||||
assert r.status_code == 403 and "Gerät" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_ungueltige_eingaben(client):
|
||||
eintreten(client)
|
||||
assert client.post("/api/wish", json={"track_uri": "spotify:album:x"}).status_code == 422
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"], "name": "x" * 31}).status_code == 422
|
||||
unbekannt = "spotify:track:" + "Q" * 22
|
||||
assert client.post("/api/wish", json={"track_uri": unbekannt}).status_code == 404
|
||||
|
||||
|
||||
def test_fremde_herkunft(client):
|
||||
eintreten(client)
|
||||
r = client.post("/api/wish", json={"track_uri": URI["A"]}, headers={"Origin": "https://boese.example"})
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
# --- Status ------------------------------------------------------------------
|
||||
|
||||
def test_laeuft_gerade_und_offline(client):
|
||||
eintreten(client)
|
||||
assert client.get("/api/status").json()["now_playing"] is None
|
||||
sql("UPDATE pi_state SET payload = ?, updated_at = ?, last_seen = ? WHERE id = 1",
|
||||
'{"track": {"track_uri": "%s", "title": "Lied A", "artist": "Band"}, "is_playing": true, '
|
||||
'"progress_ms": 5000, "source": "wish"}' % URI["A"], db.jetzt(), db.jetzt())
|
||||
jetzt = client.get("/api/status").json()["now_playing"]
|
||||
assert jetzt["title"] == "Lied A" and jetzt["is_wish"] is True
|
||||
sql("UPDATE pi_state SET last_seen = '2020-01-01T00:00:00+00:00'") # Player seit Langem still → offline
|
||||
s = client.get("/api/status").json()
|
||||
assert s["online"] is False and s["now_playing"] is None
|
||||
|
||||
|
||||
def test_sicherheits_header(client):
|
||||
r = client.get("/")
|
||||
assert "default-src 'self'" in r.headers["content-security-policy"]
|
||||
@@ -0,0 +1,164 @@
|
||||
"""Tests für die Player-API (Wegwerf-Datenbank pro Test)."""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app import config, db
|
||||
|
||||
TOKEN = "test-token-0123456789"
|
||||
AUTH = {"Authorization": f"Bearer {TOKEN}"}
|
||||
URI_A = "spotify:track:" + "A" * 22
|
||||
URI_B = "spotify:track:" + "B" * 22
|
||||
URI_C = "spotify:track:" + "C" * 22
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path, monkeypatch):
|
||||
env = tmp_path / ".env"
|
||||
env.write_text(f"PLAYER_API_TOKEN={TOKEN}\nDB_PATH={tmp_path / 'audiofy.db'}\nSPOTIFY_PLAYLIST_ID=startliste\n")
|
||||
monkeypatch.setenv("AUDIOFY_ENV", str(env))
|
||||
config.konfig.cache_clear()
|
||||
from app.main import app
|
||||
with TestClient(app) as c:
|
||||
yield c
|
||||
config.konfig.cache_clear()
|
||||
|
||||
|
||||
def wunsch(uri, status="queued", position=1.0, priority=0, title="Titel"):
|
||||
conn = db.verbinden()
|
||||
with conn:
|
||||
cur = conn.execute("INSERT INTO wishes (track_uri, title, artist, status, priority, position, created_at) "
|
||||
"VALUES (?, ?, 'Interpret', ?, ?, ?, ?)", (uri, title, status, priority, position, db.jetzt()))
|
||||
conn.close()
|
||||
return cur.lastrowid
|
||||
|
||||
|
||||
def test_token_pflicht(client):
|
||||
assert client.get("/api/player/config").status_code == 401
|
||||
assert client.get("/api/player/config", headers={"Authorization": "Bearer falsch"}).status_code == 401
|
||||
assert client.get("/api/player/config", headers=AUTH).status_code == 200
|
||||
|
||||
|
||||
def test_config_standardwerte(client):
|
||||
daten = client.get("/api/player/config", headers=AUTH).json()
|
||||
assert daten == {"wish_every_n": 3, "playlist_id": "startliste", "reshuffle_token": 0,
|
||||
"stopped": False, "commands": []}
|
||||
|
||||
|
||||
def test_next_wish_leer_und_reihenfolge(client):
|
||||
assert client.get("/api/player/next-wish", headers=AUTH).json() == {"wish": None}
|
||||
wunsch(URI_A, position=2)
|
||||
b = wunsch(URI_B, position=1)
|
||||
wunsch(URI_C, status="pending", position=0) # zur Freigabe – zählt nicht
|
||||
assert client.get("/api/player/next-wish", headers=AUTH).json()["wish"]["id"] == b
|
||||
besitzer = wunsch(URI_C, position=9, priority=1) # "Als Nächstes" schlägt alles
|
||||
w = client.get("/api/player/next-wish", headers=AUTH).json()["wish"]
|
||||
assert w["id"] == besitzer and w["priority"] is True
|
||||
|
||||
|
||||
def test_wunsch_gespielt_idempotent(client):
|
||||
a = wunsch(URI_A)
|
||||
assert client.post(f"/api/player/wish/{a}/played", headers=AUTH).json() == {"ok": True, "status": "played"}
|
||||
assert client.post(f"/api/player/wish/{a}/played", headers=AUTH).json() == {"ok": True, "status": "played"}
|
||||
assert client.get("/api/player/next-wish", headers=AUTH).json() == {"wish": None}
|
||||
assert client.post("/api/player/wish/9999/played", headers=AUTH).status_code == 404
|
||||
|
||||
|
||||
def test_now_playing_schreibt_verlauf_nur_bei_wechsel(client):
|
||||
a = wunsch(URI_A)
|
||||
titel = {"track_uri": URI_A, "title": "Lied A", "artist": "Band"}
|
||||
for _ in range(3): # wiederholte Meldungen desselben Titels → ein Verlaufseintrag
|
||||
r = client.post("/api/player/now-playing", headers=AUTH,
|
||||
json={"track": titel, "is_playing": True, "source": "wish", "wish_id": a})
|
||||
assert r.status_code == 200
|
||||
client.post("/api/player/now-playing", headers=AUTH, json={"track": titel, "is_playing": True, "new_track": True})
|
||||
client.post("/api/player/now-playing", headers=AUTH,
|
||||
json={"track": {"track_uri": URI_B, "title": "Lied B", "artist": "X"}, "is_playing": True})
|
||||
conn = db.verbinden()
|
||||
verlauf = conn.execute("SELECT track_uri, source, wish_id FROM history ORDER BY id").fetchall()
|
||||
conn.close()
|
||||
assert [(v["track_uri"], v["source"], v["wish_id"]) for v in verlauf] == [
|
||||
(URI_A, "wish", a), (URI_A, None, None), (URI_B, None, None)]
|
||||
|
||||
|
||||
def test_pausierter_titel_nicht_im_verlauf(client):
|
||||
titel = {"track_uri": URI_A, "title": "Lied A", "artist": "Band"}
|
||||
client.post("/api/player/now-playing", headers=AUTH, json={"track": titel, "is_playing": False})
|
||||
conn = db.verbinden()
|
||||
assert conn.execute("SELECT COUNT(*) FROM history").fetchone()[0] == 0
|
||||
conn.close()
|
||||
client.post("/api/player/now-playing", headers=AUTH, json={"track": titel, "is_playing": True})
|
||||
conn = db.verbinden()
|
||||
assert conn.execute("SELECT COUNT(*) FROM history").fetchone()[0] == 1
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_now_playing_validiert(client):
|
||||
r = client.post("/api/player/now-playing", headers=AUTH,
|
||||
json={"track": {"track_uri": "kaputt", "title": "x", "artist": "y"}})
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_befehle_abholen_quittieren_verfallen(client):
|
||||
conn = db.verbinden()
|
||||
skip = db.befehl_anlegen(conn, "skip")
|
||||
vol = db.befehl_anlegen(conn, "volume", {"percent": 40})
|
||||
with conn: # ein uralter Befehl darf nicht mehr ausgeliefert werden
|
||||
conn.execute("INSERT INTO commands (type, created_at) VALUES ('skip', '2020-01-01T00:00:00+00:00')")
|
||||
conn.close()
|
||||
befehle = client.get("/api/player/config", headers=AUTH).json()["commands"]
|
||||
assert [(b["id"], b["type"], b["payload"]) for b in befehle] == [(skip, "skip", {}), (vol, "volume", {"percent": 40})]
|
||||
assert client.post(f"/api/player/commands/{skip}/ack", headers=AUTH, json={"ok": True}).status_code == 200
|
||||
assert [b["id"] for b in client.get("/api/player/config", headers=AUTH).json()["commands"]] == [vol]
|
||||
assert client.post("/api/player/commands/9999/ack", headers=AUTH, json={}).status_code == 404
|
||||
|
||||
|
||||
def test_not_aus(client):
|
||||
wunsch(URI_A)
|
||||
conn = db.verbinden()
|
||||
db.einstellung_setzen(conn, "emergency_stop", True)
|
||||
conn.close()
|
||||
assert client.get("/api/player/config", headers=AUTH).json()["stopped"] is True
|
||||
assert client.get("/api/player/next-wish", headers=AUTH).json() == {"wish": None}
|
||||
|
||||
|
||||
def test_lebenszeichen(client):
|
||||
client.get("/api/player/config", headers=AUTH)
|
||||
conn = db.verbinden()
|
||||
assert conn.execute("SELECT last_seen FROM pi_state").fetchone()["last_seen"] is not None
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_bluetooth_melden(client):
|
||||
daten = {"current": {"mac": "AA:11:22:33:44:55", "name": "Box", "paired": True, "connected": True},
|
||||
"devices": [{"mac": "AA:BB:CC:DD:EE:FF", "name": "Neu"}], "scanning": False,
|
||||
"scanned_at": "2026-10-17T18:00:00+00:00"}
|
||||
assert client.post("/api/player/bluetooth", headers=AUTH, json=daten).status_code == 200
|
||||
from app.player_api import player_zustand
|
||||
conn = db.verbinden()
|
||||
bt = player_zustand(conn)["bluetooth"]
|
||||
conn.close()
|
||||
assert bt["current"]["name"] == "Box" and bt["devices"][0]["mac"] == "AA:BB:CC:DD:EE:FF"
|
||||
falsch = {"devices": [{"mac": "kein-mac"}]}
|
||||
assert client.post("/api/player/bluetooth", headers=AUTH, json=falsch).status_code == 422
|
||||
assert client.post("/api/player/bluetooth", json=daten).status_code == 401
|
||||
|
||||
|
||||
def test_alte_datenbank_wird_nachgeruestet(tmp_path):
|
||||
import sqlite3
|
||||
pfad = tmp_path / "alt.db"
|
||||
alt = sqlite3.connect(pfad)
|
||||
alt.execute("CREATE TABLE pi_state (id INTEGER PRIMARY KEY CHECK (id = 1), track_uri TEXT, "
|
||||
"payload TEXT NOT NULL DEFAULT '{}', updated_at TEXT, last_seen TEXT)")
|
||||
alt.commit()
|
||||
alt.close()
|
||||
db.initialisieren(pfad)
|
||||
conn = sqlite3.connect(pfad)
|
||||
assert "bluetooth" in {r[1] for r in conn.execute("PRAGMA table_info(pi_state)")}
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_alter_pfad_api_pi_bleibt_nutzbar(client):
|
||||
"""Ältere Player sprechen noch /api/pi/… an."""
|
||||
assert client.get("/api/pi/config", headers=AUTH).status_code == 200
|
||||
assert client.get("/api/pi/config").status_code == 401
|
||||
@@ -0,0 +1,122 @@
|
||||
"""Tests für Cloudflare Turnstile (siteverify wird durch einen Platzhalter ersetzt)."""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from argon2 import PasswordHasher
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app import admin, config, gaeste, turnstile
|
||||
from app.spotify import spotify
|
||||
from tests.test_gaeste import URI, FakeSpotify, eintreten
|
||||
|
||||
PASSWORT = "richtig-langes-passwort"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def antworten(monkeypatch):
|
||||
"""Steuert, was siteverify antwortet; merkt sich die Anfragen."""
|
||||
zustand = {"ergebnis": {"success": True, "hostname": "musik.example.de"}, "anfragen": []}
|
||||
|
||||
def siteverify(daten):
|
||||
zustand["anfragen"].append(daten)
|
||||
if isinstance(zustand["ergebnis"], Exception):
|
||||
raise zustand["ergebnis"]
|
||||
return zustand["ergebnis"]
|
||||
|
||||
monkeypatch.setattr(turnstile, "_siteverify", siteverify)
|
||||
return zustand
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path, monkeypatch, antworten):
|
||||
env = tmp_path / ".env"
|
||||
env.write_text(f"DB_PATH={tmp_path / 'audiofy.db'}\nSESSION_SECRET=geheim\nCOOKIE_SECURE=0\n"
|
||||
f"PUBLIC_URL=https://musik.example.de\nADMIN_PASSWORD_HASH={PasswordHasher().hash(PASSWORT)}\n"
|
||||
"TURNSTILE_SITE_KEY=site-oeffentlich\nTURNSTILE_SECRET_KEY=secret-geheim\n")
|
||||
monkeypatch.setenv("AUDIOFY_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(1000, 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 test_site_key_oeffentlich(client):
|
||||
assert client.get("/api/turnstile").json() == {"site_key": "site-oeffentlich"}
|
||||
|
||||
|
||||
def test_wunsch_ohne_token_abgelehnt(client):
|
||||
eintreten(client)
|
||||
r = client.post("/api/wish", json={"track_uri": URI["A"]})
|
||||
assert r.status_code == 400 and "Sicherheitsprüfung" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_wunsch_mit_gueltigem_token(client, antworten):
|
||||
eintreten(client)
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"], "turnstile": "tok"}).status_code == 200
|
||||
anfrage = antworten["anfragen"][0]
|
||||
assert anfrage["secret"] == "secret-geheim" and anfrage["response"] == "tok"
|
||||
|
||||
|
||||
def test_wunsch_ungueltiges_token(client, antworten):
|
||||
eintreten(client)
|
||||
antworten["ergebnis"] = {"success": False, "error-codes": ["invalid-input-response"]}
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"], "turnstile": "tok"}).status_code == 403
|
||||
|
||||
|
||||
def test_falscher_hostname(client, antworten):
|
||||
eintreten(client)
|
||||
antworten["ergebnis"] = {"success": True, "hostname": "boese.example"}
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"], "turnstile": "tok"}).status_code == 403
|
||||
|
||||
|
||||
def test_falsche_aktion(client, antworten):
|
||||
"""Ein Token vom Login-Formular darf keinen Wunsch abschicken (und umgekehrt)."""
|
||||
eintreten(client)
|
||||
antworten["ergebnis"] = {"success": True, "hostname": "musik.example.de", "action": "login"}
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"], "turnstile": "tok"}).status_code == 403
|
||||
|
||||
|
||||
def test_wunsch_bei_cloudflare_ausfall_erlaubt(client, antworten):
|
||||
eintreten(client)
|
||||
antworten["ergebnis"] = httpx.ConnectError("weg")
|
||||
assert client.post("/api/wish", json={"track_uri": URI["A"], "turnstile": "tok"}).status_code == 200
|
||||
|
||||
|
||||
def test_limit_greift_vor_turnstile(client, antworten):
|
||||
"""Abgewiesene Anfragen (z. B. Limit) kosten keine Turnstile-Prüfung."""
|
||||
eintreten(client)
|
||||
client.post("/api/wish", json={"track_uri": URI["A"], "turnstile": "t1"})
|
||||
client.post("/api/wish", json={"track_uri": URI["B"], "turnstile": "t2"})
|
||||
antworten["anfragen"].clear()
|
||||
assert client.post("/api/wish", json={"track_uri": URI["E"], "turnstile": "t3"}).status_code == 429
|
||||
assert antworten["anfragen"] == []
|
||||
|
||||
|
||||
def test_login_ohne_token(client):
|
||||
assert client.post("/admin/login", json={"password": PASSWORT}).status_code == 400
|
||||
|
||||
|
||||
def test_login_mit_token(client):
|
||||
assert client.post("/admin/login", json={"password": PASSWORT, "turnstile": "tok"}).status_code == 200
|
||||
|
||||
|
||||
def test_login_ungueltiges_token(client, antworten):
|
||||
antworten["ergebnis"] = {"success": False}
|
||||
assert client.post("/admin/login", json={"password": PASSWORT, "turnstile": "tok"}).status_code == 403
|
||||
|
||||
|
||||
def test_login_bei_cloudflare_ausfall_gesperrt(client, antworten):
|
||||
antworten["ergebnis"] = httpx.ConnectError("weg")
|
||||
assert client.post("/admin/login", json={"password": PASSWORT, "turnstile": "tok"}).status_code == 503
|
||||
|
||||
|
||||
def test_csp_erlaubt_turnstile(client):
|
||||
csp = client.get("/admin/login").headers["content-security-policy"]
|
||||
assert "script-src 'self' https://challenges.cloudflare.com" in csp
|
||||
assert "frame-src https://challenges.cloudflare.com" in csp
|
||||
Reference in New Issue
Block a user