lintunes/tests/test_round29.py

"""Round 29: casting to a Chromecast.

Covers the parts that are verifiable without a device on the network — the
format gate, byte-range arithmetic, and the local media server that hands the
Chromecast its files. The server tests drive a real socket on an ephemeral
port, so range requests, HEAD and token containment are exercised end to end.

Not covered here (hand-verified against a real Chromecast): the pychromecast
handshake and discovery, the receiver's own decode of any given file, seek
behavior over Wi-Fi, and the MainWindow menu/dialog wiring.
"""

import http.client
import socket
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest
from PyQt6.QtCore import QObject, pyqtSignal

from lintunes import player as player_module, tagging
from lintunes.cast.discovery import CastDevice
from lintunes.cast.server import KEEP_TOKENS, TrackServer
from lintunes.cast.sink import (
    BACKDROP_APP_ID, DEFAULT_RECEIVER_APP_ID, CastSink, app_stolen,
    connection_verdict, duration_ms_from, is_load_error, is_natural_end,
    is_playing_state, position_ms_from)
from lintunes.cast.support import (
    content_type_for, image_type_for, local_ip_for, parse_range,
    uncastable_reason)
from lintunes.gui.cast_dialog import ChromecastDialog
from lintunes.gui.icons import transport_icon
from lintunes.gui.spinner import Spinner
from lintunes.models import Track
from lintunes.models.library import Library
from lintunes.player import PlaybackSink, Player


# ---- the format gate ----

class TestContentType:
    def test_mp3_and_aac_cast_natively(self):
        assert content_type_for("/m/a.mp3", "MPEG audio file") == "audio/mpeg"
        assert content_type_for("/m/a.mpga") == "audio/mpeg"
        assert content_type_for("/m/a.m4a", "Matched AAC audio file") == "audio/mp4"
        assert content_type_for("/m/a.m4a", "AAC audio file") == "audio/mp4"
        assert content_type_for("/m/a.wav", "WAV audio file") == "audio/wav"

    def test_purchased_aac_is_drm_free_and_casts(self):
        # iTunes Plus downloads are unencrypted; only "Protected AAC" isn't.
        assert content_type_for("/m/a.m4a", "Purchased AAC audio file") == "audio/mp4"

    def test_lossless_and_protected_are_refused(self):
        assert content_type_for("/m/a.m4a", "Apple Lossless audio file") is None
        assert content_type_for("/m/a.m4a", "Protected AAC audio file") is None
        assert content_type_for("/m/a.m4p") is None
        assert content_type_for("/m/a.aiff", "AIFF audio file") is None
        assert content_type_for("/m/a.aif") is None

    def test_unknown_container_is_refused(self):
        assert content_type_for("/m/a.wma") is None
        assert content_type_for("") is None

    def test_suffix_case_is_ignored(self):
        assert content_type_for("/m/A.MP3") == "audio/mpeg"

    def test_reason_names_the_format(self):
        assert "Apple Lossless" in uncastable_reason(
            "/m/a.m4a", "Apple Lossless audio file")
        assert "copy-protected AAC" in uncastable_reason("/m/a.m4p")
        assert "AIFF" in uncastable_reason("/m/a.aiff")
        assert "WMA" in uncastable_reason("/m/a.wma")


class TestImageType:
    def test_sniffs_the_common_cover_formats(self):
        assert image_type_for(b"\xff\xd8\xff\xe0rest") == "image/jpeg"
        assert image_type_for(b"\x89PNG\r\n\x1a\nrest") == "image/png"
        assert image_type_for(b"GIF89a...") == "image/gif"
        assert image_type_for(b"RIFF\x00\x00\x00\x00WEBPVP8 ") == "image/webp"

    def test_unrecognized_bytes_are_refused(self):
        # Better no cover than a cover the receiver silently drops: the type
        # is sniffed rather than trusted, because ID3 APIC mimes are often
        # wrong or blank.
        assert image_type_for(b"") is None
        assert image_type_for(b"not an image at all") is None
        assert image_type_for(b"\xff\xd8") is None       # truncated JPEG magic


# ---- byte ranges ----

