diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..b72666e --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import os + +os.environ["SOMAFM_IMAGES"] = "0" + +import pytest + +from somafm_tui import api, config + + +@pytest.fixture(autouse=True) +def isolated_state(tmp_path, monkeypatch): + """Keep tests out of the real ~/.config and ~/.cache.""" + config_dir = tmp_path / "config" + cache_dir = tmp_path / "cache" + songs_dir = cache_dir / "songs" + runtime_dir = tmp_path / "run" + + for attr, value in ( + ("CONFIG_DIR", config_dir), + ("CACHE_DIR", cache_dir), + ("SONGS_CACHE_DIR", songs_dir), + ("RUNTIME_DIR", runtime_dir), + ("CONFIG_PATH", config_dir / "config.json"), + ("CHANNELS_CACHE", cache_dir / "channels.json"), + ("PLAYLIST_CACHE", cache_dir / "playlists.json"), + ): + monkeypatch.setattr(config, attr, value) + + def ensure_dirs() -> None: + for path in (config_dir, cache_dir, songs_dir, runtime_dir): + path.mkdir(parents=True, exist_ok=True) + + monkeypatch.setattr(config, "ensure_dirs", ensure_dirs) + monkeypatch.setattr(api, "_PLAYLIST_CACHE", None) + ensure_dirs() + yield diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..d8c5d79 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import httpx +import pytest + +from somafm_tui import api, config + +CHANNELS_PAYLOAD = { + "channels": [ + { + "id": "groovesalad", + "title": "Groove Salad", + "description": "A nicely chilled plate.", + "dj": "Rusty Hodge", + "genre": "ambient|electronic", + "image": "https://api.somafm.com/logos/120/groovesalad120.png", + "listeners": "1,873", + "lastPlaying": "Jens Buchert - Rauschebart", + "playlists": [ + {"url": "https://api.somafm.com/groovesalad32.pls", "format": "aacp", "quality": "low"}, + {"url": "https://api.somafm.com/groovesalad256.pls", "format": "mp3", "quality": "highest"}, + {"url": "https://api.somafm.com/groovesalad130.pls", "format": "aac", "quality": "highest"}, + ], + }, + { + "id": "bad", + "title": "No playlists", + "genre": "test", + }, + {"title": "missing id"}, + ] +} + + +def test_parse_channels() -> None: + stations = api.parse_channels(CHANNELS_PAYLOAD) + assert len(stations) == 2 + station = stations[0] + assert station.id == "groovesalad" + assert station.title == "Groove Salad" + assert station.listeners == 1873 + assert station.genre_label == "ambient · electronic" + assert station.favorite is False + assert station.now_playing is not None + assert station.now_playing.artist == "Jens Buchert" + assert station.now_playing.title == "Rauschebart" + assert station.format_label == "mp3 · 256k" + assert next(pl["url"] for pl in station.sorted_playlists()).endswith("256.pls") + assert stations[1].id == "bad" + + +def test_parse_last_playing() -> None: + song = api.parse_last_playing("Jens Buchert - Rauschebart") + assert song is not None and song.artist == "Jens Buchert" and song.title == "Rauschebart" + en_dash = api.parse_last_playing("Tosca – Amber November") + assert en_dash is not None and en_dash.artist == "Tosca" + plain = api.parse_last_playing("Just A Title") + assert plain is not None and plain.title == "Just A Title" and plain.artist == "" + assert api.parse_last_playing("") is None + assert api.parse_last_playing(None) is None + + +def test_parse_pls() -> None: + text = """[playlist] +NumberOfEntries=2 +File1=https://ice2.somafm.com/groovesalad-256-mp3 +File2=https://ice6.somafm.com/groovesalad-256-mp3 +Version=2 +""" + assert api.parse_pls(text) == [ + "https://ice2.somafm.com/groovesalad-256-mp3", + "https://ice6.somafm.com/groovesalad-256-mp3", + ] + assert api.parse_pls("https://ice1.somafm.com/x-128-mp3\n# comment\n") == [ + "https://ice1.somafm.com/x-128-mp3" + ] + assert api.is_playlist("https://x/y.pls?foo=1") is True + assert api.is_direct_stream("https://ice1.somafm.com/x-128-mp3") is True + assert api.is_direct_stream("https://x/y.m3u") is False + + +def test_parse_songs_and_dates() -> None: + payload = { + "id": "groovesalad", + "songs": [ + { + "title": "Rauschebart", + "artist": "Jens Buchert", + "album": "Gravity", + "albumArt": "https://example.com/art.jpg", + "date": "1793312765", + }, + { + "title": "Room 505", + "artist": "Alex Cortiz", + "album": "The Chill Out Room", + "albumArt": "", + "date": 1793312438, + }, + ], + } + songs = api.parse_songs(payload) + assert len(songs) == 2 + assert songs[0].title == "Rauschebart" + assert songs[0].art_url == "https://example.com/art.jpg" + assert songs[0].played_at is not None and songs[0].played_at.year == 2026 + # newest first + assert songs[0].played_at > songs[1].played_at # type: ignore[operator] + assert api.parse_songs({"songs": [{"title": ""}]}) == [] + + +def test_history_cache_roundtrip() -> None: + path = api._songs_cache_path("groovesalad") + config.write_json( + path, + { + "fetched": 1, + "songs": [ + { + "title": "Eva", + "artist": "Heights Of Abraham", + "album": "Pork", + "albumArt": "", + "date": "2026-10-29T22:00:00+00:00", + } + ], + }, + ) + songs = api.load_cached_songs("groovesalad") + assert len(songs) == 1 + assert songs[0].artist == "Heights Of Abraham" + assert api.load_cached_songs("nope") == [] + + +@pytest.mark.asyncio +async def test_stream_candidates_with_mock_transport() -> None: + pls = "[playlist]\nFile1=https://ice2.somafm.com/x-256-mp3\nFile2=https://ice6.somafm.com/x-256-mp3\n" + requests: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(str(request.url)) + return httpx.Response(200, text=pls) + + station = api.parse_channels(CHANNELS_PAYLOAD)[0] + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + urls = await api.stream_candidates(client, station) + + assert urls[:2] == [ + "https://ice2.somafm.com/x-256-mp3", + "https://ice6.somafm.com/x-256-mp3", + ] + assert len(requests) == 3 # three playlist URLs, cached by URL + # second run hits the playlist cache + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + await api.stream_candidates(client, station) + assert len(requests) == 3 + + +@pytest.mark.asyncio +async def test_fetch_channels_and_history_cache() -> None: + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/channels.json": + return httpx.Response(200, json=CHANNELS_PAYLOAD) + if request.url.path.endswith(".json"): + return httpx.Response( + 200, + json={ + "songs": [ + {"title": "T", "artist": "A", "album": "B", "date": "1793312765"} + ] + }, + ) + return httpx.Response(404) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + stations = await api.fetch_channels(client) + assert len(stations) == 2 + assert config.CHANNELS_CACHE.exists() + assert len(api.load_cached_channels()) == 2 + + songs = await api.fetch_history(client, "groovesalad") + assert len(songs) == 1 + assert api.load_cached_songs("groovesalad")[0].title == "T" + + # a failing request still returns the cached history + def failing(request: httpx.Request) -> httpx.Response: + return httpx.Response(500) + + async with httpx.AsyncClient(transport=httpx.MockTransport(failing)) as client: + songs = await api.fetch_history(client, "groovesalad") + assert songs[0].title == "T" + + +def test_config_roundtrip() -> None: + config.Config(favorites=["a"], volume=42, visualizer="wave", autoplay=True).save() + loaded = config.Config.load() + assert loaded.favorites == ["a"] + assert loaded.volume == 42 + assert loaded.visualizer == "wave" + assert loaded.autoplay is True + assert loaded.is_favorite("a") is True + assert loaded.toggle_favorite("a") is False + assert loaded.toggle_favorite("b") is True + assert loaded.favorites == ["b"] + + +def test_channel_cache_missing() -> None: + assert api.load_cached_channels() == [] diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..4fd9955 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,135 @@ +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 diff --git a/tests/test_art.py b/tests/test_art.py new file mode 100644 index 0000000..5653069 --- /dev/null +++ b/tests/test_art.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import pytest +from PIL import Image + +from somafm_tui import artwork +from somafm_tui.app import SomaFMApp +from somafm_tui.widgets.art import ArtPane, ITerm2Image, KittyImage +from somafm_tui.widgets.stage import StagePanel + + +async def _no_fetch(client, url: str) -> None: + return None + + +@pytest.mark.asyncio +async def test_art_pane_switches_between_image_and_mosaic(monkeypatch) -> None: + monkeypatch.setattr(artwork, "PROTOCOL", "tgp") + monkeypatch.setattr( + artwork, + "fit_image", + lambda data, width, height: Image.new( + "RGBA", (width * 10, height * 20), (10, 20, 30, 255) + ), + ) + monkeypatch.setattr(artwork, "load_cached", lambda url: None) + monkeypatch.setattr(artwork, "fetch", _no_fetch) + + app = SomaFMApp(autoplay=False, offline=True, audio=False) + async with app.run_test(size=(110, 40)) as pilot: + await pilot.pause() + pane = app.query_one(ArtPane) + panel = app.query_one(StagePanel) + + app.artwork_bytes = b"fake-image-bytes" + panel.refresh_art() + await pilot.pause() + assert isinstance(pane._image_widget, KittyImage) + + app.artwork_bytes = None + panel.refresh_art() + await pilot.pause() + assert pane._image_widget is None + # The mosaic is always painted by ArtPane.render(), even under images. + assert pane.render() is not None + + +def test_iterm2_sequence_uses_cell_units() -> None: + widget = ITerm2Image(Image.new("RGBA", (32, 32), (10, 20, 30, 255))) + sequence = widget._sequence(13, 8) + assert sequence.startswith( + "\x1b]1337;File=inline=1;width=13;height=8;" + ) + + +def test_kitty_direct_placement_sequences() -> None: + widget = KittyImage(Image.new("RGBA", (32, 32), (10, 20, 30, 255))) + commands = widget._sequences(13, 8) + joined = "".join(commands) + assert "a=t" in joined + assert "a=p" in joined + assert "c=13,r=8" in joined + assert "U=1" not in joined # no Unicode placeholders + assert widget._sequences(13, 8) == [] + widget.cleanup() diff --git a/tests/test_artwork.py b/tests/test_artwork.py new file mode 100644 index 0000000..b29b523 --- /dev/null +++ b/tests/test_artwork.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from io import BytesIO + +import httpx +import pytest +from PIL import Image + +from somafm_tui import artwork + + +def png_bytes(size=(400, 200)) -> bytes: + buffer = BytesIO() + Image.new("RGBA", size, (20, 180, 220, 180)).save(buffer, format="PNG") + return buffer.getvalue() + + +def test_protocol_detection() -> None: + assert artwork.detect_protocol({"KITTY_WINDOW_ID": "1"}) == "tgp" + assert artwork.detect_protocol({"TERM": "xterm-kitty"}) == "tgp" + assert artwork.detect_protocol({"TERM_PROGRAM": "WezTerm"}) == "iterm2" + assert artwork.detect_protocol({"WEZTERM_EXECUTABLE": "/bin/wezterm"}) == "iterm2" + assert artwork.detect_protocol({"TERM_PROGRAM": "iTerm.app"}) == "iterm2" + assert artwork.detect_protocol({"ITERM_SESSION_ID": "x"}) == "iterm2" + assert artwork.detect_protocol({"TERM_PROGRAM": "Apple_Terminal"}) is None + assert artwork.detect_protocol({"TMUX": "/tmp/tmux", "KITTY_WINDOW_ID": "1"}) == "tgp" + assert artwork.detect_protocol({"TMUX": "/tmp/tmux", "TERM_PROGRAM": "WezTerm"}) is None + assert artwork.detect_protocol({"ZELLIJ": "1", "TERM_PROGRAM": "WezTerm"}) is None + assert artwork.detect_protocol({"KITTY_PID": "1"}) == "tgp" + assert artwork.detect_protocol({"NO_COLOR": "", "KITTY_WINDOW_ID": "1"}) is None + assert artwork.detect_protocol({"SOMAFM_IMAGES": "0", "KITTY_WINDOW_ID": "1"}) is None + assert artwork.detect_protocol({"SOMAFM_IMAGES": "tgp"}) == "tgp" + assert artwork.detect_protocol({"SOMAFM_IMAGES": "iterm2"}) == "iterm2" + + +def test_fit_image_crops_to_the_cell_box() -> None: + fitted = artwork.fit_image(png_bytes(), 13, 8) + assert fitted is not None + cell_width, cell_height = artwork.cell_size() + expected = (13 * cell_width) / (8 * cell_height) + assert fitted.width / fitted.height == pytest.approx(expected, rel=0.01) + assert artwork.fit_image(b"not an image", 13, 8) is None + + +def test_image_cache_roundtrip() -> None: + data = png_bytes() + url = "https://example.invalid/art.png" + artwork.save_cached(url, data) + assert artwork.load_cached(url) == data + assert artwork.load_cached("https://example.invalid/other.png") is None + + +@pytest.mark.asyncio +async def test_fetch_validates_and_caches() -> None: + data = png_bytes((120, 120)) + url = "https://example.invalid/logo.png" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=data, headers={"content-type": "image/png"}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + fetched = await artwork.fetch(client, url) + assert fetched == data + assert artwork.load_cached(url) == data diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..170c0ec --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +from somafm_tui.__main__ import _build_parser, _list_stations + + +def test_parser_defaults_and_flags() -> None: + parser = _build_parser() + args = parser.parse_args([]) + assert args.autoplay is None + assert args.offline is False + assert args.no_audio is False + + assert args.spectrum_source == "auto" + + args = parser.parse_args( + ["--station", "defcon", "--play", "--no-mpris", "--spectrum-source", "monitor"] + ) + assert args.station == "defcon" + assert args.autoplay is True + assert args.no_mpris is True + assert args.spectrum_source == "monitor" + + +def test_list_stations_offline(capsys) -> None: + assert _list_stations(offline=True) == 0 + output = capsys.readouterr().out + assert "groovesalad" in output + assert "Groove Salad" in output diff --git a/tests/test_mpris.py b/tests/test_mpris.py new file mode 100644 index 0000000..3213114 --- /dev/null +++ b/tests/test_mpris.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import queue + +import pytest + +from somafm_tui import mpris + + +def test_build_metadata() -> None: + metadata = mpris.build_metadata( + station="Groove Salad", + title="Rauschebart", + artist="Jens Buchert", + album="Gravity", + art_url="https://example.com/art.jpg", + stream_url="https://ice2.somafm.com/groovesalad-256-mp3", + track_key="groovesalad|Jens Buchert|Rauschebart|Gravity", + ) + assert metadata["mpris:trackid"].value.startswith("/org/somafm/track/") + assert metadata["xesam:title"].value == "Rauschebart" + assert metadata["xesam:artist"].value == ["Jens Buchert"] + assert metadata["xesam:album"].value == "Gravity" + assert metadata["mpris:artUrl"].value == "https://example.com/art.jpg" + assert metadata["xesam:url"].value.startswith("https://ice2.") + assert metadata["xesam:comment"].value == ["SomaFM · Groove Salad"] + + minimal = mpris.build_metadata(station="X", title="", artist="") + assert minimal["xesam:artist"].value == ["SomaFM"] + assert "xesam:album" not in minimal + assert "mpris:artUrl" not in minimal + + +def test_disabled_service_is_inert() -> None: + service = mpris.MprisService(queue.Queue(), enabled=False) + assert service.status_label == "off" + service.start() + service.set_track(station="X", title="T", artist="A") + service.set_status("Playing") + service.set_volume(0.5) + service.stop() + assert service.available is False + + +@pytest.mark.skipif(not mpris.DBUS_AVAILABLE, reason="dbus-next not installed") +def test_interfaces_introspect_and_queue_actions() -> None: + actions: queue.Queue[str] = queue.Queue() + service = mpris.MprisService(actions, enabled=False) + + player = mpris._PlayerInterface(service) + interface = player.introspect() + properties = {prop.name: prop for prop in interface.properties} + assert properties["Metadata"].signature == "a{sv}" + assert properties["PlaybackStatus"].signature == "s" + assert properties["Volume"].signature == "d" + assert properties["Volume"].access == mpris.PropertyAccess.READWRITE + assert properties["Position"].signature == "x" + + methods = {method.name: method.in_signature for method in interface.methods} + assert methods["Play"] == "" + assert methods["Seek"] == "x" + assert methods["SetPosition"] == "ox" + assert methods["OpenUri"] == "s" + + player.Volume = 0.33 + assert actions.get_nowait() == "volume:33" + player.Play() + player.Pause() + player.PlayPause() + player.Stop() + player.Next() + player.Previous() + assert [actions.get_nowait() for _ in range(6)] == [ + "play", + "pause", + "toggle", + "stop", + "next", + "prev", + ] + + root = mpris._RootInterface(service) + root_interface = root.introspect() + root_props = {prop.name: prop.signature for prop in root_interface.properties} + assert root_props["Identity"] == "s" + assert root_props["SupportedUriSchemes"] == "as" + assert "Quit" in {method.name for method in root_interface.methods} + + +@pytest.mark.skipif(not mpris.DBUS_AVAILABLE, reason="dbus-next not installed") +def test_service_stores_state_without_bus() -> None: + service = mpris.MprisService(queue.Queue(), enabled=False) + service.set_track( + station="Groove Salad", + title="Rauschebart", + artist="Jens Buchert", + album="Gravity", + art_url="https://example.com/art.jpg", + stream_url="https://ice2.somafm.com/x-256-mp3", + ) + service.set_status("Playing") + service.set_volume(0.8) + assert service._metadata["xesam:title"].value == "Rauschebart" + assert service._status == "Playing" + assert service._volume == 0.8 diff --git a/tests/test_player.py b/tests/test_player.py new file mode 100644 index 0000000..369d54d --- /dev/null +++ b/tests/test_player.py @@ -0,0 +1,168 @@ +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() diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py new file mode 100644 index 0000000..9f68722 --- /dev/null +++ b/tests/test_spectrum.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import time +from pathlib import Path + +import numpy as np + +from somafm_tui import spectrum as spectrum_module +from somafm_tui.spectrum import Frame, SpectrumEngine + +PCM_SCRIPT = """#!/usr/bin/env python3 +import math, struct, sys, time +sr = 44100 +samples = [int(0.6 * 32767 * math.sin(2 * math.pi * 1000 * i / sr)) for i in range(4096)] +chunk = struct.pack("<%dh" % len(samples), *samples) +out = sys.stdout.buffer +try: + while True: + out.write(chunk) + out.flush() + time.sleep(0.08) +except (BrokenPipeError, KeyboardInterrupt): + pass +""" + + +def write_pcm_script(path: Path) -> None: + path.write_text(PCM_SCRIPT) + path.chmod(0o755) + + +def wait_for_spectrum(engine: SpectrumEngine, status: str, timeout: float = 6.0): + deadline = time.monotonic() + timeout + snap = engine.snapshot() + while time.monotonic() < deadline: + snap = engine.snapshot() + if snap.status == status and sum(snap.bands) > 1.0: + break + time.sleep(0.05) + return snap + + +def test_fft_detects_a_1khz_tone() -> None: + engine = SpectrumEngine(ffmpeg_path="") + sample_rate = 44100 + t = np.arange(8192) / sample_rate + tone = np.sin(2 * np.pi * 1000 * t) * 0.5 + engine.feed_pcm((tone * 32767).astype(" 0.8 + assert snap.peaks[top] >= snap.bands[top] - 1e-6 + assert any(abs(value) > 0.3 for value in snap.wave) + assert 0.0 < snap.level <= 1.0 + + +def test_frame_labels() -> None: + assert Frame((), (), (), 0.0, "monitor").label == "SYNC" + assert Frame((), (), (), 0.0, "live").label == "FFT" + assert Frame((), (), (), 0.0, "sim").label == "SIM" + assert Frame((), (), (), 0.0, "idle").label == "IDLE" + assert Frame((), (), (), 0.0, "off").label == "OFF" + assert Frame((), (), (), 0.0, "monitor").is_live is True + + +def test_simulation_fallback() -> None: + engine = SpectrumEngine(ffmpeg_path="", source_mode="sim") + engine.start() + try: + engine.set_source("https://example.invalid/demo") + deadline = time.monotonic() + 5 + snap = engine.snapshot() + while time.monotonic() < deadline: + snap = engine.snapshot() + if snap.status == "sim" and max(snap.bands) > 0.2: + break + time.sleep(0.05) + assert snap.status == "sim" + assert snap.label == "SIM" + assert max(snap.bands) > 0.2 + assert 0.0 <= snap.level <= 1.0 + finally: + engine.stop() + + +def test_monitor_capture_prefers_real_output(tmp_path, monkeypatch) -> None: + script = tmp_path / "parec" + write_pcm_script(script) + + def fake_which(name: str) -> str | None: + return str(script) if name == "parec" else None + + monkeypatch.setattr(spectrum_module.shutil, "which", fake_which) + engine = SpectrumEngine(source_mode="auto") + assert engine.monitor_available is True + + engine.start() + try: + engine.set_source("https://example.invalid/live") + snap = wait_for_spectrum(engine, "monitor") + assert snap.status == "monitor", snap + assert snap.label == "SYNC" + + top = max(range(len(snap.bands)), key=lambda index: snap.bands[index]) + assert 36 <= top <= 42 + assert snap.bands[top] > 0.5 + + engine.set_source(None) + time.sleep(0.3) + assert engine.snapshot().status == "idle" + finally: + engine.stop() + + +def test_stream_tap_pipeline(tmp_path) -> None: + script = tmp_path / "fake-ffmpeg" + write_pcm_script(script) + + engine = SpectrumEngine(ffmpeg_path=str(script), source_mode="stream") + engine.start() + try: + engine.set_source("https://example.invalid/stream") + snap = wait_for_spectrum(engine, "live") + assert snap.status == "live", snap + assert snap.label == "FFT" + top = max(range(len(snap.bands)), key=lambda index: snap.bands[index]) + assert 36 <= top <= 42 + assert snap.bands[top] > 0.5 + + engine.set_source(None) + time.sleep(0.3) + assert engine.snapshot().status == "idle" + finally: + engine.stop() diff --git a/tests/test_theme.py b/tests/test_theme.py new file mode 100644 index 0000000..282d57b --- /dev/null +++ b/tests/test_theme.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import re +from pathlib import Path + +import pytest + +from somafm_tui.app import SomaFMApp +from somafm_tui.widgets.header import HeaderBar +from somafm_tui.widgets.search import SearchInput +from somafm_tui.widgets.spectrum import SpectrumPanel +from somafm_tui.widgets.stage import StagePanel +from somafm_tui.widgets.station_list import StationList +from somafm_tui.widgets.status import StatusBar + +PACKAGE = Path(__file__).resolve().parents[1] / "somafm_tui" + + +def test_no_rgb_colors_in_source() -> None: + """The app must follow the terminal palette, not force its own RGB colours.""" + pattern = re.compile(r"#[0-9a-fA-F]{6}") + offenders = [ + str(path.relative_to(PACKAGE)) + for path in PACKAGE.rglob("*") + if path.suffix in {".py", ".tcss"} and pattern.search(path.read_text()) + ] + assert not offenders, offenders + + +@pytest.mark.asyncio +async def test_backgrounds_use_terminal_default_and_borders_are_ansi() -> None: + app = SomaFMApp(autoplay=False, offline=True, audio=False) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + + assert app.theme == "terminal" + assert app.ansi_color is True + assert app.native_ansi_color is True + assert app.screen.styles.background.ansi == -1 # SGR 49, terminal default + assert app.screen.styles.color.ansi == -1 + for widget_type in ( + HeaderBar, + SearchInput, + StationList, + StagePanel, + SpectrumPanel, + StatusBar, + ): + widget = app.query_one(widget_type) + assert widget.styles.background.ansi == -1, widget_type + assert widget.styles.color.ansi == -1, widget_type + + border = app.query_one(StationList).styles.border.top + assert border[1].ansi is not None # an ANSI palette index, not RGB + + await pilot.press("slash") + await pilot.pause() + search = app.query_one(SearchInput) + assert search.styles.background.ansi == -1 + assert search.styles.background_tint.a == 0 + assert search.styles.color.ansi == -1 # ansi_default + await pilot.press("escape") + await pilot.pause() + + await pilot.press("question_mark") + await pilot.pause() + assert app.screen.styles.background.ansi == -1 + assert app.screen.query_one("#help").styles.background.ansi == -1 + await pilot.press("x") + await pilot.pause() diff --git a/tests/test_timezone.py b/tests/test_timezone.py new file mode 100644 index 0000000..de73917 --- /dev/null +++ b/tests/test_timezone.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from somafm_tui.widgets import common + +BERLIN = Path("/usr/share/zoneinfo/Europe/Berlin") +UTC_FILE = Path("/usr/share/zoneinfo/UTC") + + +@pytest.fixture(autouse=True) +def clear_zone_cache(): + common.local_zone.cache_clear() + yield + common.local_zone.cache_clear() + + +def _copy_zone(tmp_path: Path, source: Path) -> Path: + target = tmp_path / "localtime" + target.write_bytes(source.read_bytes()) + return target + + +def test_build_time_tz_utc_does_not_hide_the_system_zone(tmp_path, monkeypatch) -> None: + if not BERLIN.exists(): + pytest.skip("system tzdata is not installed") + monkeypatch.setattr(common, "LOCALTIME_PATH", str(_copy_zone(tmp_path, BERLIN))) + monkeypatch.setenv("TZ", "UTC") + + zone = common.local_zone() + assert zone.utcoffset(datetime(2026, 1, 1)) == timedelta(hours=1) + + aware = datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc) + assert common.local_time(aware) == "13:00" + + +def test_explicit_non_utc_tz_is_respected(tmp_path, monkeypatch) -> None: + if not BERLIN.exists() or not UTC_FILE.exists(): + pytest.skip("system tzdata is not installed") + monkeypatch.setattr(common, "LOCALTIME_PATH", str(_copy_zone(tmp_path, UTC_FILE))) + monkeypatch.setenv("TZ", "Europe/Berlin") + + zone = common.local_zone() + assert zone.utcoffset(datetime(2026, 1, 1)) == timedelta(hours=1) + assert common.local_time(datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc)) == "13:00"