snowspeeder/hope-voice-api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: split music streaming into Hope Music
Rebuild & Deploy / rebuild (push) Successful in 2m7s
Details

Browse Source 
Move the media server, downloader, library volume, and browser player to the
standalone hope-media-music product. Keep Hope's music tools as a thin HTTP
client of the LAN service on port 8094.

Use a stable voice-room player scope instead of ephemeral LiveKit room names.
The agent now owns the conversation handshake: fresh reported playback mutes
its input, joining pauses music, and the last participant leaving resumes only
music it paused. Peer failures remain non-fatal.

Also move the voice deploy workflow onto the active Unraid Gitea runner and
remove yt-dlp/ffmpeg/media health and volume dependencies from the image.
This commit is contained in:
snowspeeder authored and Shane Maynard committed 2026-09-12 06:35:15 -04:00
1 parent d4ea4fc984
commit 436b3fc8b1
23 files changed
+356 -2518
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/README.md
+2 -2
View File
@@ -15,7 +15,7 @@ tool servers, and agent process.
## Run
```bash
cd /home/shane/dev/voice
cd ~/dev/hope-voice-api
python -m pytest tests/ -v # everything
python -m pytest tests/ -v -m "not slow" # skip LLM/network-heavy tests
```
@@ -27,7 +27,7 @@ python -m pytest tests/ -v -m "not slow" # skip LLM/network-heavy tests
| `test_web_ui.py` | nginx HTTPS UI (:9178) + token endpoint (:8091) | static assets, `/token` (direct + via nginx), `/livekit/` proxy |
| `test_llm_api.py` | LLM API directly | completion, tool calling, `enable_thinking=false` latency, streaming |
| `test_mcp_tools.py` | MCP servers via `docker exec` | weather/time, authenticated remote memory save/recall/delete, and local skills — probes are cleaned up |
| `test_media.py` | media server API + player UI + media MCP | library/player state flow, Range proxy, static mp3 Range, live YT Music search (`-m slow`), gated full download (`HOPE_RUN_YTDLP_DOWNLOAD=1`) |
| `test_media.py` | external Hope Music boundary + media MCP client | LAN service health, stable room scope, split regression guards, local-container MCP probes |
| `test_agent_integration.py` | running container | agent process alive, worker registered with LiveKit, all 4 supervised processes up, container health, LiveKit HTTP API |
| `test_compaction.py` | `HopeLLM._compact_context` in-container | compaction triggers >24 items, system prompt preserved, short context untouched |
tests/conftest.py
+1 -2
View File
@@ -27,8 +27,7 @@ WEB_BASE_URL = os.environ.get("VOICE_WEB_URL", "https://192.168.86.2:9178").rstr
TOKEN_URL = os.environ.get(
"VOICE_TOKEN_URL", f"{WEB_BASE_URL}/token"
)
# Filesystem-coupled tests (repo ./media bind mount) only make sense when
# the test host IS the deployment host.
# docker-exec integration probes only make sense when the test target is local.
IS_LOCAL_DEPLOYMENT = urlparse(WEB_BASE_URL).hostname in ("localhost", "127.0.0.1")
LLM_BASE_URL = (
os.environ.get("LLM_BASE_URL")
tests/test_media.py
+103 -534
View File
@@ -1,484 +1,112 @@
"""Tests for the media player subsystem.
"""Voice-side integration tests for the standalone Hope Music service.
Covers the media server API (via nginx HTTPS and direct), the stream proxy
with HTTP Range passthrough, static file serving of the ./media volume, the
web player markup, and Hope's media MCP toolset. Download tests are gated
behind HOPE_RUN_YTDLP_DOWNLOAD=1 (network-heavy, occasionally flaky).
The library, downloads, stream proxy, web player, and their exhaustive tests
live in ``hope-media-music`` now. This module guards the remaining product
boundary: the voice repo contains only the MCP/HTTP client, uses one stable
player scope, and can reach the LAN service.
"""
import json
import os
import subprocess
import socket
import threading
import time
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.parse import urlparse
import httpx
import pytest
from conftest import (
TOKEN_URL, docker_exec, docker_exec_python, IS_LOCAL_DEPLOYMENT, WEB_BASE_URL
from conftest import IS_LOCAL_DEPLOYMENT, WEB_BASE_URL
ROOT = Path(__file__).parents[1]
MUSIC_BASE_URL = os.environ.get(
"MUSIC_BASE_URL", "http://192.168.86.2:8094"
).rstrip("/")
MUSIC_ROOM = os.environ.get("MUSIC_ROOM", "voice-room")
MEDIA_API = "/media/api"
def test_external_music_service_health_and_library():
"""The peer service used by the deployed agent is reachable over the LAN."""
with httpx.Client(base_url=MUSIC_BASE_URL, timeout=15) as client:
health = client.get(f"{MEDIA_API}/healthz")
assert health.status_code == 200
assert health.json()["ok"] is True
library = client.get(f"{MEDIA_API}/library")
assert library.status_code == 200
assert isinstance(library.json()["tracks"], list)
def test_voice_repo_contains_only_the_music_client():
"""Regression guard for the product split: backend/player cannot creep back."""
agent = (ROOT / "agent" / "agent.py").read_text()
client = (ROOT / "agent" / "media_mcp.py").read_text()
web_js = (ROOT / "web" / "app.js").read_text()
web_html = (ROOT / "web" / "index.html").read_text()
assert not (ROOT / "web" / "media_server.py").exists()
assert not (ROOT / "media").exists()
assert "MUSIC_BASE_URL" in agent and "MUSIC_ROOM" in agent
assert "MUSIC_BASE_URL" in client and "MUSIC_ROOM" in client
assert "MEDIA_SERVER_URL" not in agent + client
assert "media-control" not in agent + web_js
assert "media_state" not in agent + web_js
assert "/media/api" not in web_js
assert 'id="mediaPanel"' not in web_html
assert 'id="mediaAudio"' not in web_html
def test_voice_uses_a_stable_music_scope():
"""The ephemeral LiveKit room must not be the standalone player's scope."""
assert MUSIC_ROOM == "voice-room" or os.environ.get("MUSIC_ROOM")
source = (ROOT / "agent" / "agent.py").read_text()
assert 'MUSIC_ROOM = os.environ.get("MUSIC_ROOM", "voice-room")' in source
assert '"MUSIC_ROOM": MUSIC_ROOM' in source
# These exercise the actual MCP subprocess inside a local voice container. The
# default suite targets Unraid remotely, where docker-exec intentionally skips.
needs_local_container = pytest.mark.skipif(
not IS_LOCAL_DEPLOYMENT,
reason="docker-exec MCP probes require a local voice container",
)
MEDIA_API = "/media/api"
TEST_ROOM = f"media-test-{uuid.uuid4().hex[:8]}"
MEDIA_MCP_ENV = {
"MUSIC_BASE_URL": MUSIC_BASE_URL,
"MUSIC_ROOM": "mcp-test-room",
}
# ── local upstream for stream-proxy tests ───────────────────────────────────
def _serve_fake_audio(payload: bytes):
"""Tiny Range-capable HTTP server on an ephemeral port.
Binds 0.0.0.0 and advertises this host's LAN IP so the deployment host's
media server (which may be a different machine — Unraid) can reach it
over the LAN when fetching external stream URLs.
"""
import socket
"""Tiny Range-capable upstream reachable by a host-networked container."""
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
rng = self.headers.get("Range")
if rng and rng.startswith("bytes="):
start_s, _, end_s = rng[6:].partition("-")
start = int(start_s or 0)
end = int(end_s) if end_s else len(payload) - 1
end = min(end, len(payload) - 1)
body = payload[start : end + 1]
self.send_response(206)
self.send_header(
"Content-Range", f"bytes {start}-{end}/{len(payload)}"
)
else:
body = payload
self.send_response(200)
self.send_response(200)
self.send_header("Content-Type", "audio/mpeg")
self.send_header("Content-Length", str(len(body)))
self.send_header("Accept-Ranges", "bytes")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(body)
self.wfile.write(payload)
def log_message(self, *args):
def log_message(self, format, *args): # noqa: A002
pass
server = HTTPServer(("0.0.0.0", 0), Handler)
port = server.server_address[1]
# The local IP toward the deployment host (works across the LAN).
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
probe.connect((urlparse(WEB_BASE_URL).hostname or "192.168.86.2", 9178))
lan_ip = probe.getsockname()[0]
target = urlparse(WEB_BASE_URL).hostname or "192.168.86.2"
probe.connect((target, urlparse(WEB_BASE_URL).port or 9178))
host = probe.getsockname()[0]
except OSError:
lan_ip = "127.0.0.1"
host = "127.0.0.1"
finally:
probe.close()
threading.Thread(target=server.serve_forever, daemon=True).start()
return server, f"http://{lan_ip}:{port}/audio.mp3"
return server, f"http://{host}:{port}/audio.mp3"
# ── health + basic API shape ────────────────────────────────────────────────
def test_media_server_health_via_nginx(client):
resp = client.get(f"{MEDIA_API}/healthz")
assert resp.status_code == 200
assert resp.json()["ok"] is True
def test_media_library_endpoint(client):
resp = client.get(f"{MEDIA_API}/library")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data["tracks"], list)
assert data["version"] >= 1
for track in data["tracks"]:
assert {"id", "kind", "title", "status"} <= set(track)
def test_media_search_requires_query(client):
resp = client.post(f"{MEDIA_API}/search", json={"query": ""})
assert resp.status_code == 400
def test_media_play_requires_target(client):
resp = client.post(f"{MEDIA_API}/play", json={"room": TEST_ROOM})
assert resp.status_code == 400
def test_media_unknown_stream_404(client):
resp = client.get(f"{MEDIA_API}/stream/does-not-exist")
assert resp.status_code == 404
# ── player state flow (uses a live stream track as the fixture) ─────────────
def test_media_player_state_flow(client):
"""add stream -> play -> state -> report -> pause -> seek hint -> stop."""
upstream, url = _serve_fake_audio(b"\x00" * 64)
try:
added = client.post(
f"{MEDIA_API}/streams", json={"name": "Test FM", "url": url}
)
assert added.status_code == 200
track = added.json()["track"]
track_id = track["id"]
assert track["kind"] == "stream"
assert track["status"] == "ready"
assert track["audio_url"].startswith(f"{MEDIA_API}/stream/")
played = client.post(
f"{MEDIA_API}/play", json={"room": TEST_ROOM, "track_id": track_id}
).json()
assert played["desired"] == "playing"
assert played["track"]["id"] == track_id
state = client.get(f"{MEDIA_API}/state", params={"room": TEST_ROOM}).json()
assert state["track"]["id"] == track_id
client.post(
f"{MEDIA_API}/report",
json={"room": TEST_ROOM, "track_id": track_id,
"state": "playing", "position": 12.5},
)
state = client.get(f"{MEDIA_API}/state", params={"room": TEST_ROOM}).json()
assert abs(state["reported"]["position"] - 12.5) < 0.01
client.post(f"{MEDIA_API}/control", json={"room": TEST_ROOM, "action": "pause"})
assert client.get(f"{MEDIA_API}/state", params={"room": TEST_ROOM}).json()["desired"] == "paused"
client.post(
f"{MEDIA_API}/control",
json={"room": TEST_ROOM, "action": "seek", "position": 42},
)
hint = client.get(f"{MEDIA_API}/state", params={"room": TEST_ROOM}).json()["seek_hint"]
assert hint["position"] == 42
# start_paused variant: scope transfer without starting playback
transfer = client.post(
f"{MEDIA_API}/play",
json={"room": TEST_ROOM + "-b", "track_id": track_id, "start_paused": True},
).json()
assert transfer["desired"] == "paused"
# playing clears any open suggestion choices
client.post(
f"{MEDIA_API}/choices",
json={"room": TEST_ROOM, "choices": [{"video_id": "x", "title": "X"}]},
)
client.post(
f"{MEDIA_API}/play", json={"room": TEST_ROOM, "track_id": track_id}
)
assert client.get(f"{MEDIA_API}/state", params={"room": TEST_ROOM}).json()["choices"] == []
stopped = client.post(
f"{MEDIA_API}/control", json={"room": TEST_ROOM, "action": "stop"}
).json()
assert stopped["desired"] == "stopped"
finally:
upstream.shutdown()
# Remove the test stream from the permanent library.
try:
client.delete(f"{MEDIA_API}/tracks/{track_id}")
except (httpx.HTTPError, UnboundLocalError):
pass
def test_media_stream_proxy_and_range(client):
"""External audio URLs must be reachable same-origin (HTTPS) and support
Range passthrough so the browser can seek progressive streams.
Local deployments test against a controlled fake upstream; remote
deployments (Unraid) cannot reach the test host's ephemeral ports, so a
public reference mp3 is used and Range semantics are asserted without
byte equality."""
if IS_LOCAL_DEPLOYMENT:
payload = b"0123456789" * 100
upstream, url = _serve_fake_audio(payload)
track_id = None
try:
track_id = client.post(
f"{MEDIA_API}/streams", json={"name": "Range FM", "url": url}
).json()["track"]["id"]
full = client.get(f"{MEDIA_API}/stream/{track_id}")
assert full.status_code == 200
assert full.content == payload
ranged = client.get(
f"{MEDIA_API}/stream/{track_id}", headers={"Range": "bytes=100-199"}
)
assert ranged.status_code == 206
assert ranged.content == payload[100:200]
assert ranged.headers["content-range"] == f"bytes 100-199/{len(payload)}"
finally:
upstream.shutdown()
if track_id:
try:
client.delete(f"{MEDIA_API}/tracks/{track_id}")
except httpx.HTTPError:
pass
return
# Remote deployment: public reference audio (SoundHelix).
url = "https://www.soundhelix.com/examples/mp3/SoundHelix-Song-1.mp3"
track_id = None
try:
track_id = client.post(
f"{MEDIA_API}/streams", json={"name": "Range FM public", "url": url}
).json()["track"]["id"]
full = client.get(f"{MEDIA_API}/stream/{track_id}")
assert full.status_code == 200
assert full.headers.get("content-type", "").startswith("audio/")
assert len(full.content) > 100_000
ranged = client.get(
f"{MEDIA_API}/stream/{track_id}", headers={"Range": "bytes=0-1023"}
)
assert ranged.status_code == 206
assert len(ranged.content) == 1024
assert (ranged.headers.get("content-range") or "").startswith("bytes 0-1023/")
finally:
if track_id:
try:
client.delete(f"{MEDIA_API}/tracks/{track_id}")
except httpx.HTTPError:
pass
def test_media_files_static_range_support(client):
"""Downloaded tracks are nginx static files with native Range support.
Writes a probe file into the repo's ./media bind mount, so it only runs
when the test host IS the deployment host (local dev deployment)."""
if not IS_LOCAL_DEPLOYMENT:
pytest.skip("writes to the repo ./media bind mount — local deployments only")
import pathlib
marker = uuid.uuid4().hex[:8]
host_media = pathlib.Path(__file__).parents[1] / "media" / "files"
host_media.mkdir(parents=True, exist_ok=True)
fname = f"test-{marker}.mp3"
payload = b"IMHEADDATA" + b"\x01" * 2048
(host_media / fname).write_bytes(payload)
try:
full = client.get(f"/media/files/{fname}")
assert full.status_code == 200
assert full.content == payload
ranged = client.get(f"/media/files/{fname}", headers={"Range": "bytes=0-9"})
assert ranged.status_code == 206
assert ranged.content == b"IMHEADDATA"
finally:
(host_media / fname).unlink(missing_ok=True)
def test_media_library_persists_in_volume(client):
"""The library JSON is written to the ./media bind mount (permanent).
Reads the repo's ./media directory — local deployments only."""
if not IS_LOCAL_DEPLOYMENT:
pytest.skip("reads the repo ./media bind mount — local deployments only")
import pathlib
upstream, url = _serve_fake_audio(b"\x00" * 16)
track_id = None
try:
track_id = client.post(
f"{MEDIA_API}/streams", json={"name": "Persist FM", "url": url}
).json()["track"]["id"]
library_file = pathlib.Path(__file__).parents[1] / "media" / "library.json"
data = json.loads(library_file.read_text())
assert track_id in data["tracks"]
finally:
upstream.shutdown()
if track_id:
try:
client.delete(f"{MEDIA_API}/tracks/{track_id}")
except httpx.HTTPError:
pass
# ── web player UI ───────────────────────────────────────────────────────────
def test_ui_media_panel_serves(client):
resp = client.get("/")
body = resp.text
assert 'id="mediaBtn"' in body
assert 'id="mediaPanel"' in body
assert 'id="playPauseBtn"' in body
assert 'id="seekBar"' in body
assert 'id="mediaAudio"' in body
def test_ui_media_logic_serves(client):
resp = client.get("/app.js")
body = resp.text
assert "reconcileMedia" in body
assert "media-control" in body # data-channel topic for the agent
assert "media_state" in body # playback state reports
assert "/media/api/state" in body # reconciliation polling
def test_ui_media_styles_serve(client):
resp = client.get("/style.css")
assert ".media-panel" in resp.text
assert ".media-row" in resp.text
def test_token_response_includes_room():
resp = httpx.post(TOKEN_URL, json={}, timeout=10, verify=False)
assert resp.status_code == 200
assert resp.json()["room"].startswith("voice-")
# ── Hope's media MCP toolset ────────────────────────────────────────────────
# These spawn the MCP server inside the LOCAL dev container (docker exec),
# whose media server is the local one — they can only be consistent when
# the test target IS the local deployment.
MEDIA_MCP_ENV = {
"MEDIA_SERVER_URL": "http://127.0.0.1:8093",
"MEDIA_ROOM": "mcp-test-room",
}
needs_local_container = pytest.mark.skipif(
not IS_LOCAL_DEPLOYMENT,
reason="docker-exec MCP probes run against the local dev container",
)
# ── media server internals (in-process, inside the container) ───────────────
@needs_local_container
def test_media_search_prefers_library_matches(client):
"""Search returns ready library matches ahead of YouTube results, so
Hope prefers an already-downloaded copy. Runs the endpoint in-process
with _run_search stubbed — fully deterministic, no network."""
code = '''
import sys
sys.path.insert(0, "/opt/voice")
import media_server as ms
from fastapi.testclient import TestClient
ms._library["tracks"]["libpref01"] = {
"id": "libpref01", "kind": "youtube", "video_id": "libpref01",
"title": "Preference Probe Song", "channel": "Probe Artist",
"duration": 120, "status": "ready", "ext": "webm",
"error": None, "added_at": 1.0,
}
ms._run_search = lambda q, limit: [
{"video_id": "remote01", "title": "Preference Probe Song (cover)",
"channel": "", "duration": 99, "view_count": 1,
"in_library": False, "track_id": None, "library_status": None},
]
c = TestClient(ms.app)
r = c.post("/media/api/search", json={"query": "preference probe", "limit": 5})
assert r.status_code == 200, r.text
results = r.json()["results"]
assert results, "no results"
assert results[0]["video_id"] == "libpref01", results
assert results[0]["in_library"] is True
assert any(x["video_id"] == "remote01" for x in results), results
print("OK")
'''
proc = docker_exec_python(code, timeout=60, check=False)
assert proc.returncode == 0, proc.stderr[-800:]
assert "OK" in proc.stdout
@needs_local_container
def test_media_stream_endpoint_serves_ready_track(client):
"""GET /media/api/stream/{id} for a finished youtube track serves the
file bytes with native Range support (progressive-playback endpoint)."""
code = '''
import sys
sys.path.insert(0, "/opt/voice")
import media_server as ms
from fastapi.testclient import TestClient
probe = ms.FILES_DIR / "probetrk.mp3"
probe.write_bytes(b"PROBEDATA" * 100)
ms._library["tracks"]["probetrk"] = {
"id": "probetrk", "kind": "youtube", "video_id": "probetrk",
"title": "Probe", "channel": "", "duration": 1,
"status": "ready", "ext": "mp3", "error": None, "added_at": 1.0,
}
try:
c = TestClient(ms.app)
full = c.get("/media/api/stream/probetrk")
assert full.status_code == 200, full.status_code
assert full.content == b"PROBEDATA" * 100
part = c.get("/media/api/stream/probetrk", headers={"Range": "bytes=0-8"})
assert part.status_code == 206, part.status_code
assert part.content == b"PROBEDATA"
assert (part.headers.get("content-range") or "").startswith("bytes 0-8/")
# track dict exposes a stable stream URL while downloading and ready
d = ms._track_dict(ms._library["tracks"]["probetrk"])
assert d["audio_url"] == "/media/api/stream/probetrk", d
print("OK")
finally:
probe.unlink(missing_ok=True)
ms._library["tracks"].pop("probetrk", None)
'''
proc = docker_exec_python(code, timeout=60, check=False)
assert proc.returncode == 0, proc.stderr[-800:]
assert "OK" in proc.stdout
@needs_local_container
def test_media_play_resolves_cached_ready_track(client):
"""Playing an ALREADY-cached song must resolve to the on-disk ready track
(no redownload) and hand the browser a playable stream URL, so a cached
pick starts instantly rather than re-fetching from YouTube.
Regression: a cached track could be left as paused with currentTime 0 if
the room the agent played into diverged from the room the browser polled,
or if autoplay was blocked. This pins the server-side resolve + track
dict so a ready file is always returned with a stable audio_url."""
code = '''
import sys
sys.path.insert(0, "/opt/voice")
import media_server as ms
from fastapi.testclient import TestClient
probe = ms.FILES_DIR / "cachetrk.webm"
probe.write_bytes(b"CACHEDBYTES" * 40)
ms._library["tracks"]["cachetrk"] = {
"id": "cachetrk", "kind": "youtube", "video_id": "cachetrk",
"title": "Cached Song", "channel": "Artist", "duration": 200,
"status": "ready", "ext": "webm", "error": None, "added_at": 1.0,
}
ms._start_download = lambda *a, **k: (_ for _ in ()).throw(
AssertionError("must not re-download an already cached track")
)
try:
c = TestClient(ms.app)
# play by video_id (as Hope's play_song does) must reuse the ready track
r = c.post("/media/api/play", json={
"room": "cached-test-room", "video_id": "cachetrk",
"title": "Cached Song", "channel": "Artist",
}, timeout=10)
assert r.status_code == 200, r.text
body = r.json()
assert body["desired"] == "playing", body
track = body["track"]
assert track["status"] == "ready", track
assert track["audio_url"] == "/media/api/stream/cachetrk", track
# the stream endpoint serves the cached bytes (browser plays it directly)
data = c.get("/media/api/stream/cachetrk")
assert data.status_code == 200
assert data.content == b"CACHEDBYTES" * 40
print("OK")
finally:
probe.unlink(missing_ok=True)
ms._library["tracks"].pop("cachetrk", None)
'''
proc = docker_exec_python(code, timeout=60, check=False)
assert proc.returncode == 0, proc.stderr[-800:]
assert "OK" in proc.stdout
@needs_local_container
def test_mcp_media_list_library(mcp_client_factory):
@@ -490,8 +118,9 @@ def test_mcp_media_list_library(mcp_client_factory):
@needs_local_container
def test_mcp_media_add_stream_and_now_playing(mcp_client_factory, client):
def test_mcp_media_add_stream_and_now_playing(mcp_client_factory):
upstream, url = _serve_fake_audio(b"\x00" * 16)
track_id = None
try:
text, is_error = mcp_client_factory(
"media_mcp.py", "add_stream",
@@ -500,99 +129,39 @@ def test_mcp_media_add_stream_and_now_playing(mcp_client_factory, client):
assert not is_error, text
assert "Added" in text
library = client.get(f"{MEDIA_API}/library").json()["tracks"]
probe = next(t for t in library if t["title"] == "MCP Test Radio")
try:
text, is_error = mcp_client_factory(
"media_mcp.py", "now_playing", {}, env=MEDIA_MCP_ENV
)
assert not is_error
# Nothing loaded in mcp-test-room: the tool must say so plainly.
assert "Nothing is loaded" in text
with httpx.Client(base_url=MUSIC_BASE_URL, timeout=15) as music:
library = music.get(f"{MEDIA_API}/library").json()["tracks"]
probe = next(t for t in library if t["title"] == "MCP Test Radio")
track_id = probe["id"]
music.post(
f"{MEDIA_API}/play",
json={"room": "mcp-test-room", "track_id": track_id},
).raise_for_status()
client.post(
f"{MEDIA_API}/play", json={"room": "mcp-test-room", "track_id": probe["id"]}
)
text, is_error = mcp_client_factory(
"media_mcp.py", "now_playing", {}, env=MEDIA_MCP_ENV
)
assert not is_error
assert "MCP Test Radio" in text
finally:
client.delete(f"{MEDIA_API}/tracks/{probe['id']}")
text, is_error = mcp_client_factory(
"media_mcp.py", "now_playing", {}, env=MEDIA_MCP_ENV
)
assert not is_error
assert "MCP Test Radio" in text
finally:
upstream.shutdown()
if track_id:
with httpx.Client(base_url=MUSIC_BASE_URL, timeout=15) as music:
music.delete(f"{MEDIA_API}/tracks/{track_id}")
@needs_local_container
def test_mcp_media_resume_without_load(mcp_client_factory, client):
"""Errors from the media server reach the LLM as tool text (not MCP
errors) so Hope can explain them naturally."""
client.post(
f"{MEDIA_API}/control", json={"room": "mcp-empty-room", "action": "stop"}
)
def test_mcp_media_resume_without_load(mcp_client_factory):
room = "mcp-empty-room"
with httpx.Client(base_url=MUSIC_BASE_URL, timeout=15) as music:
music.post(
f"{MEDIA_API}/control",
json={"room": room, "action": "stop"},
).raise_for_status()
text, is_error = mcp_client_factory(
"media_mcp.py", "resume_music", {}, env={
"MEDIA_SERVER_URL": "http://127.0.0.1:8093",
"MEDIA_ROOM": "mcp-empty-room",
}
"media_mcp.py", "resume_music", {},
env={"MUSIC_BASE_URL": MUSIC_BASE_URL, "MUSIC_ROOM": room},
)
assert not is_error
assert "nothing loaded" in text.lower()
# ── live YouTube Music (network-dependent) ──────────────────────────────────
@pytest.mark.slow
def test_media_search_live(client):
resp = client.post(
f"{MEDIA_API}/search",
json={"query": "blind guardian timeless miracle", "limit": 3},
)
assert resp.status_code == 200
results = resp.json()["results"]
assert len(results) >= 1
for r in results:
assert r["video_id"]
assert r["title"]
@pytest.mark.slow
def test_media_download_and_stream_end_to_end(client):
"""Full pipeline: search -> play -> yt-dlp download -> playable stream.
The audio_url stays on /media/api/stream/ for the whole lifecycle; once
ready it serves the finished file with Range support. Gated: run with
HOPE_RUN_YTDLP_DOWNLOAD=1 (takes 1-3 minutes)."""
if os.environ.get("HOPE_RUN_YTDLP_DOWNLOAD") != "1":
pytest.skip("set HOPE_RUN_YTDLP_DOWNLOAD=1 to run the real download")
search = client.post(
f"{MEDIA_API}/search", json={"query": "kingdom hearts dearly beloved", "limit": 3}
)
results = search.json()["results"]
assert results, "search returned nothing"
video = results[0]
played = client.post(
f"{MEDIA_API}/play",
json={"room": TEST_ROOM, "video_id": video["video_id"],
"title": video["title"], "channel": video["channel"]},
).json()
track_id = played["track"]["id"]
try:
deadline = time.time() + 240
track = played["track"]
while time.time() < deadline:
library = client.get(f"{MEDIA_API}/library").json()["tracks"]
track = next(t for t in library if t["id"] == track_id)
if track["status"] in ("ready", "failed"):
break
time.sleep(3)
assert track["status"] == "ready", f"download failed: {track.get('error')}"
stream = client.get(track["audio_url"], headers={"Range": "bytes=0-1023"})
assert stream.status_code in (200, 206)
assert len(stream.content) >= 1
finally:
client.delete(f"{MEDIA_API}/tracks/{track_id}")