class TestParseRange:
    def test_no_header_serves_whole_file(self):
        assert parse_range(None, 100) is None
        assert parse_range("", 100) is None

    def test_closed_range(self):
        assert parse_range("bytes=10-19", 100) == (10, 19)

    def test_open_ended_range_runs_to_the_end(self):
        assert parse_range("bytes=10-", 100) == (10, 99)

    def test_suffix_range_takes_the_tail(self):
        assert parse_range("bytes=-20", 100) == (80, 99)
        assert parse_range("bytes=-500", 100) == (0, 99)  # longer than the file

    def test_end_past_eof_is_clamped(self):
        assert parse_range("bytes=90-999", 100) == (90, 99)

    def test_unsatisfiable_and_malformed_fall_back_to_whole_file(self):
        # A 200 with the full body is a legal answer to any Range request and
        # keeps the device playing instead of erroring it out.
        assert parse_range("bytes=100-", 100) is None   # start at/past EOF
        assert parse_range("bytes=50-10", 100) is None  # inverted
        assert parse_range("bytes=abc-def", 100) is None
        assert parse_range("items=0-10", 100) is None
        assert parse_range("bytes=0-10,20-30", 100) is None  # multi-range
        assert parse_range("bytes=0-10", 0) is None     # empty file


# ---- source address selection ----

class TestLocalIp:
    def test_returns_the_address_that_routes_to_the_peer(self):
        # Route toward a public address; any answer is fine as long as it is a
        # real local v4 address and not loopback-by-accident.
        ip = local_ip_for("8.8.8.8")
        assert ip.count(".") == 3
        socket.inet_aton(ip)  # raises if it isn't a v4 address

    def test_unreachable_peer_still_resolves(self):
        # connect() on UDP only picks a route, so an unroutable-but-valid
        # address still yields the interface the kernel would use.
        assert local_ip_for("192.0.2.1").count(".") == 3


# ---- the media server ----

@pytest.fixture
def served(tmp_path):
    """A running TrackServer plus a helper to talk to it over loopback."""
    server = TrackServer()
    server.start()
    yield server
    server.stop()


def _request(server, path, method="GET", headers=None):
    conn = http.client.HTTPConnection("127.0.0.1", server.port, timeout=5)
    try:
        conn.request(method, path, headers=headers or {})
        response = conn.getresponse()
        return response.status, dict(response.getheaders()), response.read()
    finally:
        conn.close()


def _audio(tmp_path, name="song.mp3", data=None):
    path = tmp_path / name
    path.write_bytes(data if data is not None else bytes(range(256)) * 8)
    return path


