from __future__ import annotations
import html
import re
from datetime import datetime, timedelta, timezone
import pytest
from somafm_tui import config
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(len(config.VISUALIZERS)):
await pilot.press("v")
await pilot.pause()
modes.add(app.visualizer)
assert modes == set(config.VISUALIZERS)
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_bandcamp_link_survives_history_refresh(monkeypatch) -> 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
url = "https://jensbuchert.bandcamp.com/track/rauschebart"
old_song = Song(
title="Rauschebart",
artist="Jens Buchert",
album="Gravity",
bandcamp_url=url,
)
station.now_playing = old_song
station.history = [old_song]
async def fake_fetch_history(client, station_id):
assert station_id == station.id
return [
Song(
title="Rauschebart",
artist="Jens Buchert",
album="Gravity",
)
]
monkeypatch.setattr("somafm_tui.app.api.fetch_history", fake_fetch_history)
worker = app._fetch_history(station.id)
await worker.wait()
await pilot.pause()
assert station.now_playing is not None
assert station.now_playing is not old_song
assert station.now_playing.bandcamp_url == url
assert "Bandcamp" in rendered_text(app)
@pytest.mark.asyncio
async def test_cached_bandcamp_link_is_reapplied_to_new_song() -> 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
url = "https://jensbuchert.bandcamp.com/track/rauschebart"
song = Song(title="Rauschebart", artist="Jens Buchert", album="Gravity")
app._bandcamp_urls[f"{station.id}|{song.key()}"] = url
station.now_playing = song
app._update_bandcamp()
await pilot.pause()
assert song.bandcamp_url == url
assert "Bandcamp" 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