asot_player/tests/test_asot.py

"""Unit tests for asot_player (stdlib unittest only)."""

from __future__ import annotations

import contextlib
import io
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
import asot_player.ui as ui_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


# --------------------------------------------------------------------------
# mpv socket death: a raw BrokenPipeError from sendall used to escape as a
# plain OSError and crash the player mid-playback on a remote machine.  It
# must surface as EngineError and the player must recover, not die.
# --------------------------------------------------------------------------

def _dead_mpv_engine(poll_rc=None):
    """MpEngine whose mpv has died: alive() passes, socket writes fail.
    poll_rc=None models the race that matters: poll() still reports the
    process alive while its IPC server has already closed the socket."""
    eng = MpEngine.__new__(MpEngine)        # skip __init__: no mpv spawn
    eng.mpv_bin = "mpv"
    eng.mpv_log = "/tmp/.asot-ipc-99.log"
    eng.proc = Mock()
    eng.proc.poll.return_value = poll_rc
    eng._cid = 0
    eng._prop_names = {}
    eng._adv = None
    eng._sock = Mock()
    eng._sock.sendall.side_effect = BrokenPipeError(32, "Broken pipe")
    return eng


class MpSocketDeathTests(unittest.TestCase):
    def test_cmd_surfaces_engine_error(self):
        eng = _dead_mpv_engine()
        with self.assertRaises(engine_mod.EngineError) as cm:
            eng._cmd(["get_property", "pause"])
        msg = str(cm.exception)
        self.assertIn("socket closed", msg)
        self.assertIn("asot-ipc-99.log", msg)      # and where to look

    def test_dead_process_is_reported_without_socket_write(self):
        eng = _dead_mpv_engine(poll_rc=1)
        with self.assertRaises(engine_mod.EngineError) as cm:
            eng._cmd(["get_property", "pause"])
        self.assertIn("mpv process is dead", str(cm.exception))

    def test_status_surfaces_engine_error(self):
        # alive() passes (poll → None), the write itself fails — the exact
        # shape of the reported crash traceback
        eng = _dead_mpv_engine()
        with self.assertRaises(engine_mod.EngineError):
            eng.status()


class PlayerRecoverTests(unittest.TestCase):
    def setUp(self):
        self._fetch = main_mod.fetch_meta
        main_mod.fetch_meta = lambda url: Meta()   # no network in tests
        self.eps = fake_eps()

    def tearDown(self):
        main_mod.fetch_meta = self._fetch

    def _player(self, ep, position, length):
        engine = Mock()
        engine.alive.return_value = False          # _load: restore saved vol
        p = Player(self.eps, engine, Mock(), State())
        p.ep = ep
        p.state.ep, p.state.url = ep.number, ep.url
        p.state.position, p.state.length = position, length
        return p

    def test_recover_restarts_same_episode_at_saved_position(self):
        p = self._player(self.eps[1], 3_000.0, 5_400.0)   # mid-episode
        self.assertTrue(p.recover())
        self.assertEqual(p.ep.number, 1300)               # same episode
        self.assertEqual(p.engine.play.call_args.kwargs["seek_to"], 3_000.0)

    def test_recover_advances_only_at_true_end(self):
        p = self._player(self.eps[1], 5_380.0, 5_400.0)   # within 30 s of end
        self.assertTrue(p.recover())
        self.assertEqual(p.ep.number, 1301)               # advanced
        self.assertEqual(p.engine.play.call_args.kwargs["seek_to"], 0.0)

    def test_recover_reports_failure_when_restart_fails(self):
        p = self._player(self.eps[1], 1_000.0, 5_400.0)
        p.engine.play.side_effect = engine_mod.EngineError("mpv gone")
        self.assertFalse(p.recover())

    def test_recover_without_episode_is_noop(self):
        p = Player(self.eps, Mock(), Mock(), State())
        self.assertFalse(p.recover())

    def test_recover_preserves_paused_state(self):
        # regression: a player left paused used to resume *playing* after a
        # stream drop recovered — the restart must keep it paused
        p = self._player(self.eps[1], 3_000.0, 5_400.0)
        p.state.paused = True
        self.assertTrue(p.recover())
        self.assertTrue(p.state.paused)
        p.engine.set_pause.assert_called_once_with(True)