class TestTrackServer:
    def test_serves_the_published_file_verbatim(self, served, tmp_path):
        path = _audio(tmp_path)
        token = served.publish(path, "audio/mpeg")

        status, headers, body = _request(served, f"/t/{token}")

        assert status == 200
        assert body == path.read_bytes()
        assert headers["Content-Type"] == "audio/mpeg"
        assert headers["Content-Length"] == str(path.stat().st_size)
        # Without this the device won't seek and won't report a duration.
        assert headers["Accept-Ranges"] == "bytes"

    def test_head_reports_size_without_a_body(self, served, tmp_path):
        path = _audio(tmp_path)
        token = served.publish(path, "audio/mpeg")

        status, headers, body = _request(served, f"/t/{token}", method="HEAD")

        assert status == 200
        assert body == b""
        assert headers["Content-Length"] == str(path.stat().st_size)

    def test_range_request_returns_the_slice(self, served, tmp_path):
        path = _audio(tmp_path)
        size = path.stat().st_size
        token = served.publish(path, "audio/mpeg")

        status, headers, body = _request(
            served, f"/t/{token}", headers={"Range": "bytes=10-19"})

        assert status == 206
        assert body == path.read_bytes()[10:20]
        assert headers["Content-Range"] == f"bytes 10-19/{size}"
        assert headers["Content-Length"] == "10"

    def test_open_ended_range_returns_the_tail(self, served, tmp_path):
        path = _audio(tmp_path)
        size = path.stat().st_size
        token = served.publish(path, "audio/mpeg")

        status, headers, body = _request(
            served, f"/t/{token}", headers={"Range": "bytes=100-"})

        assert status == 206
        assert body == path.read_bytes()[100:]
        assert headers["Content-Range"] == f"bytes 100-{size - 1}/{size}"

    def test_unknown_token_is_not_found(self, served):
        status, _, _ = _request(served, "/t/nonexistent")
        assert status == 404

    def test_no_path_can_be_smuggled_through_the_url(self, served, tmp_path):
        # There is no path in the request to traverse with — only a token
        # lookup — so these are 404s by construction, not by sanitizing.
        _audio(tmp_path)
        for probe in ("/t/../../../etc/passwd", "/etc/passwd",
                      "/t/%2e%2e%2fetc%2fpasswd", "/", f"/t/{tmp_path}"):
            status, _, _ = _request(served, probe)
            assert status == 404, probe

    def test_only_the_most_recent_tokens_stay_live(self, served, tmp_path):
        tokens = [served.publish(_audio(tmp_path, f"{i}.mp3"), "audio/mpeg")
                  for i in range(KEEP_TOKENS + 2)]

        for stale in tokens[:2]:
            status, _, _ = _request(served, f"/t/{stale}")
            assert status == 404
        for live in tokens[2:]:
            status, _, _ = _request(served, f"/t/{live}")
            assert status == 200

    def test_previous_track_stays_fetchable(self, served, tmp_path):
        # The device re-requests the outgoing file while draining its buffer.
        first = served.publish(_audio(tmp_path, "a.mp3"), "audio/mpeg")
        served.publish(_audio(tmp_path, "b.mp3"), "audio/mpeg")

        status, _, _ = _request(served, f"/t/{first}")
        assert status == 200

    def test_file_deleted_after_publishing_is_not_found(self, served, tmp_path):
        path = _audio(tmp_path)
        token = served.publish(path, "audio/mpeg")
        path.unlink()

        status, _, _ = _request(served, f"/t/{token}")
        assert status == 404

    def test_a_client_hanging_up_mid_stream_does_not_kill_the_server(
            self, served, tmp_path):
        # Exactly what the Chromecast does on every seek, pause and stop.
        big = _audio(tmp_path, "big.mp3", b"x" * (4 * 1024 * 1024))
        token = served.publish(big, "audio/mpeg")

        conn = http.client.HTTPConnection("127.0.0.1", served.port, timeout=5)
        conn.request("GET", f"/t/{token}")
        conn.getresponse().read(1024)
        conn.close()  # hang up mid-body

        # The server thread survived and still answers.
        status, _, body = _request(served, f"/t/{token}")
        assert status == 200
        assert len(body) == big.stat().st_size

    def test_url_for_uses_the_bound_port(self, served, tmp_path):
        token = served.publish(_audio(tmp_path), "audio/mpeg")
        assert served.url_for("192.168.1.5", token) == (
            f"http://192.168.1.5:{served.port}/t/{token}")

    def test_stop_is_idempotent_and_closes_the_port(self, tmp_path):
        server = TrackServer()
        port = server.start()
        server.publish(_audio(tmp_path), "audio/mpeg")
        server.stop()
        server.stop()  # must not raise

        with pytest.raises(OSError):
            conn = http.client.HTTPConnection("127.0.0.1", port, timeout=2)
            conn.request("GET", "/t/anything")
            conn.getresponse()

    def test_serves_album_art_from_memory(self, served):
        # Cover art lives in the audio file's tags, not as a file of its own.
        art = b"\x89PNG\r\n\x1a\n" + bytes(range(256)) * 4
        token = served.publish_bytes(art, "image/png")

        status, headers, body = _request(served, f"/t/{token}")

        assert status == 200
        assert body == art
        assert headers["Content-Type"] == "image/png"
        assert headers["Content-Length"] == str(len(art))

    def test_range_and_head_work_on_art_too(self, served):
        art = b"\x89PNG\r\n\x1a\n" + bytes(range(256))
        token = served.publish_bytes(art, "image/png")

        status, _, body = _request(
            served, f"/t/{token}", headers={"Range": "bytes=8-15"})
        assert status == 206
        assert body == art[8:16]

        status, headers, body = _request(served, f"/t/{token}", method="HEAD")
        assert status == 200 and body == b""
        assert headers["Content-Length"] == str(len(art))

    def test_art_and_audio_evict_separately(self, served, tmp_path):
        # A track's cover must not push the previous track's audio out from
        # under a device that is still fetching it.
        audio = served.publish(_audio(tmp_path), "audio/mpeg")
        for i in range(KEEP_TOKENS + 2):
            served.publish_bytes(b"\x89PNG\r\n\x1a\n" + bytes([i]), "image/png")

        status, _, _ = _request(served, f"/t/{audio}")
        assert status == 200

    def test_tokens_do_not_survive_a_restart(self, tmp_path):
        server = TrackServer()
        server.start()
        token = server.publish(_audio(tmp_path), "audio/mpeg")
        server.stop()
        server.start()
        try:
            assert server.lookup(token) is None
        finally:
            server.stop()


# ---- reading the device's status ----

def _status(**fields):
    fields.setdefault("player_state", "PLAYING")
    fields.setdefault("idle_reason", None)
    fields.setdefault("media_session_id", 1)
    return SimpleNamespace(**fields)


