"""Round 32: custom start times survive a track change. Round 22 armed the start-time seek and consumed it on the next LoadedMedia. But QMediaPlayer.setSource() synchronously emits *two* status changes before it returns: a LoadedMedia that still reports the outgoing source (the previous track dropping back from BufferedMedia), then LoadingMedia for the new one. The stale first event consumed the armed seek, so the new track's real LoadedMedia found nothing armed and playback started at 0:00 — for every track change, i.e. everything except the first track after launch. The armed seek is now tagged with the URL it belongs to, so only the matching source's LoadedMedia can consume it. These tests model the real Qt event sequence, which Round 22's plain MagicMock never emitted. """ from unittest.mock import MagicMock, patch from PyQt6.QtCore import QUrl from lintunes.models import Library, Track from lintunes import player as player_module from lintunes.player import Player def _track(tid, **kw): return Track(track_id=tid, name=kw.pop("name", f"Track {tid}"), **kw) def _player(qapp, tracks): library = Library() for t in tracks: library.tracks[t.track_id] = t manager = MagicMock() manager.library = library with patch.multiple( player_module, QMediaPlayer=MagicMock(), QAudioOutput=MagicMock(), QAudioBufferOutput=MagicMock(), QMediaDevices=MagicMock(), ): player = Player(manager) return player, manager def _status(name): """The patched QMediaPlayer's enum members are stable MagicMock identities, which is all `_on_media_status` compares against.""" return getattr(player_module.QMediaPlayer.MediaStatus, name) def _wire_real_qt_sequence(sink): """Make setSource() behave like the FFmpeg backend really does: fire a stale LoadedMedia for the media being replaced, *then* switch the reported source and fire LoadingMedia — all synchronously, inside the call.""" def side_effect(url): if sink._media.source.return_value not in (None, QUrl()): sink._on_media_status(_status("LoadedMedia")) # outgoing media sink._media.source.return_value = url sink._on_media_status(_status("LoadingMedia")) sink._media.source.return_value = QUrl() sink._media.setSource.side_effect = side_effect class TestStartTimeAcrossTrackChange: def test_stale_loadedmedia_does_not_eat_the_armed_start(self, qapp, tmp_path): for name in ("a.mp3", "b.mp3"): (tmp_path / name).write_bytes(b"x") t1 = _track(1, location=str(tmp_path / "a.mp3"), total_time=200_000) t2 = _track(2, location=str(tmp_path / "b.mp3"), total_time=202_240, start_time=40_000) player, _ = _player(qapp, [t1, t2]) sink = player._local _wire_real_qt_sequence(sink) player.play_queue([1, 2], 0) # a.mp3 becomes the loaded source sink._on_media_status(_status("LoadedMedia")) sink._media.setPosition.reset_mock() player.next() # switch to the start-timed track # The stale LoadedMedia fired inside setSource() must not have consumed # the armed seek, nor issued one against the outgoing media. sink._media.setPosition.assert_not_called() assert sink._pending_start_ms == 40_000 sink._on_media_status(_status("LoadedMedia")) # the real one sink._media.setPosition.assert_called_once_with(40_000) assert sink._pending_start_ms == 0 def test_stale_loadedmedia_alone_leaves_the_seek_armed(self, qapp, tmp_path): (tmp_path / "b.mp3").write_bytes(b"x") track = _track(1, location=str(tmp_path / "b.mp3"), total_time=200_000, start_time=30_000) player, _ = _player(qapp, [track]) sink = player._local sink._media.source.return_value = QUrl.fromLocalFile("/somewhere/old.mp3") player.play_queue([1], 0) assert sink._pending_start_ms == 30_000 # source() still reports the outgoing file: this status change is not ours. sink._on_media_status(_status("LoadedMedia")) sink._media.setPosition.assert_not_called() assert sink._pending_start_ms == 30_000 def test_end_of_track_advance_lands_on_the_start_time(self, qapp, tmp_path): for name in ("a.mp3", "b.mp3"): (tmp_path / name).write_bytes(b"x") t1 = _track(1, location=str(tmp_path / "a.mp3"), total_time=200_000) t2 = _track(2, location=str(tmp_path / "b.mp3"), total_time=200_000, start_time=40_000) player, _ = _player(qapp, [t1, t2]) sink = player._local _wire_real_qt_sequence(sink) player.play_queue([1, 2], 0) sink._on_media_status(_status("LoadedMedia")) sink._media.setPosition.reset_mock() player._on_sink_ended() # t1 finished naturally # Nothing may be seeked while the new source is still loading — a seek # issued here lands on the outgoing pipeline and is thrown away. sink._media.setPosition.assert_not_called() sink._on_media_status(_status("LoadedMedia")) sink._media.setPosition.assert_called_once_with(40_000) def test_track_without_start_time_never_seeks(self, qapp, tmp_path): for name in ("a.mp3", "b.mp3"): (tmp_path / name).write_bytes(b"x") t1 = _track(1, location=str(tmp_path / "a.mp3"), total_time=200_000) t2 = _track(2, location=str(tmp_path / "b.mp3"), total_time=200_000) player, _ = _player(qapp, [t1, t2]) sink = player._local _wire_real_qt_sequence(sink) player.play_queue([1, 2], 0) player.next() sink._on_media_status(_status("LoadedMedia")) sink._media.setPosition.assert_not_called() class TestSameSourceReplayStillWorks: """Round 22's path, re-checked against the restructured load().""" def test_unchanged_url_seeks_directly_and_arms_nothing(self, qapp, tmp_path): (tmp_path / "a.mp3").write_bytes(b"x") loc = str(tmp_path / "a.mp3") track = _track(1, location=loc, total_time=200_000, start_time=30_000) player, _ = _player(qapp, [track]) sink = player._local sink._media.source.return_value = QUrl.fromLocalFile(loc) player.play_queue([1], 0) sink._media.setSource.assert_not_called() sink._media.stop.assert_called() sink._media.setPosition.assert_called_once_with(30_000) assert sink._pending_start_ms == 0 assert sink._pending_start_url == QUrl() def test_unchanged_url_without_start_time_does_not_seek(self, qapp, tmp_path): (tmp_path / "a.mp3").write_bytes(b"x") loc = str(tmp_path / "a.mp3") track = _track(1, location=loc, total_time=200_000) player, _ = _player(qapp, [track]) sink = player._local sink._media.source.return_value = QUrl.fromLocalFile(loc) player.play_queue([1], 0) sink._media.stop.assert_called() sink._media.setPosition.assert_not_called()