macOS has no D-Bus/MPRIS: the audio engine is what the OS media UI talks to. mpv's Cocoa backend registers with MPRemoteCommandCenter and MPNowPlayingInfoCenter, so it already publishes the live ICY title and receives the media keys. Add a platform-neutral media session (mediasession.py) so the app keeps a single interface: MprisService on D-Bus platforms, NowPlayingSession on macOS. Map mpv's NEXT/PREV/PLAY/PLAYONLY/PAUSEONLY/STOP keys onto script-messages and read them back as client-message IPC events, so the hardware media keys and Control Center drive the TUI (station changes, play/pause/stop) instead of fighting it. --no-mpris (now also --no-media) disables the media keys too. macOS-only mpv options are gated on sys.platform so the Linux command line is unchanged; mpv aborts on unknown flags.
180 lines
5.8 KiB
Python
180 lines
5.8 KiB
Python
from __future__ import annotations
|
|
|
|
import html
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from somafm_tui.api import Song
|
|
from somafm_tui.app import SomaFMApp
|
|
from somafm_tui.widgets.help import HelpScreen
|
|
from somafm_tui.widgets.stage import StagePanel
|
|
from somafm_tui.widgets.station_list import StationList
|
|
|
|
|
|
def rendered_text(app: SomaFMApp) -> str:
|
|
svg = html.unescape(app.export_screenshot()).replace("\xa0", " ")
|
|
return re.sub(r"<[^>]+>", " ", svg)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_boot_navigation_filter_favourite_visualizer_help() -> None:
|
|
app = SomaFMApp(autoplay=False, offline=True, audio=False)
|
|
async with app.run_test(size=(110, 36)) as pilot:
|
|
await pilot.pause()
|
|
|
|
assert len(app.stations) >= 40
|
|
assert app.current is not None
|
|
assert app.current.id == "groovesalad"
|
|
assert app.focused is None
|
|
assert len(app.view) == len(app.stations)
|
|
|
|
start = app.index
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
assert app.index == min(start + 1, len(app.view) - 1)
|
|
|
|
await pilot.press("g")
|
|
await pilot.pause()
|
|
assert app.index == 0
|
|
|
|
await pilot.press("shift+g")
|
|
await pilot.pause()
|
|
assert app.index == len(app.view) - 1
|
|
|
|
await pilot.press("slash")
|
|
await pilot.pause()
|
|
search = app.query_one("#search")
|
|
assert search.has_class("visible")
|
|
assert app.focused is search
|
|
|
|
for char in "defcon":
|
|
await pilot.press(char)
|
|
await pilot.pause()
|
|
assert app.query == "defcon"
|
|
assert [station.id for station in app.view] == ["defcon"]
|
|
|
|
await pilot.press("enter")
|
|
await pilot.pause()
|
|
assert app.current is not None
|
|
assert app.current.id == "defcon"
|
|
assert app.query == ""
|
|
assert len(app.view) == len(app.stations)
|
|
assert app.playing_state in ("stopped", "error")
|
|
|
|
app.action_favorite()
|
|
await pilot.pause()
|
|
assert "defcon" in app.config.favorites
|
|
assert app.view[0].id == "defcon"
|
|
assert app.index == 0
|
|
|
|
modes = set()
|
|
for _ in range(3):
|
|
await pilot.press("v")
|
|
await pilot.pause()
|
|
modes.add(app.visualizer)
|
|
assert modes == {"both", "bars", "wave"}
|
|
|
|
await pilot.press("question_mark")
|
|
await pilot.pause()
|
|
assert isinstance(app.screen, HelpScreen)
|
|
await pilot.press("x")
|
|
await pilot.pause()
|
|
assert not isinstance(app.screen, HelpScreen)
|
|
|
|
text = rendered_text(app)
|
|
for needle in ("SOMA FM RADIO", "STATIONS", "ON AIR", "SPECTRUM"):
|
|
assert needle in text
|
|
assert "DEF CON" in text.upper()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stage_renders_history_and_listener_count() -> None:
|
|
app = SomaFMApp(autoplay=False, offline=True, audio=False)
|
|
async with app.run_test(size=(110, 40)) as pilot:
|
|
await pilot.pause()
|
|
station = app.current
|
|
assert station is not None
|
|
station.listeners = 1234
|
|
now = datetime.now(timezone.utc)
|
|
station.now_playing = Song(
|
|
title="Rauschebart",
|
|
artist="Jens Buchert",
|
|
album="Gravity",
|
|
played_at=now - timedelta(seconds=26),
|
|
)
|
|
station.history = [
|
|
station.now_playing,
|
|
Song(
|
|
title="Room 505",
|
|
artist="Alex Cortiz",
|
|
album="The Chill Out Room",
|
|
played_at=now - timedelta(minutes=5),
|
|
),
|
|
]
|
|
app.query_one(StagePanel).refresh_stage()
|
|
app.query_one(StationList).refresh()
|
|
await pilot.pause()
|
|
|
|
text = rendered_text(app)
|
|
for needle in ("LAST PLAYED", "Jens Buchert", "Gravity", "1.2k listeners"):
|
|
assert needle in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_narrow_layout() -> None:
|
|
app = SomaFMApp(autoplay=False, offline=True, audio=False)
|
|
async with app.run_test(size=(64, 20)) as pilot:
|
|
await pilot.pause()
|
|
app.query_one(StagePanel).refresh_stage()
|
|
app.query_one(StationList).refresh()
|
|
await pilot.pause()
|
|
text = rendered_text(app)
|
|
assert "STATIONS" in text
|
|
assert "ON AIR" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stage_shows_bandcamp_link_only_when_present() -> None:
|
|
app = SomaFMApp(autoplay=False, offline=True, audio=False)
|
|
async with app.run_test(size=(110, 40)) as pilot:
|
|
await pilot.pause()
|
|
station = app.current
|
|
assert station is not None
|
|
song = Song(
|
|
title="Rauschebart",
|
|
artist="Jens Buchert",
|
|
album="Gravity",
|
|
bandcamp_url="https://jensbuchert.bandcamp.com/track/rauschebart",
|
|
)
|
|
station.now_playing = song
|
|
app.query_one(StagePanel).refresh_stage()
|
|
await pilot.pause()
|
|
assert "Bandcamp" in rendered_text(app)
|
|
|
|
song.bandcamp_url = ""
|
|
app.query_one(StagePanel).refresh_stage()
|
|
await pilot.pause()
|
|
assert "Bandcamp" not in rendered_text(app)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_media_session_commands_drive_the_app() -> None:
|
|
app = SomaFMApp(autoplay=False, offline=True, audio=False)
|
|
async with app.run_test(size=(100, 30)) as pilot:
|
|
await pilot.pause()
|
|
# The status bar names whichever backend this platform uses.
|
|
assert app.media.label in {"mpris", "nowplaying"}
|
|
assert app.media.status_label == "off" # audio disabled
|
|
|
|
# The ticker drains the queue, so give it a couple of ticks.
|
|
before = app.current.id
|
|
app.media_actions.put("next")
|
|
await pilot.pause(0.2)
|
|
assert app.current.id != before
|
|
|
|
app.media_actions.put("volume:25")
|
|
await pilot.pause(0.2)
|
|
assert app.volume == 25
|