class TestStatusReading:
    def test_finished_is_a_natural_end(self):
        assert is_natural_end(_status(player_state="IDLE", idle_reason="FINISHED"))

    def test_other_idle_reasons_are_not_a_finish(self):
        # Counting a play or advancing on these would be wrong: ERROR means it
        # never played, INTERRUPTED means something else took the device.
        for reason in ("ERROR", "INTERRUPTED", "CANCELLED", None):
            assert not is_natural_end(
                _status(player_state="IDLE", idle_reason=reason))
        assert is_load_error(_status(player_state="IDLE", idle_reason="ERROR"))

    def test_buffering_still_counts_as_playing(self):
        # Otherwise the play button flickers every time the device re-buffers.
        assert is_playing_state(_status(player_state="BUFFERING"))
        assert is_playing_state(_status(player_state="PLAYING"))
        assert not is_playing_state(_status(player_state="PAUSED"))

    def test_position_extrapolates_only_while_playing(self):
        playing = _status(player_state="PLAYING",
                          adjusted_current_time=12.5, current_time=10.0)
        assert position_ms_from(playing, 0) == 12_500

        # adjusted_current_time keeps creeping off the wall clock while
        # paused, which would drift the seek slider during a pause.
        paused = _status(player_state="PAUSED",
                         adjusted_current_time=99.0, current_time=10.0)
        assert position_ms_from(paused, 0) == 10_000

    def test_position_falls_back_when_the_device_says_nothing(self):
        assert position_ms_from(None, 4321) == 4321
        assert position_ms_from(
            _status(adjusted_current_time=None, current_time=None), 4321) == 4321

    def test_duration_falls_back_to_the_library_value(self):
        assert duration_ms_from(_status(duration=180.0), 0) == 180_000
        assert duration_ms_from(_status(duration=None), 200_000) == 200_000
        assert duration_ms_from(None, 200_000) == 200_000

    def test_foreign_receiver_reads_as_stolen(self):
        assert app_stolen("YouTube")
        assert not app_stolen(DEFAULT_RECEIVER_APP_ID)
        assert not app_stolen(BACKDROP_APP_ID)
        assert not app_stolen(None)
        assert not app_stolen("")

    def test_lost_is_a_grace_period_not_a_disconnect(self):
        # pychromecast retries a dropped socket on its own, so a Wi-Fi blip
        # must not dump playback back to the laptop speakers.
        assert connection_verdict("LOST") == "grace"
        assert connection_verdict("CONNECTING") == "grace"
        assert connection_verdict("CONNECTED") == "ok"
        assert connection_verdict("DISCONNECTED") == "lost"
        assert connection_verdict("FAILED") == "lost"
        assert connection_verdict("FAILED_RESOLVE") == "lost"


# ---- Player's sink contract ----

class _FakeSink(PlaybackSink):
    """A sink that records what Player asked of it, standing in for a cast."""

    def __init__(self, playable=True, parent=None):
        super().__init__(parent)
        self.name = "Kitchen"
        self.loads = []
        self.volume = None
        self.stopped = 0
        self.shutdowns = 0
        self._playing = False
        self._position = 0
        self._playable = playable

    def load(self, track, autoplay, start_ms):
        self.loads.append((track.track_id, autoplay, start_ms))
        self._playing = autoplay
        self._position = start_ms

    def play(self):
        self._playing = True

    def pause(self):
        self._playing = False

    def stop(self):
        self.stopped += 1
        self._playing = False

    def seek(self, position_ms):
        self._position = position_ms

    def position_ms(self):
        return self._position

    def duration_ms(self):
        return 0

    def is_playing(self):
        return self._playing

    def set_volume(self, level):
        self.volume = level

    def can_play(self, track):
        return self._playable

    def unplayable_reason(self, track):
        return "a Chromecast can't play Apple Lossless"

    def shutdown(self):
        self.shutdowns += 1


def _player(qapp, tracks):
    library = Library()
    for track in tracks:
        library.tracks[track.track_id] = track
    with patch.multiple(
        player_module,
        QMediaPlayer=MagicMock(),
        QAudioOutput=MagicMock(),
        QAudioBufferOutput=MagicMock(),
        QMediaDevices=MagicMock(),
    ):
        return Player(MagicMock(library=library))


def _local_track(tmp_path, track_id, name="Song", **fields):
    path = tmp_path / f"{track_id}.mp3"
    path.write_bytes(b"x")
    return Track(track_id=track_id, name=name, location=str(path), **fields)


