329 lines
11 KiB
Python
329 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import html
|
|
import json
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from beefy_player import config
|
|
from beefy_player.models import Song, Station
|
|
from beefy_player.providers import PROVIDERS, available_providers, get_provider
|
|
from beefy_player.providers.base import RadioProvider
|
|
from beefy_player.providers.bassdrive import BassdriveProvider
|
|
from beefy_player.providers.somafm import SomaFMProvider
|
|
from beefy_player.providers.tilderadio import TilderadioProvider
|
|
from beefy_player.widgets.header import HeaderBar
|
|
|
|
|
|
def rendered_text(app) -> str:
|
|
svg = html.unescape(app.export_screenshot()).replace("\xa0", " ")
|
|
return re.sub(r"<[^>]+>", " ", svg)
|
|
|
|
|
|
def test_registry_lists_somafm_as_the_default() -> None:
|
|
# the menu is alphabetical by name: aNONradio ... tilderadio
|
|
assert available_providers() == (
|
|
"anonradio",
|
|
"bassdrive",
|
|
"cashmere",
|
|
"datafruits",
|
|
"dublab",
|
|
"lyl",
|
|
"radiobrowser",
|
|
"radioparadise",
|
|
"refuge",
|
|
"rinse",
|
|
"somafm",
|
|
"lot",
|
|
"tilderadio",
|
|
)
|
|
assert isinstance(get_provider(), SomaFMProvider)
|
|
assert get_provider("somafm") is get_provider()
|
|
assert isinstance(get_provider("tilderadio"), TilderadioProvider)
|
|
assert isinstance(get_provider("bassdrive"), BassdriveProvider)
|
|
# a stale slug in the config file must not stop the app from starting
|
|
assert get_provider("no-such-catalogue") is get_provider()
|
|
|
|
|
|
def test_somafm_provider_describes_its_catalogue() -> None:
|
|
provider = get_provider()
|
|
assert provider.name == "SomaFM"
|
|
assert provider.brand == "SOMA FM"
|
|
assert provider.station_source == "somafm.com/channels.json"
|
|
assert "somafm.com" in provider.user_agent
|
|
|
|
stations = provider.fallback_stations()
|
|
assert len(stations) > 40
|
|
station = stations[0]
|
|
assert station.provider == "somafm"
|
|
assert station.uid == f"somafm:{station.id}"
|
|
assert all("somafm.com" in url for url in provider.fallback_urls(station))
|
|
|
|
|
|
class FakeProvider(RadioProvider):
|
|
"""A second catalogue, as small as the interface allows."""
|
|
|
|
id = "fake"
|
|
name = "Fake FM"
|
|
homepage = "fake.test"
|
|
brand = "FAKE FM"
|
|
tagline = "not real radio"
|
|
station_source = "fake.test/stations.json"
|
|
history_source = "recent.json"
|
|
|
|
def __init__(self) -> None:
|
|
self.station = Station(
|
|
id="one", title="Fake One", genre="test", provider=self.id
|
|
)
|
|
|
|
async def fetch_stations(self, client):
|
|
return [self.station]
|
|
|
|
def load_cached_stations(self):
|
|
return [self.station]
|
|
|
|
def fallback_stations(self):
|
|
return [self.station]
|
|
|
|
async def fetch_history(self, client, station_id):
|
|
return [Song(title="Tune", artist="Someone")]
|
|
|
|
def load_cached_history(self, station_id):
|
|
return []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_app_runs_against_a_different_catalogue(monkeypatch) -> None:
|
|
"""The UI must not know which catalogue it is showing."""
|
|
from beefy_player.app import BeefyPlayerApp
|
|
|
|
fake = FakeProvider()
|
|
monkeypatch.setitem(PROVIDERS, "fake", fake)
|
|
app = BeefyPlayerApp(provider="fake", autoplay=False, offline=True, audio=False)
|
|
assert app.provider is fake
|
|
async with app.run_test(size=(100, 30)) as pilot:
|
|
await pilot.pause()
|
|
assert app.current is not None
|
|
assert app.current.uid == "fake:one"
|
|
app.query_one("StagePanel").refresh_stage()
|
|
await pilot.pause()
|
|
text = rendered_text(app)
|
|
assert "FAKE ONE" in text
|
|
assert "Fake FM" in app.title
|
|
|
|
|
|
def test_config_remembers_the_provider() -> None:
|
|
assert config.Config.load().provider == "somafm"
|
|
config.Config(provider="fake").save()
|
|
assert config.Config.load().provider == "fake"
|
|
|
|
|
|
def test_legacy_config_is_migrated_once(monkeypatch) -> None:
|
|
legacy = config.LEGACY_CONFIG_PATH
|
|
legacy.parent.mkdir(parents=True, exist_ok=True)
|
|
legacy.write_text(
|
|
json.dumps({"favorites": ["defcon"], "volume": 33, "last_station": "dronezone"})
|
|
)
|
|
|
|
loaded = config.Config.load()
|
|
assert loaded.favorites == ["defcon"]
|
|
assert loaded.volume == 33
|
|
assert loaded.last_station == "dronezone"
|
|
assert config.CONFIG_PATH.exists()
|
|
|
|
# the new file wins from then on
|
|
config.Config(volume=11).save()
|
|
legacy.write_text(json.dumps({"volume": 99}))
|
|
assert config.Config.load().volume == 11
|
|
|
|
|
|
def test_environment_variables_prefer_the_beefy_prefix(monkeypatch) -> None:
|
|
monkeypatch.setenv("BEEFY_IMAGES", "tgp")
|
|
monkeypatch.setenv("SOMAFM_IMAGES", "0")
|
|
assert config.env_var("IMAGES") == "tgp"
|
|
|
|
monkeypatch.delenv("BEEFY_IMAGES")
|
|
assert config.env_var("IMAGES") == "0" # the old spelling still works
|
|
|
|
monkeypatch.delenv("SOMAFM_IMAGES")
|
|
assert config.env_var("IMAGES") == ""
|
|
|
|
|
|
def test_tilderadio_replaces_the_station_list_with_a_schedule() -> None:
|
|
provider = TilderadioProvider()
|
|
assert provider.has_schedule is True
|
|
# the SomaFM wordmark's slot now holds tilderadio's own art and motto
|
|
assert provider.hero and not provider.hero_gradient
|
|
assert provider.hero_lines == (
|
|
"TILDERADIO",
|
|
"STREAMED BY / FOR USERS OF THE",
|
|
"TILDEVERSE",
|
|
)
|
|
assert SomaFMProvider().has_schedule is False
|
|
|
|
|
|
class _CatalogueApp:
|
|
"""A headless app on a fixed catalogue, for the picker tests."""
|
|
|
|
def __init__(self, provider: str = "somafm") -> None:
|
|
from beefy_player.app import BeefyPlayerApp
|
|
|
|
self.app = BeefyPlayerApp(
|
|
provider=provider, autoplay=False, offline=True, audio=False
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p_opens_the_picker_and_switches_catalogue() -> None:
|
|
from beefy_player.widgets.providers import ProviderScreen
|
|
from beefy_player.widgets.schedule import SchedulePanel
|
|
from beefy_player.widgets.station_list import StationList
|
|
|
|
app = _CatalogueApp().app
|
|
async with app.run_test(size=(110, 40)) as pilot:
|
|
await pilot.pause()
|
|
assert app.provider.id == "somafm"
|
|
assert app.query_one(StationList).display is True
|
|
assert app.query_one(SchedulePanel).display is False
|
|
assert app.query_one(HeaderBar).styles.height.value == 4
|
|
|
|
await pilot.press("p")
|
|
await pilot.pause()
|
|
assert isinstance(app.screen, ProviderScreen)
|
|
|
|
# walk down the alphabetical menu from the current entry to tilderadio
|
|
slugs = available_providers()
|
|
steps = (slugs.index("tilderadio") - slugs.index("somafm")) % len(slugs)
|
|
for _ in range(steps):
|
|
await pilot.press("down")
|
|
await pilot.pause()
|
|
await pilot.press("enter")
|
|
await pilot.pause()
|
|
await pilot.pause()
|
|
|
|
assert app.provider.id == "tilderadio"
|
|
assert app.config.provider == "tilderadio"
|
|
assert [station.id for station in app.stations] == ["tilderadio"]
|
|
# the left pane becomes the schedule; every hero is the same height
|
|
assert app.query_one(StationList).display is False
|
|
assert app.query_one(SchedulePanel).display is True
|
|
assert app.query_one(HeaderBar).styles.height.value == 4
|
|
assert "tilderadio" in app.title
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_escape_closes_the_picker_without_switching() -> None:
|
|
app = _CatalogueApp().app
|
|
async with app.run_test(size=(110, 40)) as pilot:
|
|
await pilot.pause()
|
|
await pilot.press("p")
|
|
await pilot.pause()
|
|
await pilot.press("escape")
|
|
await pilot.pause()
|
|
assert app.provider.id == "somafm"
|
|
assert app.config.provider != "tilderadio"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_help_names_both_catalogues_and_the_switch_key() -> None:
|
|
app = _CatalogueApp().app
|
|
async with app.run_test(size=(120, 60)) as pilot:
|
|
await pilot.pause()
|
|
await pilot.press("question_mark")
|
|
await pilot.pause()
|
|
text = rendered_text(app)
|
|
assert "CATALOGUES" in text
|
|
assert "somafm" in text and "tilderadio" in text
|
|
assert "switch station catalogue" in text
|
|
|
|
|
|
def test_config_remembers_a_station_per_catalogue() -> None:
|
|
config.Config(last_stations={"somafm": "defcon", "tilderadio": "tilderadio"}).save()
|
|
loaded = config.Config.load()
|
|
assert loaded.last_station_for("somafm") == "defcon"
|
|
assert loaded.last_station_for("tilderadio") == "tilderadio"
|
|
# a catalogue nobody has tuned yet falls back to the legacy single field
|
|
assert loaded.last_station_for("no-such-catalogue") == loaded.last_station
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_every_catalogue_header_is_the_same_height() -> None:
|
|
for slug in available_providers():
|
|
app = _CatalogueApp(slug).app
|
|
async with app.run_test(size=(110, 40)) as pilot:
|
|
await pilot.pause()
|
|
header = app.query_one(HeaderBar)
|
|
assert header.styles.height.value == 4, slug
|
|
lines = [line.plain for line in header.render().renderables]
|
|
assert any("STOPPED" in line or "PLAYING" in line for line in lines), slug
|
|
|
|
|
|
class _StubPlayer:
|
|
"""Just enough Player for a tune: records what mpv would have played."""
|
|
|
|
def __init__(self) -> None:
|
|
import queue
|
|
|
|
self.events: queue.Queue = queue.Queue()
|
|
self.played: list[list[str]] = []
|
|
self.volume: int | None = None
|
|
self.backend: str | None = None
|
|
|
|
def play(self, urls) -> None:
|
|
self.played.append(list(urls))
|
|
|
|
def set_volume(self, value: int) -> None:
|
|
self.volume = value
|
|
|
|
def stop(self) -> None:
|
|
pass
|
|
|
|
def toggle_pause(self) -> None:
|
|
pass
|
|
|
|
|
|
async def _switch_to(app, pilot, slug: str) -> None:
|
|
slugs = available_providers()
|
|
steps = (slugs.index(slug) - slugs.index(app.provider.id)) % len(slugs)
|
|
await pilot.press("p")
|
|
await pilot.pause()
|
|
for _ in range(steps):
|
|
await pilot.press("down")
|
|
await pilot.press("enter")
|
|
await pilot.pause()
|
|
await pilot.pause()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switching_to_a_single_station_catalogue_tunes_it() -> None:
|
|
app = _CatalogueApp("somafm").app
|
|
async with app.run_test(size=(110, 40)) as pilot:
|
|
await pilot.pause()
|
|
stub = _StubPlayer()
|
|
app.player = stub
|
|
app.audio_enabled = True
|
|
|
|
await _switch_to(app, pilot, "tilderadio")
|
|
|
|
assert app.provider.id == "tilderadio"
|
|
assert app.current is not None and app.current.id == "tilderadio"
|
|
assert stub.played, "a one-station catalogue should start playing"
|
|
assert app.playing_state != "stopped"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switching_to_a_multi_station_catalogue_waits_for_enter() -> None:
|
|
app = _CatalogueApp("tilderadio").app
|
|
async with app.run_test(size=(110, 40)) as pilot:
|
|
await pilot.pause()
|
|
stub = _StubPlayer()
|
|
app.player = stub
|
|
app.audio_enabled = True
|
|
|
|
await _switch_to(app, pilot, "somafm")
|
|
|
|
assert app.provider.id == "somafm"
|
|
assert len(app.stations) > 1
|
|
assert stub.played == [], "multi-station catalogues wait for enter"
|