hope-voice-api/tests/test_media.py

"""Tests for the media player subsystem.

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).
"""

import json
import os
import subprocess
import threading
import time
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer

import httpx
import pytest

from conftest import TOKEN_URL, docker_exec

MEDIA_API = "/media/api"
TEST_ROOM = f"media-test-{uuid.uuid4().hex[:8]}"


# ── local upstream for stream-proxy tests ───────────────────────────────────

def _serve_fake_audio(payload: bytes):
    """Tiny Range-capable HTTP server on an ephemeral host port.

    The container uses host networking, so the media server inside the
    container reaches it on 127.0.0.1.
    """

    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_header("Content-Type", "audio/mpeg")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Accept-Ranges", "bytes")
            self.end_headers()
            self.wfile.write(body)

        def log_message(self, *args):
            pass

    server = HTTPServer(("127.0.0.1", 0), Handler)
    port = server.server_address[1]
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server, f"http://127.0.0.1:{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."""
    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


def test_media_files_static_range_support(client):
    """Downloaded tracks are nginx static files with native Range support."""
    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)."""
    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)
    assert resp.status_code == 200
    assert resp.json()["room"].startswith("voice-")


# ── Hope's media MCP toolset ────────────────────────────────────────────────

MEDIA_MCP_ENV = {
    "MEDIA_SERVER_URL": "http://127.0.0.1:8092",
    "MEDIA_ROOM": "mcp-test-room",
}


def test_mcp_media_list_library(mcp_client_factory):
    text, is_error = mcp_client_factory(
        "media_mcp.py", "list_library", {}, env=MEDIA_MCP_ENV
    )
    assert not is_error
    assert "library" in text.lower()


def test_mcp_media_add_stream_and_now_playing(mcp_client_factory, client):
    upstream, url = _serve_fake_audio(b"\x00" * 16)
    try:
        text, is_error = mcp_client_factory(
            "media_mcp.py", "add_stream",
            {"name": "MCP Test Radio", "url": url}, env=MEDIA_MCP_ENV,
        )
        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

            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']}")
    finally:
        upstream.shutdown()


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"}
    )
    text, is_error = mcp_client_factory(
        "media_mcp.py", "resume_music", {}, env={
            "MEDIA_SERVER_URL": "http://127.0.0.1:8092",
            "MEDIA_ROOM": "mcp-empty-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 + ffmpeg transcode ->
    static mp3 with Range support in the player. 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}")