class TestPlayerSinks:
    def test_default_sink_is_local(self, qapp):
        player = _player(qapp, [])
        assert player.active_sink() is player._local
        assert player.provides_pcm() is True

    def test_swap_carries_track_position_and_playing_state(self, qapp, tmp_path):
        track = _local_track(tmp_path, 1, total_time=200_000)
        player = _player(qapp, [track])
        player.play_queue([1], 0)
        player._local._media.position.return_value = 42_000
        player._local._media.playbackState.return_value = \
            player_module.QMediaPlayer.PlaybackState.PlayingState

        sink = _FakeSink()
        player.set_sink(sink)

        assert player.active_sink() is sink
        assert sink.loads == [(1, True, 42_000)]   # same song, same spot
        assert player.provides_pcm() is False

    def test_swap_does_not_re_announce_the_track(self, qapp, tmp_path):
        # A second track_changed would fire a duplicate Last.fm now-playing
        # and reset an in-progress bpm tap, for what is the same song.
        track = _local_track(tmp_path, 1)
        player = _player(qapp, [track])
        player.play_queue([1], 0)
        seen = []
        player.track_changed.connect(seen.append)

        player.set_sink(_FakeSink())

        assert seen == []

    def test_volume_follows_the_swap_both_ways(self, qapp):
        player = _player(qapp, [])
        player.set_volume(0.42)

        sink = _FakeSink()
        player.set_sink(sink)
        assert sink.volume == 0.42          # the slider keeps its meaning

        player.set_volume(0.7)
        assert sink.volume == 0.7

        player.set_sink(None)
        assert player.volume() == 0.7
        assert player.active_sink() is player._local

    def test_swapping_to_the_same_sink_is_a_no_op(self, qapp):
        player = _player(qapp, [])
        sink = _FakeSink()
        player.set_sink(sink)
        player.set_sink(sink)
        assert sink.stopped == 0

    def test_natural_end_on_a_cast_records_the_play_and_advances(
            self, qapp, tmp_path):
        t1 = _local_track(tmp_path, 1, total_time=200_000)
        t2 = _local_track(tmp_path, 2, total_time=200_000)
        player = _player(qapp, [t1, t2])
        finished = []
        player.track_finished.connect(lambda t: finished.append(t.track_id))
        player.play_queue([1, 2], 0)
        sink = _FakeSink()
        player.set_sink(sink)

        sink.ended.emit()

        player._manager.record_play.assert_called_once_with(1)
        assert finished == [1]
        assert player.current_track.track_id == 2

    def test_repeated_end_counts_the_play_once(self, qapp, tmp_path):
        track = _local_track(tmp_path, 1, total_time=200_000)
        player = _player(qapp, [track])
        player.play_queue([1], 0)
        sink = _FakeSink()
        player.set_sink(sink)

        sink.ended.emit()
        sink.ended.emit()

        player._manager.record_play.assert_called_once_with(1)

    def test_custom_stop_time_still_ends_early_on_a_cast(self, qapp, tmp_path):
        t1 = _local_track(tmp_path, 1, total_time=200_000, stop_time=60_000)
        t2 = _local_track(tmp_path, 2, total_time=200_000)
        player = _player(qapp, [t1, t2])
        player.play_queue([1, 2], 0)
        sink = _FakeSink()
        player.set_sink(sink)

        sink.position_changed.emit(59_000)
        assert player._manager.record_play.call_count == 0
        sink.position_changed.emit(60_500)

        player._manager.record_play.assert_called_once_with(1)
        assert player.current_track.track_id == 2

    def test_custom_start_time_reaches_the_sink(self, qapp, tmp_path):
        track = _local_track(tmp_path, 1, total_time=200_000, start_time=30_000)
        player = _player(qapp, [track])
        sink = _FakeSink()
        player.set_sink(sink)

        player.play_queue([1], 0)

        assert sink.loads[-1] == (1, True, 30_000)

    def test_unplayable_track_is_skipped_not_dialogged(self, qapp, tmp_path):
        t1 = _local_track(tmp_path, 1, name="Lossless")
        t2 = _local_track(tmp_path, 2, name="Fine")
        player = _player(qapp, [t1, t2])
        skipped, errors = [], []
        player.track_unplayable.connect(
            lambda track, reason: skipped.append((track.name, reason)))
        # error_occurred opens a modal dialog in the GUI — the wrong channel
        # for "this one track's format isn't supported".
        player.error_occurred.connect(errors.append)

        sink = _FakeSink()
        # Only the second track is castable.
        sink.can_play = lambda track: track.track_id == 2
        player.set_sink(sink)
        player.play_queue([1, 2], 0)

        assert [name for name, _ in skipped] == ["Lossless"]
        assert "Apple Lossless" in skipped[0][1]
        assert errors == []
        assert player.current_track.track_id == 2

    def test_a_queue_of_unplayable_tracks_stops_without_recursing(
            self, qapp, tmp_path):
        tracks = [_local_track(tmp_path, i) for i in range(1, 600)]
        player = _player(qapp, tracks)
        skipped = []
        player.track_unplayable.connect(lambda t, r: skipped.append(t))

        player.set_sink(_FakeSink(playable=False))
        player.play_queue([t.track_id for t in tracks], 0)

        assert len(skipped) == len(tracks)
        assert player.current_track is None      # stopped cleanly, no crash

    def test_a_dead_sink_falls_back_to_local_playback(self, qapp, tmp_path):
        track = _local_track(tmp_path, 1, total_time=200_000)
        player = _player(qapp, [track])
        player.play_queue([1], 0)
        sink = _FakeSink()
        player.set_sink(sink)
        sink._position = 55_000
        sink._playing = True
        errors = []
        player.error_occurred.connect(errors.append)

        sink.unavailable.emit("Lost the connection to Kitchen.")

        assert player.active_sink() is player._local
        assert errors == ["Lost the connection to Kitchen."]

    def test_missing_file_still_reports_on_a_cast_sink(self, qapp, tmp_path):
        track = Track(track_id=1, name="Ghost",
                      location=str(tmp_path / "gone" / "a.mp3"))
        player = _player(qapp, [track])
        missing = []
        player.track_missing.connect(missing.append)

        player.set_sink(_FakeSink())
        player.play_queue([1], 0)

        assert missing == [track]
        assert player.current_track is None

    def test_shutdown_tears_down_both_sinks(self, qapp):
        player = _player(qapp, [])
        sink = _FakeSink()
        player.set_sink(sink)

        player.shutdown()

        assert sink.shutdowns == 1
        # The local sink is kept alive across a swap, so its Qt Multimedia
        # pipeline still needs the ordered teardown that avoids a segfault.
        player._local._media.setAudioOutput.assert_called_with(None)


