NikkeDoy
c865637b8e
All checks were successful
SonarQube Code Quality Scan / SonarQube Scan (push) Successful in 1m51s
577 lines
22 KiB
Python
577 lines
22 KiB
Python
"""Integration tests for GUI widget methods (requires real Qt runtime).
|
|
|
|
These tests create actual QWidget instances and test the new/modified
|
|
methods that were added during the recent code quality refactoring.
|
|
|
|
When PySide6 cannot connect to a display server (e.g. headless CI)
|
|
the conftest.py stubs out all Qt types with MagicMock instances — in that
|
|
environment widget instantiation returns MagicMock objects that do not
|
|
execute the real ``__init__`` body, so these tests are skipped.
|
|
"""
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
# The conftest stubs PySide6.QtWidgets as a MagicMock when Qt is
|
|
# unavailable (headless CI). We check whether we have a real Qt runtime
|
|
# by verifying QApplication is not a MagicMock and that it can be created.
|
|
from PySide6.QtWidgets import QApplication, QWidget
|
|
|
|
def _qt_available() -> bool:
|
|
"""Return True if a real QApplication can be created (display available)."""
|
|
if isinstance(QApplication, mock.MagicMock):
|
|
return False
|
|
inst = QApplication.instance()
|
|
if inst is not None:
|
|
return True
|
|
try:
|
|
app = QApplication([])
|
|
app.quit()
|
|
return True
|
|
except (RuntimeError, OSError):
|
|
return False
|
|
|
|
if not _qt_available():
|
|
pytest.skip("Real Qt runtime not available (stubs active / headless)",
|
|
allow_module_level=True)
|
|
|
|
from PySide6.QtCore import QTimer
|
|
from PySide6.QtWidgets import QPushButton
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _cleanup(widget: QWidget, app: QApplication) -> None:
|
|
"""Safely delete a widget and process pending events."""
|
|
widget.deleteLater()
|
|
app.processEvents()
|
|
app.processEvents()
|
|
|
|
|
|
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def app():
|
|
"""Reuse the singleton QApplication."""
|
|
inst = QApplication.instance()
|
|
if inst is None:
|
|
inst = QApplication([])
|
|
return inst
|
|
|
|
|
|
@pytest.fixture
|
|
def player_mock():
|
|
return mock.MagicMock(name="player")
|
|
|
|
|
|
@pytest.fixture
|
|
def visualizer_mock():
|
|
return mock.MagicMock(name="visualizer")
|
|
|
|
|
|
class TestAudioVisualizer:
|
|
|
|
def test_reset_activity(self, app):
|
|
"""reset_activity sets _last_activity to 0."""
|
|
from gui import AudioVisualizer
|
|
|
|
viz = AudioVisualizer()
|
|
viz._last_activity = 999
|
|
viz.reset_activity()
|
|
assert viz._last_activity == 0
|
|
_cleanup(viz, app)
|
|
|
|
|
|
class TestPlaybackBar:
|
|
|
|
def test_clear_seek_flag(self, app, player_mock, visualizer_mock):
|
|
"""_clear_seek_flag sets _updating_progress to False."""
|
|
from gui import PlaybackBar
|
|
|
|
bar = PlaybackBar(player=player_mock, visualizer=visualizer_mock)
|
|
bar._updating_progress = True
|
|
bar._clear_seek_flag()
|
|
assert bar._updating_progress is False
|
|
_cleanup(bar, app)
|
|
|
|
def test_flush_volume_stops_timer_and_saves(self, app, player_mock,
|
|
visualizer_mock):
|
|
"""flush_volume persists the volume and stops the debounce timer."""
|
|
from gui import PlaybackBar
|
|
|
|
bar = PlaybackBar(player=player_mock, visualizer=visualizer_mock)
|
|
bar._saved_volume = 42
|
|
|
|
bar._vol_save_timer = QTimer()
|
|
bar._vol_save_timer.setSingleShot(True)
|
|
bar._vol_save_timer.start(5000)
|
|
assert bar._vol_save_timer.isActive()
|
|
|
|
with mock.patch("gui.save_volume") as mock_save:
|
|
bar.flush_volume()
|
|
mock_save.assert_called_once_with(42)
|
|
|
|
assert not bar._vol_save_timer.isActive()
|
|
_cleanup(bar, app)
|
|
|
|
def test_flush_volume_no_timer(self, app, player_mock, visualizer_mock):
|
|
"""flush_volume does not crash when _vol_save_timer doesn't exist."""
|
|
from gui import PlaybackBar
|
|
|
|
bar = PlaybackBar(player=player_mock, visualizer=visualizer_mock)
|
|
if hasattr(bar, "_vol_save_timer"):
|
|
del bar._vol_save_timer
|
|
|
|
bar._saved_volume = 75
|
|
with mock.patch("gui.save_volume") as mock_save:
|
|
bar.flush_volume()
|
|
mock_save.assert_called_once_with(75)
|
|
_cleanup(bar, app)
|
|
|
|
def test_show_error_uses_method_ref(self, app, player_mock,
|
|
visualizer_mock):
|
|
"""show_error schedules _clear_error via method ref, not lambda."""
|
|
from gui import PlaybackBar
|
|
|
|
bar = PlaybackBar(player=player_mock, visualizer=visualizer_mock)
|
|
with mock.patch.object(bar, "_clear_error") as mock_clear:
|
|
bar.show_error("Test error")
|
|
assert "Test error" in bar._np_title.text()
|
|
mock_clear.assert_not_called()
|
|
_cleanup(bar, app)
|
|
|
|
|
|
class TestSearchPage:
|
|
|
|
def test_on_show_more_clicked_from_button(self, app):
|
|
"""_on_show_more_clicked expands the correct category."""
|
|
from gui import SearchPage
|
|
|
|
page = SearchPage()
|
|
page._videos_expanded = False
|
|
|
|
btn = QPushButton("Show more")
|
|
btn.setProperty("category", "videos")
|
|
btn.clicked.connect(page._on_show_more_clicked)
|
|
|
|
with mock.patch.object(page, "_toggle_expand") as mock_toggle:
|
|
btn.click()
|
|
mock_toggle.assert_called_once_with("videos")
|
|
|
|
_cleanup(btn, app)
|
|
_cleanup(page, app)
|
|
|
|
def test_on_show_more_clicked_songs(self, app):
|
|
"""_on_show_more_clicked expands songs category."""
|
|
from gui import SearchPage
|
|
|
|
page = SearchPage()
|
|
page._songs_expanded = False
|
|
|
|
btn = QPushButton("Show more")
|
|
btn.setProperty("category", "songs")
|
|
btn.clicked.connect(page._on_show_more_clicked)
|
|
|
|
with mock.patch.object(page, "_toggle_expand") as mock_toggle:
|
|
btn.click()
|
|
mock_toggle.assert_called_once_with("songs")
|
|
|
|
_cleanup(btn, app)
|
|
_cleanup(page, app)
|
|
|
|
def test_toggle_expand_videos(self, app):
|
|
"""_toggle_expand sets videos expanded flag."""
|
|
from gui import SearchPage
|
|
|
|
page = SearchPage()
|
|
page._videos_expanded = False
|
|
|
|
with mock.patch.object(page, "_rebuild"):
|
|
page._toggle_expand("videos")
|
|
assert page._videos_expanded is True
|
|
_cleanup(page, app)
|
|
|
|
def test_toggle_expand_songs(self, app):
|
|
"""_toggle_expand sets songs expanded flag."""
|
|
from gui import SearchPage
|
|
|
|
page = SearchPage()
|
|
page._songs_expanded = False
|
|
|
|
with mock.patch.object(page, "_rebuild"):
|
|
page._toggle_expand("songs")
|
|
assert page._songs_expanded is True
|
|
_cleanup(page, app)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# TunettiWindow accessors
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestAudioPlayer:
|
|
"""Tests for AudioPlayer state methods (no playback required)."""
|
|
|
|
@pytest.fixture
|
|
def player(self, app):
|
|
from player import AudioPlayer
|
|
p = AudioPlayer()
|
|
yield p
|
|
p.shutdown()
|
|
|
|
def test_initial_state(self, player):
|
|
"""Initial player state is stopped, no current song, empty queue."""
|
|
assert player.is_stopped() is True
|
|
assert player.is_playing() is False
|
|
assert player.is_paused() is False
|
|
assert player.get_current() is None
|
|
assert player.get_queue() == []
|
|
assert player.get_queue_length() == 0
|
|
assert player.has_previous() is False
|
|
|
|
def test_volume_default(self, player):
|
|
"""Default volume should be 50."""
|
|
assert player.volume() == 50
|
|
|
|
def test_set_volume(self, player):
|
|
"""set_volume changes the volume."""
|
|
player.set_volume(75)
|
|
assert player.volume() == 75
|
|
player.set_volume(100)
|
|
assert player.volume() == 100
|
|
player.set_volume(0)
|
|
assert player.volume() == 0
|
|
|
|
def test_set_volume_clamps(self, player):
|
|
"""set_volume clamps values outside 0-100."""
|
|
player.set_volume(-10)
|
|
assert player.volume() == 0
|
|
player.set_volume(150)
|
|
assert player.volume() == 100
|
|
|
|
def test_loop_default(self, player):
|
|
"""Loop mode is initially off."""
|
|
assert player.get_loop() is False
|
|
|
|
def test_set_loop(self, player):
|
|
"""set_loop toggles loop mode."""
|
|
player.set_loop(True)
|
|
assert player.get_loop() is True
|
|
player.set_loop(False)
|
|
assert player.get_loop() is False
|
|
|
|
def test_stop_playback_when_stopped(self, player):
|
|
"""stop_playback when already stopped doesn't crash."""
|
|
player.stop_playback()
|
|
assert player.is_stopped() is True
|
|
|
|
def test_queue_operations(self, player):
|
|
"""queue_song, queue_next, and queue_list work correctly.
|
|
|
|
``queue_song`` triggers auto-play when queue is empty, which pops
|
|
the song immediately. We test queue_list (no auto-play) instead.
|
|
"""
|
|
s2 = {"videoId": "b", "title": "B"}
|
|
s3 = {"videoId": "c", "title": "C"}
|
|
|
|
# queue_list adds without triggering auto-play
|
|
player.queue_list([s2, s3])
|
|
assert player.get_queue_length() == 2
|
|
assert player.get_queue()[0]["videoId"] == "b"
|
|
|
|
# queue_next inserts at position 0
|
|
s1 = {"videoId": "a", "title": "A"}
|
|
player.queue_next(s1)
|
|
assert player.get_queue_length() == 3
|
|
assert player.get_queue()[0]["videoId"] == "a"
|
|
|
|
def test_seek(self, player):
|
|
"""seek sets position (no crash test)."""
|
|
player.seek(0) # should not crash
|
|
player.seek(5000)
|
|
|
|
def test_shutdown(self, player):
|
|
"""shutdown cleans up without error."""
|
|
player.shutdown() # already called in fixture cleanup
|
|
|
|
def test_cleanup_current_temp_no_current(self, player):
|
|
"""_cleanup_current_temp does nothing when no song is loaded."""
|
|
player._cleanup_current_temp() # should not crash
|
|
|
|
def test_cleanup_temp_files_no_files(self, player):
|
|
"""_cleanup_temp_files handles missing files gracefully."""
|
|
player._current = {"_local_path": "/nonexistent/file.mp3"}
|
|
player._cleanup_temp_files() # should not crash
|
|
player._current = None
|
|
|
|
def test_cleanup_temp_files_with_cache(self, player, tmp_path):
|
|
"""_cleanup_temp_files removes cached temp files."""
|
|
f = tmp_path / "test_song.mp3"
|
|
f.write_text("audio data")
|
|
player._prefetch_cache["vid1"] = {"_local_path": str(f)}
|
|
player._cleanup_temp_files()
|
|
assert not f.exists()
|
|
|
|
def test_position_and_duration_signals(self, player):
|
|
"""_on_position and _on_duration emit signals."""
|
|
results = []
|
|
player.position_changed.connect(lambda ms: results.append(("pos", ms)))
|
|
player.duration_changed.connect(lambda ms: results.append(("dur", ms)))
|
|
|
|
player._on_position(5000)
|
|
player._on_duration(200000)
|
|
|
|
assert ("pos", 5000) in results
|
|
assert ("dur", 200000) in results
|
|
|
|
def test_is_loaded(self, player):
|
|
"""is_loaded returns False when no current song."""
|
|
assert player.is_loaded() is False
|
|
player._current = {"videoId": "x"}
|
|
assert player.is_loaded() is True
|
|
player._current = None
|
|
|
|
def test_toggle_pause_when_stopped(self, player):
|
|
"""toggle_pause when stopped does nothing."""
|
|
result = player.toggle_pause()
|
|
# When stopped, toggle_pause returns None (no state change)
|
|
assert result is None or result is False
|
|
|
|
def test_previous_no_history(self, player):
|
|
"""previous does nothing when history is empty."""
|
|
player.previous() # should not crash
|
|
assert player.has_previous() is False
|
|
|
|
def test_cancel_active_downloads(self, player):
|
|
"""_cancel_active_downloads emits cancel signal when active."""
|
|
player._active_task_id = 42
|
|
with mock.patch.object(player._worker, "cancel_requested") as mock_sig:
|
|
player._cancel_active_downloads()
|
|
mock_sig.emit.assert_called_once()
|
|
|
|
def test_cancel_active_downloads_noop(self, player):
|
|
"""_cancel_active_downloads does nothing when no active task."""
|
|
with mock.patch.object(player._worker, "cancel_requested") as mock_sig:
|
|
player._cancel_active_downloads()
|
|
mock_sig.emit.assert_not_called()
|
|
|
|
def test_on_download_failed_matching_task(self, player):
|
|
"""_on_download_failed emits error when task matches active."""
|
|
results = []
|
|
player.error.connect(lambda d: results.append(d))
|
|
player._active_task_id = 42
|
|
player._on_download_failed(42)
|
|
assert len(results) == 1
|
|
assert "error" in results[0]
|
|
assert player._active_task_id is None
|
|
|
|
def test_on_download_failed_stale_task(self, player):
|
|
"""_on_download_failed ignores stale task IDs."""
|
|
results = []
|
|
player.error.connect(lambda d: results.append(d))
|
|
player._active_task_id = 42
|
|
player._on_download_failed(99) # different task ID
|
|
assert results == [] # no signal emitted
|
|
assert player._active_task_id == 42 # unchanged
|
|
|
|
def test_skip_no_current(self, player):
|
|
"""skip does nothing when no current song."""
|
|
player.skip() # should not crash
|
|
|
|
def test_on_download_succeeded_stale(self, player):
|
|
"""_on_download_succeeded ignores stale task IDs."""
|
|
player._active_task_id = 42
|
|
player._on_download_succeeded(99, {"videoId": "x"}) # different task
|
|
# No crash, no state change
|
|
|
|
def test_on_download_succeeded_missing_file(self, player, tmp_path):
|
|
"""_on_download_succeeded emits error when file is missing."""
|
|
results = []
|
|
player.error.connect(lambda d: results.append(d))
|
|
player._active_task_id = 42
|
|
player._on_download_succeeded(42, {"videoId": "x",
|
|
"_local_path": str(tmp_path / "nonexistent.mp3")})
|
|
assert len(results) == 1
|
|
assert player._active_task_id is None
|
|
|
|
def test_on_playback_state_playing(self, player):
|
|
"""_on_playback_state with PlayingState emits 'playing'."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
results = []
|
|
player.playback_state_changed.connect(lambda s: results.append(s))
|
|
player._current = {"videoId": "x", "title": "Test"}
|
|
player._on_playback_state(QMediaPlayer.PlaybackState.PlayingState)
|
|
assert "playing" in results
|
|
player._current = None
|
|
|
|
def test_on_playback_state_paused(self, player):
|
|
"""_on_playback_state with PausedState emits 'paused'."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
results = []
|
|
player.playback_state_changed.connect(lambda s: results.append(s))
|
|
player._current = {"videoId": "x"}
|
|
player._on_playback_state(QMediaPlayer.PlaybackState.PausedState)
|
|
assert "paused" in results
|
|
player._current = None
|
|
|
|
def test_on_playback_state_stopped(self, player):
|
|
"""_on_playback_state with StoppedState emits 'stopped'."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
results = []
|
|
player.playback_state_changed.connect(lambda s: results.append(s))
|
|
player._on_playback_state(QMediaPlayer.PlaybackState.StoppedState)
|
|
assert "stopped" in results
|
|
|
|
def test_on_media_status_end_of_media(self, player):
|
|
"""_on_media_status with EndOfMedia triggers advance."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
results = []
|
|
player.song_ended.connect(lambda vid: results.append(vid))
|
|
player._current = {"videoId": "x"}
|
|
player._on_media_status(QMediaPlayer.MediaStatus.EndOfMedia)
|
|
assert "x" in results
|
|
player._current = None
|
|
|
|
def test_on_media_status_invalid_media_stopped(self, player):
|
|
"""_on_media_status with InvalidMedia does nothing when player is stopped."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
results = []
|
|
player.error.connect(lambda d: results.append(d))
|
|
player._current = {"videoId": "x"}
|
|
# When the player is stopped (no media loaded), InvalidMedia is ignored
|
|
player._on_media_status(QMediaPlayer.MediaStatus.InvalidMedia)
|
|
assert len(results) == 0 # player is stopped, so early return
|
|
player._current = None
|
|
|
|
def test_on_media_status_loaded(self, player):
|
|
"""_on_media_status with LoadedMedia does nothing special."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
player._on_media_status(QMediaPlayer.MediaStatus.LoadedMedia) # no crash
|
|
|
|
def test_get_position(self, player):
|
|
"""get_position returns current position (0 when stopped)."""
|
|
assert player.get_position() == 0
|
|
|
|
def test_start_prefetch(self, player):
|
|
"""_start_prefetch with queue emits download request."""
|
|
from player import DownloadWorker
|
|
worker = DownloadWorker()
|
|
player._worker = worker
|
|
player._queue = [{"videoId": "prefetch_test"}]
|
|
with mock.patch.object(worker, "download_requested") as mock_sig:
|
|
player._start_prefetch()
|
|
mock_sig.emit.assert_called_once()
|
|
task_id, song = mock_sig.emit.call_args[0]
|
|
assert isinstance(task_id, int)
|
|
assert song["videoId"] == "prefetch_test"
|
|
|
|
def test_on_download_succeeded_prefetch(self, player, tmp_path):
|
|
"""_on_download_succeeded caches prefetch result."""
|
|
temp_file = tmp_path / "test.mp3"
|
|
temp_file.write_text("audio")
|
|
player._prefetch_task_id = 55
|
|
player._queue = [{"videoId": "prefetch_vid"}]
|
|
player._on_download_succeeded(55, {"videoId": "prefetch_vid",
|
|
"_local_path": str(temp_file)})
|
|
assert "prefetch_vid" in player._prefetch_cache
|
|
assert player._prefetch_task_id is None
|
|
|
|
def test_advance_with_loop(self, player):
|
|
"""_advance re-queues finished song at front in loop mode."""
|
|
player._loop_mode = True
|
|
player._current = {"videoId": "loop_song"}
|
|
player._queue = [{"videoId": "next_song"}]
|
|
# _advance inserts at front, then _try_next_song pops the front
|
|
# leaving the next_song. Verify loop_song was inserted before pop.
|
|
with mock.patch.object(player, "_try_next_song"):
|
|
player._advance()
|
|
# The finished song should be at the front before _try_next_song
|
|
assert player._queue[0]["videoId"] == "loop_song"
|
|
assert player._current is None
|
|
|
|
def test_try_next_song_with_queue(self, player):
|
|
"""_try_next_song pops and plays when queue has items."""
|
|
player._queue = [{"videoId": "queued"}]
|
|
# This will call _pop_and_play which tries to download
|
|
# Just check it doesn't crash and processes the queue
|
|
with mock.patch.object(player, "_pop_and_play") as mock_pop:
|
|
player._try_next_song()
|
|
mock_pop.assert_called_once()
|
|
|
|
def test_on_player_error_network(self, player):
|
|
"""_on_player_error_occurred with NetworkError triggers advance."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
results = []
|
|
player.error.connect(lambda d: results.append(d))
|
|
player._current = {"videoId": "err_song"}
|
|
player._on_player_error_occurred(
|
|
QMediaPlayer.Error.NetworkError, "Connection lost")
|
|
assert len(results) == 1
|
|
assert "error" in results[0]
|
|
player._current = None
|
|
|
|
def test_on_player_error_format(self, player):
|
|
"""_on_player_error_occurred with FormatError does nothing."""
|
|
from PySide6.QtMultimedia import QMediaPlayer
|
|
results = []
|
|
player.error.connect(lambda d: results.append(d))
|
|
player._on_player_error_occurred(
|
|
QMediaPlayer.Error.FormatError, "Bad format")
|
|
# FormatError is not NetworkError or ResourceError, so ignored
|
|
assert len(results) == 0
|
|
|
|
def test_seek_emits_seeked_signal(self, player):
|
|
"""seek emits seeked signal with the position."""
|
|
results = []
|
|
player.seeked.connect(lambda ms: results.append(ms))
|
|
player.seek(15000)
|
|
assert 15000 in results
|
|
|
|
def test_download_worker_do_cancel(self, player):
|
|
"""DownloadWorker._do_cancel sets cancelled flag."""
|
|
from player import DownloadWorker
|
|
worker = DownloadWorker()
|
|
assert worker._cancelled is False
|
|
worker._do_cancel()
|
|
assert worker._cancelled is True
|
|
|
|
def test_on_position_zero(self, player):
|
|
"""_on_position with 0 emits position_changed."""
|
|
results = []
|
|
player.position_changed.connect(lambda ms: results.append(ms))
|
|
player._on_position(0)
|
|
assert 0 in results
|
|
|
|
|
|
class TestTunettiWindowAccessors:
|
|
|
|
@pytest.fixture
|
|
def window(self, app):
|
|
"""Create and properly clean up a TunettiWindow."""
|
|
from gui import TunettiWindow
|
|
win = TunettiWindow()
|
|
yield win
|
|
win.player.shutdown()
|
|
win.rpc.stop()
|
|
win.db.close()
|
|
win.close()
|
|
_cleanup(win, app)
|
|
|
|
def test_get_visualizer(self, window):
|
|
"""get_visualizer returns the AudioVisualizer instance."""
|
|
from gui import AudioVisualizer
|
|
assert isinstance(window.get_visualizer(), AudioVisualizer)
|
|
|
|
def test_get_rpc(self, window):
|
|
"""get_rpc returns the DiscordRPC instance."""
|
|
from discord_rpc import DiscordRPC
|
|
assert isinstance(window.get_rpc(), DiscordRPC)
|
|
|
|
def test_get_player(self, window):
|
|
"""get_player returns the AudioPlayer instance."""
|
|
from player import AudioPlayer
|
|
assert isinstance(window.get_player(), AudioPlayer)
|