"""Unit tests for asot_player (stdlib unittest only).""" from __future__ import annotations import os import re import sys import re import tempfile import unittest from unittest.mock import Mock sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)) + "/..") import asot_player.main as main_mod import asot_player.engine as engine_mod from asot_player.feed import Episode from asot_player.id3 import Meta, parse_id3 from asot_player.engine import FakeEngine, MpEngine from asot_player.main import Player, resolve_start from asot_player.state import State from asot_player.ui import UI, visible_len FIXTURE = os.environ.get("ASOT_FIXTURE", "/tmp/ep1298_head.bin") # -------------------------------------------------------------------------- # id3 helpers (writer layout: junk 0x00 0x01 + BOM + UTF-16BE text; every # declared frame size is exactly one byte short of its content) # -------------------------------------------------------------------------- def _v23_size(n: int) -> bytes: out = bytearray(4) for i in range(4): out[i] = (n >> (21 - 7 * i)) & 0x7F return bytes(out) def _text_content(text: str) -> bytes: """Writer's canonical text-frame payload: junk 0x00 0x01, an FF FE BOM byte pair, then the text in UTF-16LE byte order (the real writer's layout).""" return b"\x00\x01\xff\xfe" + text.encode("utf-16-le") def _frame(fid: bytes, content: bytes) -> bytes: """One frame honouring the writer invariant: declared size = len - 1.""" return fid + _v23_size(len(content) - 1) + b"\x00" + content def build_tag(frames: list[bytes]) -> bytes: """A 10-byte-header ID3v2.3 tag (as emitted by the real writer: ID3 + ver + flags + padding byte + 4-byte size field, first frame at offset 10) with exactly these frames (junk gaps may be inserted as bytes between them).""" payload = b"".join(frames) return b"ID3\x03\x00\x00" + _v23_size(len(payload)) + payload class Id3SyntheticTests(unittest.TestCase): def test_synthetic_basic(self): frames = [ _frame(b"TIT2", _text_content("Main Title")), _frame(b"TPE2", _text_content("Some Artist")), _frame(b"TALB", _text_content("Some Album")), _frame(b"TYER", b"\x00\x00\x00" + b"1999"), _frame(b"TXXX", _text_content("COMMENT\x00micro")), ] m = parse_id3(build_tag(frames)) self.assertEqual(m.titles, ("Main Title",)) self.assertEqual(m.title, "Main Title") self.assertEqual(m.artist, "Some Artist") self.assertEqual(m.album, "Some Album") self.assertEqual(m.year, "1999") self.assertEqual(m.comment, "micro") def test_tyer_ascii(self): # TYER: NUL-padded plain ASCII; content is fsz+1 bytes. frames = [ b"TIT2" + _v23_size(0) + b"\x00" + _text_content("x"), _frame(b"TYER", b"\x00\x00\x00\x00" + b"2026"), ] m = parse_id3(build_tag(frames)) self.assertEqual(m.year, "2026") def test_txxx_with_repeated_bom(self): # Writer repeats the BOM inside TXXX values (see real episode data); # key and value are both junk+BOM+UTF-16LE, like every real frame. content = (b"\x00\x01\xff\xfe" + "DESC".encode("utf-16-le") + b"\x00" + b"\xff\xfe" + "microsite".encode("utf-16-le")) m = parse_id3(build_tag([_frame(b"TXXX", content)])) self.assertIn("microsite", m.comment) def test_comm_fallback_with_repeated_bom(self): # No TXXX: comment must come from COMM; its text follows a second BOM. content = (b"\x00\x01eng\xff\xfe\x00\x00\xff\xfe" + "MiroPPB.com".encode("utf-16-le")) m = parse_id3(build_tag([_frame(b"COMM", content)])) self.assertEqual(m.comment, "MiroPPB.com") def test_ghost_empty_frame(self): # junk+BOM only (declared size 3) -> decodes to nothing, not a title. frames = [ _frame(b"TIT2", b"\x00\x01\xff\xfe"), _frame(b"TIT2", _text_content("Real")), ] m = parse_id3(build_tag(frames)) self.assertEqual(m.titles, ("Real",), f"titles={m.titles!r}") def test_odd_utf16_body_does_not_crash(self): frames = [_frame(b"TIT2", b"\x00\x01\xff\xfe\x41")] # odd-length body m = parse_id3(build_tag(frames)) self.assertEqual(m.title, "") def test_no_header_or_bad_major(self): self.assertEqual(parse_id3(b""), Meta()) self.assertEqual(parse_id3(b"NOTAG").title, "") m = parse_id3(b"ID3\x02\x00\x00\x00\x00" + b"\x00" * 4) self.assertEqual(m.title, "") def test_frames_scanned_among_junk_gaps(self): frames = [ b"\x00" * 5, _frame(b"TPE2", _text_content("Gap Artist")), bytes([0xaa] * 37), _frame(b"TIT2", _text_content("Later Title")), ] m = parse_id3(build_tag(frames)) self.assertIn("Later Title", m.titles) self.assertEqual(m.artist, "Gap Artist") def test_display_fallbacks(self): m = Meta(titles=("A Song",), artist="The Band") self.assertEqual(m.display, "The Band - A Song") m = Meta(titles=("One", "Two"), artist="The Band", album="Records", year="2026") self.assertEqual(m.display, "The Band - Records (2026)") self.assertEqual(Meta(artist="Solo").display, "Solo") self.assertEqual(Meta().display, "") self.assertEqual(Meta(titles=("X", "X", ""), artist="A").display, "A - X") # no artist, several tracks -> album/year fallback self.assertEqual(Meta(titles=("A", "B"), album="Alb").display, "Alb") def test_real_head_fixture(self): """Parse the 1 MiB real fixture of ASOT_Ep_1298 (saved from miroppb).""" if not os.path.isfile(FIXTURE): self.skipTest("fixture not available in this environment") with open(FIXTURE, "rb") as f: m = parse_id3(f.read()) self.assertGreaterEqual(len(m.titles), 35) self.assertEqual(m.artist, "Armin van Buuren") self.assertEqual(m.album, "A State of Trance") self.assertEqual(m.year, "2026") self.assertIn("MiroPPB.com", m.comment or "") self.assertTrue("Episode 1298" in " ".join(m.titles)) # -------------------------------------------------------------------------- # resolve_start (main.py) # -------------------------------------------------------------------------- def fake_eps(): return [ Episode(number=1301, url="https://miroppb.com/b/rss/ASOT_Ep_1301.mp3"), Episode(number=1300, url="https://miroppb.com/b/rss/ASOT_Ep_1300.mp3"), Episode(number=1299, url="https://miroppb.com/b/rss/ASOT_Ep_1299.mp3"), Episode(number=1298, url="https://miroppb.com/b/rss/ASOT_Ep_1298.mp3"), ] class ResolveStartTests(unittest.TestCase): def setUp(self): self.eps = fake_eps() def test_explicit_episode_found(self): ep, resume, pos = resolve_start(self.eps, None, 1299) self.assertEqual(ep.number, 1299) self.assertFalse(resume) self.assertEqual(pos, 0.0) def test_explicit_episode_missing_constructed(self): ep, resume, pos = resolve_start(self.eps, None, 1999) self.assertEqual(ep.number, 1999) self.assertFalse(resume) self.assertIn("1999", ep.url) self.assertTrue(ep.url.endswith(".mp3"), f"bad URL: {ep.url!r}") def test_state_resumes_matching_episode(self): st = State() st.ep = 1298 st.url = self.eps[-1].url st.position = 28.0 st.length = 30.0 ep, resume, pos = resolve_start(self.eps, st, None) self.assertEqual(ep.number, 1298) self.assertTrue(resume) # clamped to length - 5 so we don't seek into the last seconds self.assertEqual(pos, 25.0) def test_state_short_length_starts_clean(self): # length <= 5 s (fake-engine size): marked as resuming but position # is reset to 0 so playback starts clean. st = State() st.ep = 1298 st.url = self.eps[-1].url st.position = 2.0 st.length = 3.0 ep, resume, pos = resolve_start(self.eps, st, None) self.assertEqual(ep.number, 1298) self.assertTrue(resume) self.assertEqual(pos, 0.0) def test_state_not_in_feed_falls_back(self): st = State() st.ep = 500 # not present (feed was rebuilt without it) ep, resume, _ = resolve_start(self.eps, st, None) self.assertEqual(ep.number, self.eps[0].number) self.assertFalse(resume) def test_position_zero_no_resume(self): st = State() st.ep = 1298 st.url = self.eps[-1].url st.position = 0.0 _, resume, _ = resolve_start(self.eps, st, None) self.assertFalse(resume) # -------------------------------------------------------------------------- # Player.advance (direction + edge behaviour) # -------------------------------------------------------------------------- class PlayerAdvanceTests(unittest.TestCase): def setUp(self): # patch must span the whole test body, not just setUp: _load() runs # inside test methods, so the module attribute has to stay mocked. self._orig_fetch_meta = main_mod.fetch_meta main_mod.fetch_meta = Mock(return_value=Meta()) self.episodes = [ Episode(number=1301, url="https://x/1301.mp3"), Episode(number=1300, url="https://x/1300.mp3"), Episode(number=1299, url="https://x/1299.mp3"), Episode(number=1298, url="https://x/1298.mp3"), ] self.p = Player(eps=self.episodes, engine=FakeEngine("u", 3.0), ui=UI(False, 56, 40), state=State()) def tearDown(self): main_mod.fetch_meta = self._orig_fetch_meta del self._orig_fetch_meta def load(self, number): for e in self.p.eps: if e.number == number: self.p._load(e, "test") break return self.p def test_next_and_prev_by_number(self): self.load(1299) self.assertTrue(self.p.advance(+1)) self.assertEqual(self.p.ep.number, 1300) self.assertTrue(self.p.advance(-1)) self.assertEqual(self.p.ep.number, 1299) self.assertTrue(self.p.advance(-1)) self.assertEqual(self.p.ep.number, 1298) self.assertTrue(self.p.advance(+1)) self.assertEqual(self.p.ep.number, 1299) def test_edge_end_of_feed(self): self.load(1301) self.assertFalse(self.p.advance(+1)) self.assertEqual(self.p.ep.number, 1301) self.load(1298) self.assertFalse(self.p.advance(-1)) self.assertEqual(self.p.ep.number, 1298) def test_msg_override_on_advance(self): self.load(1299) self.p.advance(+1) self.assertIn("ASOT 1300", self.p.msg_override) self.assertTrue("next" in self.p.msg_override.lower()) # -------------------------------------------------------------------------- # FakeEngine behaviour # -------------------------------------------------------------------------- class FakeEngineTests(unittest.TestCase): @staticmethod def make(speed=1.0, length=3.0): return FakeEngine("u", length=length, speed=speed) def test_position_advances_per_status(self): e = self.make(speed=2.5, length=10.0) e.play("u") self.assertEqual(e.status().position, 2.5) self.assertEqual(e.status().position, 5.0) def test_eof_reached_and_clamped(self): e = self.make(length=3.0) e.play("u") e.status() # 1 e.status() # 2 s = e.status() self.assertEqual(s.state, "eof") self.assertEqual(s.position, 3.0) def test_buffering_then_playing(self): e = self.make(length=10.0) e.play("u") self.assertIn(e.status().state, ("buffering",)) self.assertEqual(e.status().state, "playing") def test_seeks_clamp(self): # speed 0 -> status() does not advance the position, keeps asserts exact e = self.make(length=5.0, speed=0.0) e.play("u") e.seek(999.0) self.assertEqual(e.status().position, 5.0) e.seek(-10.0) self.assertEqual(e.status().position, 0.0) e.seek(2.0) self.assertEqual(e.status().position, 2.0) self.assertEqual(e.seek_count, 3) def test_volume_clamps(self): e = self.make() e.play("u") e.set_volume(1.4) self.assertEqual(e.status().volume, 1.0) e.set_volume(-0.2) self.assertEqual(e.status().volume, 0.0) e.set_volume(0.75) self.assertAlmostEqual(e.status().volume, 0.75) def test_seek_to_on_play(self): # speed 0 -> no implicit advance between play() and status() e = self.make(length=10.0, speed=0.0) e.play("u", seek_to=4.0) self.assertEqual(e.status().position, 4.0) e.play("u", seek_to=99.0) # clamped to length self.assertEqual(e.status().position, 10.0) def test_stop_resets_url_tracking(self): e = self.make() e.play("u") e.stop() self.assertEqual(e.state, "stop") self.assertIsNone(e.played_url) # -------------------------------------------------------------------------- # MpEngine derived state (no mpv process — pure logic on position progress) # -------------------------------------------------------------------------- class MpDerivedStateTests(unittest.TestCase): @staticmethod def eng(): e = MpEngine.__new__(MpEngine) # skip __init__: no mpv spawn needed e._adv = None return e def test_paused_wins(self): self.assertEqual(self.eng()._derived_state(True, 100.0, 7000.0), "paused") def test_near_end_is_eof(self): self.assertEqual(self.eng()._derived_state(False, 6999.0, 7000.0), "eof") def test_progress_is_playing(self): e = self.eng() real = engine_mod.time.time try: engine_mod.time.time = lambda: 1_000.0 e._derived_state(False, 3_000.0, 7_000.0) # baseline engine_mod.time.time = lambda: 1_030.0 self.assertEqual(e._derived_state(False, 3_030.0, 7_000.0), "playing") finally: engine_mod.time.time = real def test_long_stall_is_buffering_not_eof(self): # Regression: a stall used to become "eof" after 8s, which made a # resume on an unseekable HTTP stream auto-advance to the next # episode mid-playback. Stalls must stay "buffering" indefinitely. e = self.eng() real = engine_mod.time.time try: engine_mod.time.time = lambda: 1_000.0 e._derived_state(False, 3_000.0, 7_000.0) # baseline engine_mod.time.time = lambda: 1_200.0 # 200 s no progress self.assertEqual(e._derived_state(False, 3_000.0, 7_000.0), "buffering") finally: engine_mod.time.time = real # -------------------------------------------------------------------------- # Keyboard restore: Linux termios has no TCSU — using it silently skipped the # restore (AttributeError swallowed), leaving the shell in cbreak mode. # -------------------------------------------------------------------------- class KeyboardRestoreTests(unittest.TestCase): def test_stop_uses_posix_flush_constant(self): kb = main_mod.Keyboard.__new__(main_mod.Keyboard) kb.fd, kb.oldattr = 0, Mock() t = main_mod.termios calls = [] real = t.tcsetattr t.tcsetattr = lambda f, when, attr: calls.append(when) try: kb.stop() finally: t.tcsetattr = real self.assertEqual(calls, [t.TCSAFLUSH]) # -------------------------------------------------------------------------- # State: round-trip / corrupt file / env dir # -------------------------------------------------------------------------- class StateTests(unittest.TestCase): def setUp(self): self._dir = tempfile.TemporaryDirectory() self._saved = os.environ.pop("ASOT_STATE_DIR", None) os.environ["ASOT_STATE_DIR"] = self._dir.name def tearDown(self): del os.environ["ASOT_STATE_DIR"] if self._saved is not None: os.environ["ASOT_STATE_DIR"] = self._saved self._dir.cleanup() def test_roundtrip(self): st = State(ep=1298, url="https://x/1.mp3", position=21.5, length=100.0, paused=False, muted=True, volume=0.42) st.save() loaded = State.load() self.assertIsNotNone(loaded) self.assertEqual(loaded.ep, 1298) self.assertEqual(loaded.url, "https://x/1.mp3") self.assertAlmostEqual(loaded.position, 21.5) self.assertTrue(loaded.muted) self.assertAlmostEqual(loaded.volume, 0.42) self.assertTrue(loaded.updated_at) def test_load_corrupt_file_returns_none(self): State().save() State.state_file().write_text("not json at all {{{{", encoding="utf-8") self.assertIsNone(State.load()) def test_load_empty_file_returns_none(self): State().save() State.state_file().write_text("", encoding="utf-8") self.assertIsNone(State.load()) def test_load_missing_returns_none(self): self.assertIsNone(State.load()) def test_from_raw_defaults_and_clamps(self): st = State.from_raw({"ep": 42, "volume": 3.0, "position": 5, "length": "1.0"}) self.assertEqual(st.ep, 42) self.assertEqual(st.volume, 1.0) self.assertEqual(st.position, 5.0) self.assertEqual(st.length, 1.0) def test_clear(self): State().save() self.assertTrue(os.path.exists(State.state_file())) self.assertTrue(State.clear()) self.assertFalse(os.path.exists(State.state_file())) self.assertFalse(State.clear()) # -------------------------------------------------------------------------- # UI: every line is exactly `width` visible characters in both colour modes # -------------------------------------------------------------------------- def rendered_lines(ui, date, status, kind, position, length, volume): out = ui.render(1298, date, status, kind, position, length, volume, False) body = re.sub(r"^\x1b\[\d+;1H", "", out).rstrip("\n") # strip prefix + trailing LF return body.split("\n") class UITests(unittest.TestCase): @staticmethod def widths_and_colors(): for w in (52, 53, 78, 120): for color in (True, False): yield UI(color, w, 40), w def test_line_lengths_exact(self): for ui, w in self.widths_and_colors(): lines = [ui.title_line(1298, "PLAYING", "ok", "07 Feb 2026"), ui.title_line(1298, "Finished — end of feed at ASOT 1301", "info", ""), ui.title_line(1298, "", "ok", ""), ui.bar_line(None, None, 0.5, False), ui.bar_line(0.0, 0.0, 0.5, False), ui.bar_line(1234.5, 7200.0, 0.8, True), ui.hint_line()] for ln in lines: self.assertEqual(visible_len(ln), w, f"visible len != width: {ln!r}") def test_rendered_block_lengths_exact(self): for ui, w in self.widths_and_colors(): body = rendered_lines(ui, "07 Feb 2026", "Buffering", "buffer", 333.5, 7200.0, 0.85) for ln in body: self.assertEqual(visible_len(ln), w, f"visible len != width: {ln!r}") def test_plain_mode_no_escapes(self): ui = UI(False, 80, 40) out = ui.plain_render(9, "d", "ok", "ok", 1.0, 10.0, 0.5, False) self.assertNotIn("\x1b", out) lines = [l for l in out.splitlines() if l] self.assertEqual(len(lines), 3) self.assertTrue(all(len(l) == 80 for l in lines)) if __name__ == "__main__": unittest.main(verbosity=2)