# ---- the cast sink's failure handling ----

class _FakeCast:
    """Enough of a pychromecast Chromecast to build a CastSink against."""

    def __init__(self):
        self.cast_info = SimpleNamespace(friendly_name="Kitchen", host="10.0.0.5")
        self.status = SimpleNamespace(volume_level=0.8)
        self.media_controller = SimpleNamespace(
            status=None, register_status_listener=lambda _l: None,
            play_media=lambda *a, **k: None, play=lambda: None,
            pause=lambda: None, stop=lambda: None, seek=lambda _s: None)
        self.volumes = []
        self.quit_calls = 0
        self.disconnects = 0

    def register_status_listener(self, listener):
        pass

    def register_connection_listener(self, listener):
        pass

    def set_volume(self, level):
        self.volumes.append(level)

    def quit_app(self):
        self.quit_calls += 1

    def disconnect(self):
        self.disconnects += 1


@pytest.fixture
def cast_sink(qapp):
    cast = _FakeCast()
    server = TrackServer()
    sink = CastSink(cast, server)
    yield sink, cast
    sink.shutdown()


class TestCastSinkFailure:
    def test_failure_reports_the_state_lintunes_asked_for(self, cast_sink):
        sink, _cast = cast_sink
        sink._intent_playing = True
        sink._playing = True
        # A dying connection pushes an IDLE media status just before the
        # disconnect. Believing it here would stop the music on the fallback.
        sink._on_status(_status(player_state="IDLE", idle_reason=None))
        assert sink.is_playing() is False

        reasons = []
        sink.unavailable.connect(reasons.append)
        sink._on_connection("DISCONNECTED")

        assert reasons == ["Lost the connection to Kitchen."]
        assert sink.is_playing() is True     # so Player resumes locally playing

    def test_a_paused_session_does_not_start_playing_on_failure(self, cast_sink):
        sink, _cast = cast_sink
        sink._intent_playing = False
        sink._on_connection("DISCONNECTED")
        assert sink.is_playing() is False

    def test_failure_fires_once(self, cast_sink):
        sink, _cast = cast_sink
        reasons = []
        sink.unavailable.connect(reasons.append)

        sink._on_connection("DISCONNECTED")
        sink._on_connection("FAILED")
        sink._on_grace_expired()

        assert len(reasons) == 1

    def test_a_dropped_socket_gets_a_grace_period(self, cast_sink):
        sink, _cast = cast_sink
        reasons = []
        sink.unavailable.connect(reasons.append)

        sink._on_connection("LOST")
        assert sink._lost_timer.isActive()
        assert reasons == []        # pychromecast is still retrying

        sink._on_connection("CONNECTED")
        assert not sink._lost_timer.isActive()
        assert reasons == []        # the blip healed itself

    def test_grace_expiring_gives_up(self, cast_sink):
        sink, _cast = cast_sink
        reasons = []
        sink.unavailable.connect(reasons.append)

        sink._on_connection("LOST")
        sink._on_grace_expired()

        assert reasons == ["Kitchen stopped responding."]

    def test_another_app_taking_the_device_ends_the_session(self, cast_sink):
        sink, _cast = cast_sink
        reasons = []
        sink.unavailable.connect(reasons.append)

        sink._on_app(DEFAULT_RECEIVER_APP_ID)
        sink._on_app(BACKDROP_APP_ID)
        assert reasons == []

        sink._on_app("YouTube")
        assert reasons == ["Something else started casting to Kitchen."]

    def test_natural_end_is_reported_once_per_session(self, cast_sink):
        sink, _cast = cast_sink
        ends = []
        sink.ended.connect(lambda: ends.append(True))

        done = _status(player_state="IDLE", idle_reason="FINISHED",
                       media_session_id=7)
        sink._on_status(done)
        sink._on_status(done)     # the device repeats its final status

        assert ends == [True]

    def test_shutdown_hands_the_device_volume_back(self, qapp):
        cast = _FakeCast()
        sink = CastSink(cast, TrackServer())
        sink.set_volume(0.2)

        sink.shutdown()
        sink.shutdown()           # idempotent

        # The device's own volume persists for whatever plays next, so the
        # session's setting must not be left behind.
        assert cast.volumes[-1] == 0.8
        assert cast.quit_calls == 1
        assert cast.disconnects == 1

    def test_embedded_cover_is_published_for_the_device(
            self, cast_sink, monkeypatch):
        sink, _cast = cast_sink
        art = b"\xff\xd8\xff\xe0" + b"jpegbody"
        monkeypatch.setattr(tagging, "read_embedded_artwork", lambda _p: art)

        token = sink._publish_artwork(
            Track(track_id=1, name="A", location="/m/a.mp3"))

        assert token is not None
        asset = sink._server.lookup(token)
        assert asset.data == art
        assert asset.content_type == "image/jpeg"

    def test_a_track_without_a_cover_still_plays(self, cast_sink, monkeypatch):
        sink, _cast = cast_sink
        track = Track(track_id=1, name="A", location="/m/a.mp3")

        monkeypatch.setattr(tagging, "read_embedded_artwork", lambda _p: None)
        assert sink._publish_artwork(track) is None

        # Unreadable tags, or junk where the cover should be, are equally
        # never worth failing the load over.
        monkeypatch.setattr(tagging, "read_embedded_artwork", lambda _p: b"junk")
        assert sink._publish_artwork(track) is None

        def boom(_p):
            raise OSError("drive went away")
        monkeypatch.setattr(tagging, "read_embedded_artwork", boom)
        assert sink._publish_artwork(track) is None

    def test_uncastable_track_is_refused_by_the_sink(self, cast_sink):
        sink, _cast = cast_sink
        alac = Track(track_id=1, name="Lossless", location="/m/a.m4a",
                     kind="Apple Lossless audio file")
        mp3 = Track(track_id=2, name="Fine", location="/m/a.mp3")

        assert not sink.can_play(alac)
        assert "Apple Lossless" in sink.unplayable_reason(alac)
        assert sink.can_play(mp3)