# --------------------------------------------------------------------------
# Runtime limit: a session left running (e.g. paused) for hours must not end
# silently with exit code 0 (the old 50 000-tick cap did exactly that, at
# ~0.37 s per tick ≈ 5 h)
# --------------------------------------------------------------------------

class RuntimeLimitTests(unittest.TestCase):
    def setUp(self):
        self._fetch = main_mod.fetch_meta
        main_mod.fetch_meta = lambda url: Meta()   # no network in tests
        self.eps = fake_eps()

    def tearDown(self):
        main_mod.fetch_meta = self._fetch

    def _player(self):
        engine = FakeEngine(self.eps[1].url, length=7_200.0, speed=0.0)
        p = Player(self.eps, engine, UI(color=False, width=72, bar_width=40),
                   State())
        p.ep = self.eps[1]
        p.state.ep, p.state.url = self.eps[1].number, self.eps[1].url
        p.state.paused = True
        return p

    def test_wallclock_limit_reports_and_exits_nonzero(self):
        p = self._player()
        limit = main_mod.MAX_RUNTIME_SECONDS
        main_mod.MAX_RUNTIME_SECONDS = 0.2
        try:
            buf = io.StringIO()
            with contextlib.redirect_stdout(buf):
                rc = main_mod.run_player(p, None)
        finally:
            main_mod.MAX_RUNTIME_SECONDS = limit
        self.assertEqual(rc, 1)
        self.assertIn("runtime limit", buf.getvalue())

    def test_normal_finish_exits_zero_without_limit_message(self):
        p = self._player()
        p.finished = True                      # as if the user quit
        buf = io.StringIO()
        with contextlib.redirect_stdout(buf):
            rc = main_mod.run_player(p, None)
        self.assertEqual(rc, 0)
        self.assertNotIn("limit", buf.getvalue())

    def test_main_returns_rc_when_engine_fails(self):
        # rc=2 is set on EngineError — main() must return it, not 0
        eps = fake_eps()
        engine = Mock()
        engine.alive.return_value = False
        engine.play.side_effect = engine_mod.EngineError("mpv gone")
        with tempfile.TemporaryDirectory() as tmp:
            old_argv = list(sys.argv)
            sys.argv = ["asot", "--fake", "--fresh", "--width", "60",
                        "--state-dir", tmp]
            _fetch, _fake = main_mod.fetch_episodes, main_mod.FakeEngine
            main_mod.fetch_episodes = lambda: eps
            main_mod.FakeEngine = lambda *a, **k: engine
            try:
                buf = io.StringIO()
                with contextlib.redirect_stdout(buf), \
                     contextlib.redirect_stderr(io.StringIO()), \
                     unittest.mock.patch("sys.stdin", io.StringIO()):
                    rc = main_mod.main()
            finally:
                sys.argv = old_argv
                main_mod.fetch_episodes = _fetch
                main_mod.FakeEngine = _fake
        self.assertEqual(rc, 2)


# --------------------------------------------------------------------------
# Stream closed while paused: keep it paused — do NOT restart mpv until the
# user presses play again (the old loop treated it as a drop and un-paused
# the player with an auto-restart)
# --------------------------------------------------------------------------

class PausedStreamClosedTests(unittest.TestCase):
    def setUp(self):
        self._fetch = main_mod.fetch_meta
        main_mod.fetch_meta = lambda url: Meta()   # no network in tests
        self.eps = fake_eps()

    def tearDown(self):
        main_mod.fetch_meta = self._fetch

    def _dead_paused_player(self):
        engine = Mock()
        engine.status.side_effect = engine_mod.EngineError("mpv is dead")
        p = Player(self.eps, engine, UI(color=False, width=72, bar_width=40),
                   State())
        p.ep = self.eps[1]
        p.state.ep, p.state.url = self.eps[1].number, self.eps[1].url
        p.state.position, p.state.length = 3_000.0, 5_400.0
        p.state.paused = True
        return p

    def test_paused_drop_stays_paused_without_restart(self):
        p = self._dead_paused_player()
        limit = main_mod.MAX_RUNTIME_SECONDS
        main_mod.MAX_RUNTIME_SECONDS = 0.05
        try:
            buf = io.StringIO()
            with contextlib.redirect_stdout(buf):
                main_mod.run_player(p, None)
        finally:
            main_mod.MAX_RUNTIME_SECONDS = limit
        p.engine.play.assert_not_called()          # no auto-restart
        out = buf.getvalue()
        self.assertEqual(out.count("press play"), 1)   # reported once
        self.assertNotIn("recovery attempt", out)      # budget not burned
        self.assertIn("PAUSED", out)                   # bar keeps showing it

    def test_play_key_restarts_paused_engine_playing(self):
        p = self._dead_paused_player()
        p.handle_key(" ")                          # user pressed play
        self.assertTrue(p.engine.play.called)
        self.assertFalse(p.state.paused)           # resumes playback

    def test_non_play_keys_dropped_while_paused_engine_dead(self):
        p = self._dead_paused_player()
        p.handle_key("+")
        p.handle_key("up")
        p.engine.play.assert_not_called()
        self.assertTrue(p.state.paused)

    def test_quit_key_works_while_engine_dead(self):
        p = self._dead_paused_player()
        p.handle_key("q")
        self.assertTrue(p.finished)
        p.engine.play.assert_not_called()

    def test_playing_drop_still_auto_recovers(self):
        p = self._dead_paused_player()
        p.state.paused = False                     # was playing
        p.handle_key("+")
        self.assertTrue(p.engine.play.called)


