99admandClaude Opus 5.5
fc15f2b933
Name, Seitentitel, Dienste (playwish-web, playwish-controller), Dienst-Benutzer (playwish), Pfade (/opt/playwish, /usr/local/src/playwish), Vorlagen und Repository-Adresse. „Audiofy“ war bereits von mehreren Produkten belegt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
165 lines
7.4 KiB
Python
165 lines
7.4 KiB
Python
"""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 / 'playwish.db'}\nSPOTIFY_PLAYLIST_ID=startliste\n")
|
||
monkeypatch.setenv("PLAYWISH_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,
|
||
"playlist_repeat_percent": 50, "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
|