# ---- widgets ----

class _FakeDiscovery(QObject):
    device_found = pyqtSignal(object)
    device_lost = pyqtSignal(object)
    failed = pyqtSignal(str)

    def __init__(self):
        super().__init__()
        self.started = 0
        self.stopped = 0

    def devices(self):
        return []

    def start(self):
        self.started += 1

    def stop(self):
        self.stopped += 1


def _device(uuid="u1", name="Kitchen", model="Chromecast Audio"):
    return CastDevice(uuid=uuid, name=name, model=model,
                      host="10.0.0.5", port=8009)


class TestTransportLayout:
    def test_casting_leaves_the_bar_layout_alone(self, qapp, tmp_path):
        from lintunes.gui.transport import BAR_HEIGHT, TransportBar
        from lintunes.preferences import Preferences

        player = _player(qapp, [])
        bar = TransportBar(player, MagicMock(), Preferences(tmp_path))
        bar.resize(1200, BAR_HEIGHT)
        bar.show()
        idle_y = bar._volume_slider.pos().y()

        # The visualizer panel doubles as the cast indicator, so connecting
        # adds no widget and nothing moves.
        player.set_sink(_FakeSink())
        bar.layout().activate()

        assert bar._volume_slider.pos().y() == idle_y
        assert bar.sizeHint().height() <= BAR_HEIGHT
        bar.hide()

    def test_clicking_the_visualizer_while_casting_disconnects(
            self, qapp, tmp_path):
        from lintunes.gui.transport import TransportBar
        from lintunes.preferences import Preferences

        player = _player(qapp, [])
        cast = MagicMock()
        bar = TransportBar(player, MagicMock(), Preferences(tmp_path),
                           cast=cast)
        vis = bar._visualizer
        before = vis._mode

        # Not casting: a click still cycles the brightness as it always did.
        vis.mousePressEvent(None)
        assert vis._mode != before
        cast.disconnect.assert_not_called()

        player.set_sink(_FakeSink())
        mode_while_casting = vis._mode
        vis.mousePressEvent(None)

        cast.disconnect.assert_called_once()
        # Brightness must not cycle — it means nothing when there are no bars.
        assert vis._mode == mode_while_casting