# --------------------------------------------------------------------------
# Bottom anchoring: the block erases itself cleanly (never two bars), reports
# print on the line above it, and exit parks the cursor on that line
# --------------------------------------------------------------------------

def _tty_ui(rows: int = 24):
    """A tty UI; terminal size is pinned by the test's patched size."""
    u = UI(color=False, width=72, bar_width=40)
    u.tty = True
    u.start_line = u.block_start()
    return u


class BottomAnchorTests(unittest.TestCase):
    def setUp(self):
        self._size = ui_mod.shutil.get_terminal_size
        ui_mod.shutil.get_terminal_size = lambda *a: os.terminal_size((80, 24))
        self.ui = _tty_ui()

    def tearDown(self):
        ui_mod.shutil.get_terminal_size = self._size

    def render(self):
        return self.ui.render(1301, "01 Jan 2026", "PLAYING", "ok",
                              10.0, 100.0, 0.8, False)

    def test_render_anchors_and_clears_below(self):
        out = self.render()
        # 3 lines at rows-2, clearing everything below so an older frame or a
        # printed traceback cannot survive as a second bar
        self.assertTrue(out.startswith("\x1b[22;1H\x1b[J"), repr(out[:14]))
        self.assertEqual(self.ui.start_line, 22)

    def test_block_start_tracks_terminal_height(self):
        self.assertEqual(self.ui.block_start(), 22)
        ui_mod.shutil.get_terminal_size = lambda *a: os.terminal_size((80, 40))
        self.assertEqual(self.ui.block_start(), 38)

    def test_clear_block_parks_cursor_above_the_block(self):
        self.render()
        self.assertEqual(self.ui.clear_block(), "\x1b[0m\x1b[21;1H\x1b[J")

    def test_finish_clears_only_the_block(self):
        self.render()
        fin = self.ui.finish()
        self.assertEqual(fin, "\x1b[0m\x1b[21;1H\x1b[J")
        self.assertNotIn("\x1b[H\x1b[J", fin)          # no full-screen clear

    def test_non_tty_ui_writes_no_escapes(self):
        u = UI(color=False, width=72, bar_width=40)
        u.tty = False
        self.assertEqual(u.clear_block(), "")
        self.assertEqual(u.finish(), "")


class PrintAboveTests(unittest.TestCase):
    def test_plain_mode_writes_only_the_text(self):
        u = UI(color=False, width=72, bar_width=40)     # non-tty
        buf = io.StringIO()
        real = sys.stdout
        sys.stdout = buf
        try:
            main_mod._print_above(u, "boom\n")
        finally:
            sys.stdout = real
        self.assertEqual(buf.getvalue(), "boom\n")

    def test_tty_mode_clears_the_block_before_the_text(self):
        u = _tty_ui()
        u.render(1301, "01 Jan 2026", "PLAYING", "ok", 10.0, 100.0, 0.8, False)
        buf = io.StringIO()
        real = sys.stdout
        sys.stdout = buf
        try:
            main_mod._print_above(u, "boom")
        finally:
            sys.stdout = real
        self.assertEqual(buf.getvalue(), "\x1b[0m\x1b[21;1H\x1b[Jboom\n")


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