from __future__ import annotations import os import queue import time from pathlib import Path from beefy_player.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": ["beefy", "next"]} ) event = player.events.get_nowait() assert event.kind == "media" assert event.detail == "next" player._handle_mpv_message( {"event": "client-message", "args": ["beefy", "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": ["beefy", "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 beefy_player.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 beefy toggle", "PLAYONLY script-message beefy play", "PAUSEONLY script-message beefy pause", "STOP script-message beefy stop", "NEXT script-message beefy next", "PREV script-message beefy 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" ]