hope-voice-api/tests/test_media.py

"""Voice-side integration tests for the standalone Hope Music service.

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 os
import socket
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.parse import urlparse

import httpx
import pytest

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_MCP_ENV = {
    "MUSIC_BASE_URL": MUSIC_BASE_URL,
    "MUSIC_ROOM": "mcp-test-room",
}


def _serve_fake_audio(payload: bytes):
    """Tiny Range-capable upstream reachable by a host-networked container."""

    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            self.send_response(200)
            self.send_header("Content-Type", "audio/mpeg")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)

        def log_message(self, format, *args):  # noqa: A002
            pass

    server = HTTPServer(("0.0.0.0", 0), Handler)
    port = server.server_address[1]
    probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        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:
        host = "127.0.0.1"
    finally:
        probe.close()
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server, f"http://{host}:{port}/audio.mp3"



@needs_local_container
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()


@needs_local_container
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",
            {"name": "MCP Test Radio", "url": url}, env=MEDIA_MCP_ENV,
        )
        assert not is_error, text
        assert "Added" 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()

        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):
    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={"MUSIC_BASE_URL": MUSIC_BASE_URL, "MUSIC_ROOM": room},
    )
    assert not is_error
    assert "nothing loaded" in text.lower()