beefy-player/tests/test_player.py

from __future__ import annotations

import os
import queue
import time
from pathlib import Path

from somafm_tui.player import Player

FAKE_MPV = '''#!/usr/bin/env python3
import json, os, socket, sys, time

sock_path = None
volume = 70
for arg in sys.argv[1:]:
    if arg.startswith("--input-ipc-server="):
        sock_path = arg.split("=", 1)[1]
    elif arg.startswith("--volume="):
        volume = int(arg.split("=", 1)[1])

log_path = os.environ.get("FAKE_MPV_LOG")
def log(line):
    if log_path:
        with open(log_path, "a") as handle:
            handle.write(line + "\\n")

server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
    os.unlink(sock_path)
except FileNotFoundError:
    pass
server.bind(sock_path)
server.listen(1)
conn, _ = server.accept()
stream = conn.makefile("rwb", buffering=0)

def send(payload):
    stream.write((json.dumps(payload) + "\\n").encode())

while True:
    line = stream.readline()
    if not line:
        break
    try:
        message = json.loads(line)
    except ValueError:
        continue
    command = message.get("command") or []
    if not command:
        continue
    name = command[0]
    if name == "observe_property":
        prop = command[2]
        if prop == "volume":
            send({"event": "property-change", "id": command[1], "name": "volume", "data": volume})
        elif prop == "pause":
            send({"event": "property-change", "id": command[1], "name": "pause", "data": False})
    elif name == "loadfile":
        url = command[1]
        log(url)
        send({"event": "start-file"})
        if "bad" in url:
            time.sleep(0.05)
            send({"event": "end-file", "reason": "error"})
        else:
            time.sleep(0.05)
            send({"event": "file-loaded"})
            send({"event": "property-change", "id": 1, "name": "pause", "data": False})
    elif name == "set_property":
        prop, value = command[1], command[2]
        if prop == "pause":
            send({"event": "property-change", "id": 1, "name": "pause", "data": bool(value)})
        elif prop == "volume":
            volume = int(value)
            send({"event": "property-change", "id": 2, "name": "volume", "data": volume})
    elif name == "stop":
        send({"event": "end-file", "reason": "stop"})
    elif name == "quit":
        break

try:
    stream.close()
except Exception:
    pass
conn.close()
server.close()
'''


def wait_for(player: Player, predicate, timeout: float = 8.0):
    deadline = time.monotonic() + timeout
    seen = []
    while time.monotonic() < deadline:
        try:
            event = player.events.get(timeout=0.1)
        except queue.Empty:
            continue
        seen.append(event)
        if predicate(event):
            return event, seen
    raise AssertionError(f"timed out; last events: {seen[-8:]}")


def install_fake_mpv(tmp_path: Path, monkeypatch, log_path: Path) -> None:
    script = tmp_path / "mpv"
    script.write_text(FAKE_MPV)
    script.chmod(0o755)
    monkeypatch.setenv("PATH", f"{tmp_path}{os.pathsep}{os.environ.get('PATH', '')}")
    monkeypatch.setenv("FAKE_MPV_LOG", str(log_path))


def test_mpv_ipc_play_pause_volume_retry_stop(tmp_path, monkeypatch) -> None:
    log_path = tmp_path / "mpv.log"
    install_fake_mpv(tmp_path, monkeypatch, log_path)

    player = Player()
    player.start()
    assert player.backend == "mpv"
    player.set_volume(55)
    player.play(
        [
            "https://example.invalid/bad-stream",
            "https://example.invalid/good-stream",
        ]
    )
    try:
        event, _ = wait_for(
            player,
            lambda ev: ev.kind == "state"
            and ev.state == "playing"
            and "good-stream" in ev.url,
            timeout=10,
        )
        assert event.state == "playing"
        assert "bad-stream" in log_path.read_text()
        assert "good-stream" in log_path.read_text()

        player.pause()
        wait_for(player, lambda ev: ev.kind == "state" and ev.state == "paused")

        player.resume()
        wait_for(player, lambda ev: ev.kind == "state" and ev.state == "playing")

        player.set_volume(42)
        volume_event, _ = wait_for(
            player, lambda ev: ev.kind == "volume" and ev.volume == 42
        )
        assert volume_event.volume == 42

        player.stop()
        wait_for(player, lambda ev: ev.kind == "state" and ev.state == "stopped")
    finally:
        player.shutdown()
        if player._thread is not None:
            player._thread.join(timeout=2)
        assert player._thread is None or not player._thread.is_alive()


def test_backend_detection_without_tools(tmp_path, monkeypatch) -> None:
    empty = tmp_path / "empty"
    empty.mkdir()
    monkeypatch.setenv("PATH", str(empty))
    player = Player()
    player.start()
    assert player.backend is None
    event, _ = wait_for(player, lambda ev: ev.kind == "state", timeout=1)
    assert event.state == "error"
    player.shutdown()


def test_media_key_messages_become_player_events() -> None:
    player = Player()
    player._handle_mpv_message(
        {"event": "client-message", "args": ["somafm", "next"]}
    )
    event = player.events.get_nowait()
    assert event.kind == "media"
    assert event.detail == "next"

    player._handle_mpv_message(
        {"event": "client-message", "args": ["somafm", "toggle"]}
    )
    assert player.events.get_nowait().detail == "toggle"

    # Foreign namespaces, unknown commands and malformed payloads are ignored.
    player._handle_mpv_message({"event": "client-message", "args": ["other", "next"]})
    player._handle_mpv_message({"event": "client-message", "args": ["somafm", "seek"]})
    player._handle_mpv_message({"event": "client-message", "args": None})
    player._handle_mpv_message({"event": "client-message"})
    assert player.events.empty()


def test_platform_options_never_leak_to_linux(tmp_path, monkeypatch) -> None:
    import somafm_tui.player as player_module

    monkeypatch.setattr(player_module, "RUNTIME_DIR", tmp_path)
    assert Player()._platform_options("linux") == []
    assert Player()._platform_options("win32") == []

    options = Player()._platform_options("darwin")
    assert "--input-media-keys=yes" in options
    conf = next(option for option in options if option.startswith("--input-conf="))
    assert conf.endswith("mpv-input.conf")
    bindings = (tmp_path / "mpv-input.conf").read_text()
    for line in (
        "PLAY script-message somafm toggle",
        "PLAYONLY script-message somafm play",
        "PAUSEONLY script-message somafm pause",
        "STOP script-message somafm stop",
        "NEXT script-message somafm next",
        "PREV script-message somafm prev",
    ):
        assert line in bindings

    # --no-mpris / --no-media must silence the media keys as well.
    assert Player(media_keys=False)._platform_options("darwin") == [
        "--input-media-keys=no"
    ]