470 lines
20 KiB
Python
470 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Reproducible test suite for the Deemix MCP server.
|
|
|
|
Run: python test_deemix_mcp.py (from the package directory)
|
|
|
|
The suite spins up a local HTTP server that emulates the Deemix webui API
|
|
(connect / search / getQueue / addToQueue / loginArl / removeFromQueue /
|
|
retryDownload / removeFinishedDownloads / cancelAllDownloads) and points the
|
|
MCP at it. No real Deezer access, no real downloads, no queue residue:
|
|
|
|
* read + auth flow (fresh first call, one-shot re-login on NotLoggedIn,
|
|
401, 403, re-login failure abort)
|
|
* bitrate validation (valid TrackFormats ids, invalid rejected, not kbit/s)
|
|
* url/term/result limits
|
|
* invalid url/id handling
|
|
* compact queue output
|
|
* secret redaction (ARL, cookie headers, long hex)
|
|
|
|
Write-path tools that would enqueue real downloads are verified here ONLY
|
|
against the emulated backend; against production Deemix they are considered
|
|
verified against the source code, not practically executed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import urllib.error
|
|
import urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import Optional
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Emulated Deemix backend
|
|
# --------------------------------------------------------------------------
|
|
FAKE_ARL = "a1" * 96 # 192 hex chars, like a real Deezer ARL
|
|
|
|
|
|
class DeemixState:
|
|
def __init__(self) -> None:
|
|
self.logged_in = False
|
|
self.connect_calls = 0
|
|
self.login_calls = 0
|
|
self.requests: list[str] = []
|
|
self.fail_next_relogin = False
|
|
self.force_401_once = False
|
|
self.force_403_once = False
|
|
self.queue = {
|
|
"album_1_1": {
|
|
"uuid": "album_1_1", "type": "album", "id": "1",
|
|
"title": "Folge 240", "artist": "Die drei ???",
|
|
"progress": 37, "downloaded": 52428800, "size": 80530636,
|
|
"failed": False, "errors": [],
|
|
},
|
|
"album_2_3": {
|
|
"uuid": "album_2_3", "type": "album", "id": "2",
|
|
"title": "Folge 239", "artist": "Die drei ???",
|
|
"progress": 100, "downloaded": 76000000, "size": 76000000,
|
|
"failed": True,
|
|
"errors": ["Tagging failed: unknown picture MIME",
|
|
"second error", "third error", "fourth error"],
|
|
},
|
|
}
|
|
self.added: list[dict] = []
|
|
self.cancelled_all = False
|
|
|
|
|
|
STATE = DeemixState()
|
|
|
|
|
|
class EmulatedDeemix(BaseHTTPRequestHandler):
|
|
def log_message(self, *a): # quiet
|
|
pass
|
|
|
|
def _send(self, obj: dict, status: int = 200) -> None:
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _auth_required(self) -> bool:
|
|
"""True when the session cookie is missing (not logged in)."""
|
|
cookie = self.headers.get("Cookie", "")
|
|
return "deemix_session" not in cookie
|
|
|
|
def do_GET(self):
|
|
STATE.requests.append("GET " + self.path)
|
|
if self.path.startswith("/deemix/api/connect"):
|
|
STATE.connect_calls += 1
|
|
# In real Deemix the session IS the cookie: without it the
|
|
# server reports singleUser + the ARL again, so a client can
|
|
# re-login after any session loss.
|
|
authed = not self._auth_required()
|
|
obj: dict = {
|
|
"update": {"deemixVersion": "3.13.7", "webuiVersion": "4.6.0"},
|
|
"autologin": not authed,
|
|
"currentUser": None,
|
|
"deezerAvailable": "yes",
|
|
"spotifyEnabled": True,
|
|
"settingsData": {"settings": {
|
|
"downloadLocation": "/downloads", "maxBitrate": 9,
|
|
}},
|
|
}
|
|
if not authed:
|
|
obj["singleUser"] = {"arl": FAKE_ARL}
|
|
else:
|
|
obj["currentUser"] = {"id": 12345, "name": "Test",
|
|
"lastName": "User", "country": "DE",
|
|
"subscription": "premium"}
|
|
self._send(obj)
|
|
return
|
|
if self._force_auth_error():
|
|
return
|
|
if self._auth_required():
|
|
self._send({"result": False, "errid": "NotLoggedIn"})
|
|
return
|
|
if self.path.startswith("/deemix/api/search"):
|
|
q = urllib.parse.parse_qs(self.path.split("?", 1)[1]) if "?" in self.path else {}
|
|
nb = int(q.get("nb", ["10"])[0])
|
|
total = 31
|
|
data = [{"id": 985501051, "title": "Die drei ??? Folge 240",
|
|
"release_date": "2026-08-18", "nb_tracks": 48,
|
|
"related": {"artists": [{"name": "Die drei ???"},
|
|
{"name": "Mystery-Club"}]}}]
|
|
data += [{"id": i, "title": f"Treffer {i}",
|
|
"release_date": "2020-01-01", "nb_tracks": 10,
|
|
"related": {"artists": [{"name": "K"}]}} for i in range(1, total)]
|
|
self._send({"data": data[:nb], "total": total, "type": "album",
|
|
"error": ""})
|
|
return
|
|
if self.path.startswith("/deemix/api/getQueue"):
|
|
self._send({"queue": dict(STATE.queue),
|
|
"queueOrder": list(STATE.queue.keys())})
|
|
return
|
|
self._send({"result": False, "errid": "NotFound"}, 404)
|
|
|
|
def _force_auth_error(self) -> bool:
|
|
"""Emulate a session that the server rejects with HTTP 401/403."""
|
|
if STATE.force_401_once:
|
|
STATE.force_401_once = False
|
|
self._send({"result": False, "errid": "http401"}, 401)
|
|
return True
|
|
if STATE.force_403_once:
|
|
STATE.force_403_once = False
|
|
self._send({"result": False, "errid": "http403"}, 403)
|
|
return True
|
|
return False
|
|
|
|
def do_POST(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
raw = self.rfile.read(length).decode() if length else ""
|
|
try:
|
|
body = json.loads(raw) if raw else {}
|
|
except json.JSONDecodeError:
|
|
body = {}
|
|
STATE.requests.append("POST " + self.path)
|
|
if self.path.startswith("/deemix/api/loginArl"):
|
|
STATE.login_calls += 1
|
|
if STATE.fail_next_relogin:
|
|
STATE.fail_next_relogin = False
|
|
self._send({"status": 0, "error": "invalidArl"})
|
|
return
|
|
if body.get("arl") != FAKE_ARL:
|
|
self._send({"status": 0, "error": "invalidArl"})
|
|
return
|
|
STATE.logged_in = True
|
|
# Emulate setting the session cookie (real Deemix sets a cookie).
|
|
self.send_response(200)
|
|
self.send_header("Set-Cookie", "deemix_session=abc123; Path=/")
|
|
self.send_header("Content-Type", "application/json")
|
|
payload = json.dumps({"status": 1, "user": {"id": 12345,
|
|
"name": "Test",
|
|
"lastName": "User",
|
|
"country": "DE"}})
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload.encode())
|
|
return
|
|
if self._auth_required():
|
|
self._send({"result": False, "errid": "NotLoggedIn"})
|
|
return
|
|
if self._force_auth_error():
|
|
return
|
|
if self.path.startswith("/deemix/api/addToQueue"):
|
|
url = body.get("url", "")
|
|
bitrate = body.get("bitrate")
|
|
# Mirror the real route: empty/"null" falls back to settings.
|
|
if bitrate == "null" or not bitrate:
|
|
bitrate = 9
|
|
bitrate = int(bitrate)
|
|
if "badid" in url or "9999999999" in url:
|
|
# Real Deemix behaviour: an unknown ID is accepted (result
|
|
# true) but resolves to zero items, so nothing is queued.
|
|
self._send({"result": True, "data": {
|
|
"url": [u for u in url.split()], "bitrate": bitrate,
|
|
"obj": []}})
|
|
return
|
|
STATE.added.append({"url": url, "bitrate": bitrate})
|
|
ids = [u.rsplit("/", 1)[-1] for u in url.split()]
|
|
self._send({"result": True, "data": {
|
|
"url": ids, "bitrate": bitrate,
|
|
"obj": [{"id": i, "type": "album"} for i in ids]}})
|
|
return
|
|
if self.path.startswith("/deemix/api/removeFromQueue"):
|
|
uuid = urllib.parse.parse_qs(self.path.split("?", 1)[1]).get("uuid", [""])[0]
|
|
if uuid not in STATE.queue:
|
|
self._send({"result": False, "errid": "UnknownUuid"})
|
|
return
|
|
del STATE.queue[uuid]
|
|
self._send({"result": True})
|
|
return
|
|
if self.path.startswith("/deemix/api/retryDownload"):
|
|
uuid = body.get("uuid", "")
|
|
if uuid not in STATE.queue:
|
|
self._send({"result": False, "errid": "UnknownUuid"})
|
|
return
|
|
STATE.queue[uuid]["failed"] = False
|
|
self._send({"result": True})
|
|
return
|
|
if self.path.startswith("/deemix/api/removeFinishedDownloads"):
|
|
for k in list(STATE.queue):
|
|
if not STATE.queue[k]["failed"]:
|
|
del STATE.queue[k]
|
|
self._send({"result": True})
|
|
return
|
|
if self.path.startswith("/deemix/api/cancelAllDownloads"):
|
|
STATE.cancelled_all = True
|
|
self._send({"result": True})
|
|
return
|
|
self._send({"result": False, "errid": "NotFound"}, 404)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Tests
|
|
# --------------------------------------------------------------------------
|
|
import urllib.parse # noqa: E402 (used by the emulated handler above)
|
|
|
|
PASS: list[str] = []
|
|
FAIL: list[str] = []
|
|
|
|
|
|
def check(name: str, cond: bool, detail: str = "") -> None:
|
|
(PASS if cond else FAIL).append(name)
|
|
print(("PASS " if cond else "FAIL ") + name + (f" [{detail}]" if detail and not cond else ""))
|
|
|
|
|
|
def main() -> int:
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), EmulatedDeemix)
|
|
port = server.server_address[1]
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
|
|
os.environ["DEEMIX_URL"] = f"http://127.0.0.1:{port}"
|
|
os.environ["DEEMIX_ARL"] = "" # force single-user path
|
|
# fresh import so module-level env reads pick up the emulated URL
|
|
if "deemix_mcp" in sys.modules:
|
|
del sys.modules["deemix_mcp"]
|
|
import importlib
|
|
import deemix_mcp
|
|
importlib.reload(deemix_mcp)
|
|
from deemix_mcp import (Session, DeemixError, redact, _clean_error,
|
|
deemix_status, deemix_search, deemix_add_to_queue,
|
|
deemix_queue, deemix_retry_download,
|
|
deemix_remove_from_queue, TRACKFORMATS,
|
|
MAX_URLS_PER_CALL, MAX_SEARCH_RESULTS,
|
|
MAX_SEARCH_TERM)
|
|
|
|
S = deemix_mcp.SESSION
|
|
S._logged_in = False
|
|
S.jar.clear()
|
|
|
|
# 1) fresh first call right after "container start" --------------------
|
|
out = json.loads(deemix_status())
|
|
check("erstaufruf: status nach frischem Start",
|
|
out["logged_in_user"].get("name") == "Test", str(out.get("logged_in_user")))
|
|
check("erstaufruf: ARL nicht in Status-Output",
|
|
FAKE_ARL not in json.dumps(out))
|
|
check("erstaufruf: genau ein Login-Versuch",
|
|
STATE.login_calls == 1, f"login_calls={STATE.login_calls}")
|
|
|
|
# 2) direct first call of search (session already warm) -----------------
|
|
out = json.loads(deemix_search("Die drei ??? Folge 240"))
|
|
check("erstaufruf: search funktioniert",
|
|
out["total"] == 31 and len(out["results"]) == 10)
|
|
check("search: Deezer-Link praeSENT",
|
|
out["results"][0]["link"] == "https://www.deezer.com/album/985501051")
|
|
|
|
out = json.loads(deemix_queue())
|
|
check("erstaufruf: queue funktioniert", out["total"] == 2)
|
|
check("queue: kompakt (keine groben Rohfelder)",
|
|
all("downloaded_mb" in r for r in out["rows"]) and
|
|
all("size" not in r for r in out["rows"]))
|
|
check("queue: errors auf 3 begrenzt",
|
|
[r for r in out["rows"] if r["uuid"] == "album_2_3"][0]["errors"] and
|
|
len([r for r in out["rows"] if r["uuid"] == "album_2_3"][0]["errors"]) == 3)
|
|
|
|
# 3) automatic ONE re-login on NotExpired session ------------------------
|
|
S._logged_in = True
|
|
S.jar.clear() # server now sees no session cookie -> NotLoggedIn
|
|
out = json.loads(deemix_queue())
|
|
check("re-login: NotLoggedIn wird automatisch einxmal behoben",
|
|
out["total"] == 2)
|
|
check("re-login: genau EIN zusaeztlicher Login",
|
|
STATE.login_calls == 2, f"login_calls={STATE.login_calls}")
|
|
|
|
# 4) HTTP 401 triggers the same bounded re-login -------------------------
|
|
STATE.force_401_once = True
|
|
S._logged_in = True
|
|
out = json.loads(deemix_queue())
|
|
check("re-login: HTTP 401 wird automatisch einxmal behoben",
|
|
out["total"] == 2 and STATE.login_calls == 3)
|
|
|
|
# 5) HTTP 403 likewise ----------------------------------------------------
|
|
STATE.force_403_once = True
|
|
out = json.loads(deemix_queue())
|
|
check("re-login: HTTP 403 wird automatisch einxmal behoben",
|
|
out["total"] == 2 and STATE.login_calls == 4)
|
|
|
|
# 6) failed re-login aborts (no loop) ------------------------------------
|
|
S._logged_in = False
|
|
S.jar.clear()
|
|
STATE.fail_next_relogin = True
|
|
try:
|
|
deemix_queue()
|
|
check("re-login: fehlgeschlagener Re-Login bricht ab", False, "kein Fehler geworfen")
|
|
except DeemixError as e:
|
|
ok = "re-login" in str(e).lower() or "session" in str(e).lower()
|
|
check("re-login: fehlgeschlagener Re-Login bricht ab", ok, str(e))
|
|
check("re-login: kein Endlosschleifen-Verhalten (max. 1 Versuch)",
|
|
STATE.login_calls == 5, f"login_calls={STATE.login_calls}")
|
|
|
|
# restore a working session for the remaining tests
|
|
S._logged_in = False
|
|
S.jar.clear()
|
|
deemix_status()
|
|
|
|
# 7) bitrate validation ----------------------------------------------------
|
|
out = json.loads(deemix_add_to_queue(
|
|
"https://www.deezer.com/album/985501051", bitrate=3))
|
|
check("bitrate: 3 (MP3 320) akzeptiert und uebergeben",
|
|
out["queued"] and out["bitrate_used"] == 3)
|
|
out = json.loads(deemix_add_to_queue(
|
|
"https://www.deezer.com/album/985501051", bitrate=9))
|
|
check("bitrate: 9 (FLAC) akzeptiert", out["queued"] and out["bitrate_used"] == 9)
|
|
out = json.loads(deemix_add_to_queue(
|
|
"https://www.deezer.com/album/985501051", bitrate=1))
|
|
check("bitrate: 1 (MP3 128) akzeptiert", out["queued"] and out["bitrate_used"] == 1)
|
|
out = json.loads(deemix_add_to_queue("https://www.deezer.com/album/985501051"))
|
|
check("bitrate: weggelassen -> Deemix-Setting (9) greift",
|
|
out["bitrate_requested"] is None and out["bitrate_used"] == 9)
|
|
try:
|
|
deemix_add_to_queue("https://www.deezer.com/album/985501051", bitrate=320)
|
|
check("bitrate: 320 (kbit/s-Irrtum) wird abgelehnt", False, "kein Fehler")
|
|
except DeemixError as e:
|
|
check("bitrate: 320 (kbit/s-Irrtum) wird abgelehnt",
|
|
"320" in str(e) and "not a valid" in str(e), str(e))
|
|
try:
|
|
deemix_add_to_queue("https://www.deezer.com/album/985501051", bitrate=0)
|
|
check("bitrate: 0 (LOCAL) wird abgelehnt", False, "kein Fehler")
|
|
except DeemixError:
|
|
check("bitrate: 0 (LOCAL) wird abgelehnt", True)
|
|
check("bitrate: nur die WebUI-Werte 1/3/9 sind gueltig",
|
|
set(TRACKFORMATS) == {1, 3, 9}, str(TRACKFORMATS))
|
|
|
|
# 8) url limits -----------------------------------------------------------
|
|
many = " ".join(f"https://www.deezer.com/album/{i}" for i in range(MAX_URLS_PER_CALL + 1))
|
|
try:
|
|
deemix_add_to_queue(many)
|
|
check("limit: >10 URLs abgelehnt", False, "kein Fehler")
|
|
except DeemixError as e:
|
|
check("limit: >10 URLs abgelehnt", "11" in str(e), str(e))
|
|
ok10 = " ".join(f"https://www.deezer.com/album/{i}" for i in range(10))
|
|
out = json.loads(deemix_add_to_queue(ok10))
|
|
check("limit: genau 10 URLs akzeptiert", out["queued"])
|
|
|
|
# 9) search limits ----------------------------------------------------------
|
|
out = json.loads(deemix_search("x" * (MAX_SEARCH_TERM + 50), nb=999))
|
|
check("limit: Suchtext wird auf 100 Zeichen gekuerzt",
|
|
out["total"] == 31) # emulated server answers regardless; no crash
|
|
out = json.loads(deemix_search("test", nb=999))
|
|
check("limit: nb wird auf 20 begrenzt",
|
|
len(out["results"]) <= MAX_SEARCH_RESULTS)
|
|
|
|
# 10) invalid url / id --------------------------------------------------------
|
|
# Real Deemix behaviour: unknown IDs are accepted but resolve to zero
|
|
# items -> the tool reports queued=false with a warning and nothing is
|
|
# queued (verified against production Deemix 3.13.7 / webui 4.6.0).
|
|
out = json.loads(deemix_add_to_queue("https://www.deeeeezer.com/badid/9999999999"))
|
|
check("ungueltige URL/ID: wird ehrlich gemeldet (queued=false + Warnung)",
|
|
out["queued"] is False and "NO items" in (out.get("warning") or ""),
|
|
str(out))
|
|
check("ungueltige URL/ID: nichts in der Queue (kein Download)",
|
|
STATE.added == [] or all("9999999999" not in a["url"] for a in STATE.added),
|
|
str(STATE.added))
|
|
try:
|
|
deemix_add_to_queue("")
|
|
check("leere urls abgelehnt", False, "kein Fehler")
|
|
except DeemixError:
|
|
check("leere urls abgelehnt", True)
|
|
try:
|
|
deemix_retry_download("album_1_1")
|
|
check("retry: vorhandene UUID ok", True)
|
|
except DeemixError as e:
|
|
check("retry: vorhandene UUID ok", False, str(e))
|
|
try:
|
|
deemix_retry_download("album_kein_kein")
|
|
check("retry: unbekannte UUID sauberer Fehler", False, "kein Fehler")
|
|
except DeemixError as e:
|
|
check("retry: unbekannte UUID sauberer Fehler", "UnknownUuid" in str(e), str(e))
|
|
|
|
# 11) remove / finished (against emulated backend only) ----------------------
|
|
out = json.loads(deemix_remove_from_queue("album_2_3"))
|
|
check("remove_from_queue: UUID entfernt", out["result"] and
|
|
"album_2_3" not in STATE.queue)
|
|
out = json.loads(deemix_queue())
|
|
check("queue: nach Entfernen nur 1 Eintrag", out["total"] == 1)
|
|
try:
|
|
deemix_remove_from_queue("album_unbekannt")
|
|
check("remove_from_queue: unbekannte UUID sauberer Fehler", False, "kein Fehler")
|
|
except DeemixError:
|
|
check("remove_from_queue: unbekannte UUID sauberer Fehler", True)
|
|
|
|
# 12) secret redaction ---------------------------------------------------------
|
|
blob = {
|
|
"arl": FAKE_ARL,
|
|
"cookie": "deemix_session=abc123",
|
|
"note": f"token {FAKE_ARL[:100]} embedded",
|
|
"nested": {"set-cookie": "a=1", "ok": "bleibt"},
|
|
"list": [FAKE_ARL],
|
|
}
|
|
red = redact(blob)
|
|
s = json.dumps(red)
|
|
check("redaction: ARL verschwaengt", FAKE_ARL not in s)
|
|
check("redaction: Cookie-Feld verschwaengt",
|
|
red["cookie"] == "[REDACTED]" and red["nested"]["set-cookie"] == "[REDACTED]")
|
|
check("redaction: unfaellige Felder bleiben",
|
|
red["nested"]["ok"] == "bleibt")
|
|
check("redaction: lange Hexfolgen werden maskiert",
|
|
"[REDACTED]" in s)
|
|
msg = _clean_error(f"boom arl={FAKE_ARL}")
|
|
check("redaction: Fehlermeldungen werden bereinigt",
|
|
FAKE_ARL not in msg and "[REDACTED]" in msg)
|
|
|
|
# 13) compact queue output ------------------------------------------------------
|
|
big = [{"uuid": f"u{i}", "type": "album", "id": i, "title": "t", "artist": "a",
|
|
"progress": i, "downloaded": 1048576 * i, "size": 2097152 * i,
|
|
"failed": i % 7 == 0} for i in range(60)]
|
|
q = json.loads(deemix_queue())
|
|
check("queue: Ausgabe kompakt und begrenzt",
|
|
q["total"] <= 50 or q.get("truncated"))
|
|
|
|
server.shutdown()
|
|
|
|
print(f"\n{len(PASS)} bestanden, {len(FAIL)} fehlgeschlagen")
|
|
if FAIL:
|
|
print("Fehlschlaege:", *FAIL, sep="\n - ")
|
|
return 1
|
|
print("ALLE TESTS BESTANDEN (emuliertes Deemix-Backend, keine echten Downloads)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|