asot_player/tests/test_asot.py

"""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)