class TestCastDialog:
    def test_devices_appear_and_disappear(self, qapp):
        discovery = _FakeDiscovery()
        dialog = ChromecastDialog(discovery)
        assert discovery.started == 1

        discovery.device_found.emit(_device())
        assert dialog._list.count() == 1
        assert "Kitchen" in dialog._list.item(0).text()

        # zeroconf re-announces constantly; the same device must not stack up.
        discovery.device_found.emit(_device())
        assert dialog._list.count() == 1

        discovery.device_lost.emit(_device())
        assert dialog._list.count() == 0

    def test_connect_needs_a_selection(self, qapp):
        discovery = _FakeDiscovery()
        dialog = ChromecastDialog(discovery)
        assert not dialog._connect_button.isEnabled()

        discovery.device_found.emit(_device())
        assert dialog._connect_button.isEnabled()

        dialog._on_connect()
        assert dialog.selected_device.name == "Kitchen"

    def test_closing_always_stops_the_search(self, qapp):
        for close in (lambda d: d.reject(), lambda d: d.accept(),
                      lambda d: d.done(0)):
            discovery = _FakeDiscovery()
            dialog = ChromecastDialog(discovery)
            close(dialog)
            assert discovery.stopped == 1

    def test_missing_dependency_is_reported_not_spun_forever(self, qapp):
        discovery = _FakeDiscovery()
        dialog = ChromecastDialog(discovery)

        discovery.failed.emit("Chromecast support isn't installed.")

        assert "isn't installed" in dialog._status.text()
        assert dialog._spinner.isHidden()


class TestSpinner:
    def test_animates_only_while_visible(self, qapp):
        spinner = Spinner()
        spinner.show()
        assert spinner._timer.isActive()
        spinner.hide()
        assert not spinner._timer.isActive()


class TestCastIcons:
    def test_cast_glyph_renders(self, qapp):
        from PyQt6.QtGui import QColor
        assert not transport_icon("cast", QColor("#4A4A4A")).isNull()

    def test_a_bigger_glyph_is_repainted_not_upscaled(self, qapp):
        from PyQt6.QtGui import QColor
        big = transport_icon("cast", QColor("#4A4A4A"), size=40)
        # Rendered at the requested size rather than a stretched 20px pixmap,
        # so the panel-sized glyph stays crisp.
        assert big.pixmap(40, 40).size().width() == 40
        assert big.availableSizes()[0].width() == 40

    def test_size_is_part_of_the_cache_key(self, qapp):
        from PyQt6.QtGui import QColor
        small = transport_icon("cast", QColor("#4A4A4A"), size=20)
        big = transport_icon("cast", QColor("#4A4A4A"), size=40)
        assert small.availableSizes()[0].width() == 20
        assert big.availableSizes()[0].width() == 40


class TestVisualizerWhileCasting:
    def test_stops_animating_when_the_sink_has_no_pcm(self, qapp, tmp_path):
        from lintunes.gui.visualizer import VisualizerWidget
        from lintunes.preferences import Preferences

        player = _player(qapp, [])
        vis = VisualizerWidget(player, Preferences(tmp_path))

        player.set_sink(_FakeSink())
        assert vis._no_pcm is True
        assert not vis._timer.isActive()
        # A play/pause must not restart an FFT over a buffer of zeros.
        player.playing_changed.emit(True)
        assert not vis._timer.isActive()

        player.set_sink(None)
        assert vis._no_pcm is False

    def test_names_the_device_it_is_casting_to(self, qapp, tmp_path):
        from lintunes.gui.visualizer import VisualizerWidget
        from lintunes.preferences import Preferences

        player = _player(qapp, [])
        vis = VisualizerWidget(player, Preferences(tmp_path))

        player.set_sink(_FakeSink())           # _FakeSink.name == "Kitchen"
        assert "Kitchen" in vis.toolTip()
        assert "click to stop" in vis.toolTip()

        player.set_sink(None)
        assert "cycle" in vis.toolTip()        # back to the brightness control