NikkeDoy
37d4a7d47e
All checks were successful
SonarQube Code Quality Scan / SonarQube Scan (push) Successful in 1m46s
Replace SearchWorker signal .connect() with mock.MagicMock to handle the case where the player module Signal binding is a plain function (import fallback when PySide6 is unavailable in headless CI). The MagicMock captures .emit() calls and works across all environments: real Qt, conftest stubs, and player.py fallback paths.
401 lines
14 KiB
Python
401 lines
14 KiB
Python
"""Tests for helper functions in the player module (player.py)."""
|
|
|
|
import os
|
|
import struct
|
|
import math
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
|
|
# ── _best_thumbnail ───────────────────────────────────────────────────────────
|
|
|
|
class TestBestThumbnail:
|
|
"""Tests for the _best_thumbnail helper."""
|
|
|
|
def _import(self):
|
|
from player import _best_thumbnail
|
|
return _best_thumbnail
|
|
|
|
def test_none_or_empty(self):
|
|
fn = self._import()
|
|
assert fn(None) == ""
|
|
assert fn([]) == ""
|
|
|
|
def test_string_returned_as_is(self):
|
|
fn = self._import()
|
|
assert fn("https://example.com/thumb.jpg") == "https://example.com/thumb.jpg"
|
|
|
|
def test_list_returns_last_url(self):
|
|
fn = self._import()
|
|
thumbs = [
|
|
{"url": "https://example.com/small.jpg"},
|
|
{"url": "https://example.com/medium.jpg"},
|
|
{"url": "https://example.com/large.jpg"},
|
|
]
|
|
assert fn(thumbs) == "https://example.com/large.jpg"
|
|
|
|
def test_list_without_url_keys(self):
|
|
fn = self._import()
|
|
thumbs = [{"width": 100}, {"width": 200}]
|
|
assert fn(thumbs) == ""
|
|
|
|
def test_list_with_non_dict_items(self):
|
|
fn = self._import()
|
|
assert fn(["string_thumb"]) == "string_thumb"
|
|
|
|
def test_non_list_non_string(self):
|
|
fn = self._import()
|
|
assert fn(123) == "123"
|
|
|
|
|
|
# ── _build_song_dict ─────────────────────────────────────────────────────────
|
|
|
|
class TestBuildSongDict:
|
|
"""Tests for the _build_song_dict helper."""
|
|
|
|
def _import(self):
|
|
from player import _build_song_dict
|
|
return _build_song_dict
|
|
|
|
def test_minimal_result(self):
|
|
fn = self._import()
|
|
result = {"videoId": "abc123", "title": "Test Song"}
|
|
song = fn(result)
|
|
assert song["videoId"] == "abc123"
|
|
assert song["title"] == "Test Song"
|
|
assert song["artists"] == []
|
|
assert song["album"] is None
|
|
assert song["duration"] == 0
|
|
assert song["duration_label"] == "?"
|
|
assert song["thumbnail"] == ""
|
|
assert song["videoType"] == ""
|
|
assert song["resultType"] == "song"
|
|
|
|
def test_full_result(self):
|
|
fn = self._import()
|
|
result = {
|
|
"videoId": "xyz789",
|
|
"title": "Full Song",
|
|
"artists": [{"name": "Artist", "id": "a1"}],
|
|
"album": {"name": "Album", "id": "al1"},
|
|
"duration_seconds": 300,
|
|
"duration": "5:00",
|
|
"thumbnails": [{"url": "https://ex.com/t.jpg"}],
|
|
"videoType": "MUSIC_VIDEO_TYPE_ATV",
|
|
"resultType": "video",
|
|
}
|
|
song = fn(result)
|
|
assert song["videoId"] == "xyz789"
|
|
assert song["title"] == "Full Song"
|
|
assert song["artists"] == [{"name": "Artist", "id": "a1"}]
|
|
assert song["album"] == {"name": "Album", "id": "al1"}
|
|
assert song["duration"] == 300
|
|
assert song["duration_label"] == "5:00"
|
|
assert song["thumbnail"] == "https://ex.com/t.jpg"
|
|
assert song["videoType"] == "MUSIC_VIDEO_TYPE_ATV"
|
|
assert song["resultType"] == "video"
|
|
|
|
def test_missing_video_id_defaults_to_empty(self):
|
|
fn = self._import()
|
|
song = fn({})
|
|
assert song["videoId"] == ""
|
|
assert song["title"] == "Unknown"
|
|
assert song["duration"] == 0
|
|
|
|
def test_thumbnail_from_thumbnails_list(self):
|
|
fn = self._import()
|
|
result = {
|
|
"thumbnails": [
|
|
{"url": "https://ex.com/small.jpg"},
|
|
{"url": "https://ex.com/large.jpg"},
|
|
]
|
|
}
|
|
song = fn(result)
|
|
assert song["thumbnail"] == "https://ex.com/large.jpg"
|
|
|
|
|
|
# ── _generate_test_wav ───────────────────────────────────────────────────────
|
|
|
|
class TestGenerateTestWav:
|
|
"""Tests for the _generate_test_wav helper."""
|
|
|
|
def _import(self):
|
|
from player import _generate_test_wav
|
|
return _generate_test_wav
|
|
|
|
def test_generates_valid_wav_file(self, tmp_path):
|
|
"""Generated WAV has valid RIFF header and correct size."""
|
|
fn = self._import()
|
|
wav_path = str(tmp_path / "test.wav")
|
|
fn(wav_path, duration_s=1)
|
|
|
|
with open(wav_path, "rb") as f:
|
|
data = f.read()
|
|
|
|
# RIFF header
|
|
assert data[:4] == b"RIFF"
|
|
assert data[8:12] == b"WAVE"
|
|
assert data[12:16] == b"fmt "
|
|
|
|
# PCM format
|
|
(audio_format, num_channels, sample_rate, byte_rate,
|
|
block_align, bits_per_sample) = struct.unpack_from("<HHIIHH", data, 20)
|
|
assert audio_format == 1 # PCM
|
|
assert num_channels == 1 # mono
|
|
assert sample_rate == 44100
|
|
assert bits_per_sample == 16
|
|
|
|
# data chunk: "data" marker at offset 36, size at offset 40
|
|
data_size = struct.unpack_from("<I", data, 40)[0]
|
|
expected_size = 44100 * 2 # 1 sec * 2 bytes per sample
|
|
assert data_size == expected_size
|
|
|
|
def test_generates_sine_wave_content(self, tmp_path):
|
|
"""Generated WAV contains a recognizable 440 Hz sine wave."""
|
|
fn = self._import()
|
|
wav_path = str(tmp_path / "test.wav")
|
|
fn(wav_path, duration_s=1)
|
|
|
|
with open(wav_path, "rb") as f:
|
|
f.seek(44)
|
|
samples = f.read()
|
|
|
|
# Decode first 100 samples
|
|
first_samples = []
|
|
for i in range(100):
|
|
sample = struct.unpack_from("<h", samples, i * 2)[0]
|
|
first_samples.append(sample)
|
|
|
|
# The samples should alternate (sine wave), not be all zero
|
|
assert any(s != 0 for s in first_samples), "Sine wave should produce non-zero samples"
|
|
# First sample should be 0 (sine starts at 0)
|
|
assert abs(first_samples[0]) < 3276, "Sine wave should start near 0"
|
|
|
|
|
|
# ── _stale_files ─────────────────────────────────────────────────────────────
|
|
|
|
class TestStaleFiles:
|
|
"""Tests for the _stale_files helper."""
|
|
|
|
def _import_fixtures(self):
|
|
from player import _tmp_dir, _stale_files
|
|
return _tmp_dir, _stale_files
|
|
|
|
def test_returns_empty_when_no_files(self, monkeypatch):
|
|
"""No files in tmp dir → empty list."""
|
|
_tmp_dir, _stale_files = self._import_fixtures()
|
|
monkeypatch.setattr("player._TUNETTI_TMP", tempfile.mkdtemp(prefix="tunetti_test_"))
|
|
result = _stale_files("nonexistent_video")
|
|
assert result == []
|
|
|
|
def test_returns_matching_files_ordered_by_mtime(self, monkeypatch):
|
|
"""Matching files are returned newest first."""
|
|
_tmp_dir, _stale_files = self._import_fixtures()
|
|
import time
|
|
|
|
td = tempfile.mkdtemp(prefix="tunetti_test_")
|
|
monkeypatch.setattr("player._TUNETTI_TMP", td)
|
|
|
|
# Create two files with different mtimes
|
|
old_file = os.path.join(td, "vid123.mp3")
|
|
new_file = os.path.join(td, "vid123.m4a")
|
|
|
|
with open(old_file, "w") as f:
|
|
f.write("data")
|
|
old_mtime = time.time() - 10
|
|
os.utime(old_file, (old_mtime, old_mtime))
|
|
|
|
with open(new_file, "w") as f:
|
|
f.write("data")
|
|
|
|
result = _stale_files("vid123")
|
|
assert len(result) == 2
|
|
# Newest first
|
|
assert os.path.basename(result[0]) == "vid123.m4a"
|
|
assert os.path.basename(result[1]) == "vid123.mp3"
|
|
|
|
def test_ignores_temp_files(self, monkeypatch):
|
|
"""Files with '.temp.' in name are excluded."""
|
|
_tmp_dir, _stale_files = self._import_fixtures()
|
|
|
|
td = tempfile.mkdtemp(prefix="tunetti_test_")
|
|
monkeypatch.setattr("player._TUNETTI_TMP", td)
|
|
|
|
# A "temp" download in progress and a completed file
|
|
with open(os.path.join(td, "vid123.temp.xyz.part"), "w") as f:
|
|
f.write("incomplete")
|
|
with open(os.path.join(td, "vid123.mp3"), "w") as f:
|
|
f.write("complete")
|
|
|
|
result = _stale_files("vid123")
|
|
assert len(result) == 1
|
|
assert "vid123.mp3" in result[0]
|
|
|
|
def test_ignores_empty_files(self, monkeypatch):
|
|
"""Zero-byte files are excluded."""
|
|
_tmp_dir, _stale_files = self._import_fixtures()
|
|
|
|
td = tempfile.mkdtemp(prefix="tunetti_test_")
|
|
monkeypatch.setattr("player._TUNETTI_TMP", td)
|
|
|
|
# Create an empty file to verify it's excluded
|
|
empty_path = os.path.join(td, "vid123.mp3")
|
|
Path(empty_path).touch()
|
|
|
|
result = _stale_files("vid123")
|
|
assert result == []
|
|
|
|
|
|
# ── _build_resolved_dict ──────────────────────────────────────────────────────
|
|
|
|
class TestBuildResolvedDict:
|
|
"""Tests for DownloadWorker._build_resolved_dict."""
|
|
|
|
def _import(self):
|
|
from player import DownloadWorker
|
|
return DownloadWorker._build_resolved_dict
|
|
|
|
def test_builds_resolved_dict(self):
|
|
fn = self._import()
|
|
song = {"videoId": "abc", "title": "Original"}
|
|
info = {"duration": 120, "thumbnail": "https://ex.com/t.jpg", "title": "Actual Title"}
|
|
temp_path = "/tmp/tunetti_abc.mp3"
|
|
resolved = fn(song, info, temp_path)
|
|
assert resolved["_local_path"] == temp_path
|
|
assert resolved["duration"] == 120
|
|
assert resolved["thumbnail"] == "https://ex.com/t.jpg"
|
|
assert resolved["title"] == "Actual Title"
|
|
assert resolved["videoId"] == "abc"
|
|
|
|
def test_falls_back_to_song_metadata(self):
|
|
fn = self._import()
|
|
song = {"videoId": "abc", "duration": 99, "title": "Fallback"}
|
|
info = {} # empty info
|
|
resolved = fn(song, info, "/tmp/t.mp3")
|
|
assert resolved["duration"] == 99
|
|
assert resolved["title"] == "Fallback"
|
|
assert resolved["thumbnail"] == ""
|
|
|
|
|
|
# ── _cleanup_path ───────────────────────────────────────────────────────────────
|
|
|
|
class TestCleanupPath:
|
|
"""Tests for DownloadWorker._cleanup_path."""
|
|
|
|
def _import(self):
|
|
from player import DownloadWorker
|
|
return DownloadWorker._cleanup_path
|
|
|
|
def test_removes_existing_file(self, tmp_path):
|
|
"""_cleanup_path removes a file that exists."""
|
|
fn = self._import()
|
|
path = tmp_path / "to_delete.txt"
|
|
path.write_text("hello")
|
|
assert path.exists()
|
|
fn(str(path))
|
|
assert not path.exists()
|
|
|
|
def test_noop_for_missing_file(self):
|
|
"""_cleanup_path does nothing for non-existent paths."""
|
|
fn = self._import()
|
|
fn("/nonexistent/path/file.txt") # should not raise
|
|
|
|
def test_noop_for_none(self):
|
|
"""_cleanup_path does nothing when path is None."""
|
|
fn = self._import()
|
|
fn(None)
|
|
|
|
def test_oserror_handled_gracefully(self, tmp_path, monkeypatch):
|
|
"""_cleanup_path handles OSError gracefully."""
|
|
fn = self._import()
|
|
path = tmp_path / "locked.txt"
|
|
path.write_text("data")
|
|
|
|
def _fail_unlink(p):
|
|
raise OSError("Permission denied")
|
|
|
|
monkeypatch.setattr(os, "unlink", _fail_unlink)
|
|
fn(str(path)) # should not raise
|
|
|
|
|
|
# ── _tmp_dir ───────────────────────────────────────────────────────────────────
|
|
|
|
class TestTmpDir:
|
|
"""Tests for the _tmp_dir helper."""
|
|
|
|
def _import(self):
|
|
from player import _tmp_dir
|
|
return _tmp_dir
|
|
|
|
def test_returns_string(self):
|
|
"""_tmp_dir returns a string path."""
|
|
fn = self._import()
|
|
path = fn()
|
|
assert isinstance(path, str)
|
|
assert "tunetti" in path
|
|
|
|
def test_idempotent(self):
|
|
"""Calling _tmp_dir multiple times returns the same path."""
|
|
fn = self._import()
|
|
assert fn() == fn()
|
|
|
|
|
|
# ── SearchWorker (TEST_MODE) ─────────────────────────────────────────────────
|
|
|
|
class TestSearchWorkerTestMode:
|
|
"""Tests for SearchWorker.run() with TEST_MODE enabled.
|
|
|
|
Uses ``mock.MagicMock`` to capture signal emissions instead of
|
|
``.connect()`` because the player module's ``Signal`` binding may
|
|
be a plain function (the import fallback when PySide6 is unavailable)
|
|
that does not have a ``.connect()`` method.
|
|
"""
|
|
|
|
def test_returns_dummy_results_in_test_mode(self, monkeypatch):
|
|
"""With TEST_MODE=True, SearchWorker returns 5 dummy songs."""
|
|
monkeypatch.setattr("player.TEST_MODE", True)
|
|
from player import SearchWorker
|
|
|
|
worker = SearchWorker("test query")
|
|
# Use a MagicMock to capture signal emissions — works with both
|
|
# real Signal, stub _Signal, and fallback lambda.
|
|
mock_signal = mock.MagicMock()
|
|
worker.results_ready = mock_signal
|
|
worker.run()
|
|
|
|
mock_signal.emit.assert_called_once()
|
|
data = mock_signal.emit.call_args[0][0]
|
|
assert "songs" in data
|
|
assert "videos" in data
|
|
assert len(data["songs"]) == 5
|
|
assert data["videos"] == []
|
|
assert data["songs"][0]["title"] == "Test Song 0"
|
|
|
|
def test_dummy_results_have_required_keys(self, monkeypatch):
|
|
"""Each dummy song dict has all expected keys."""
|
|
monkeypatch.setattr("player.TEST_MODE", True)
|
|
from player import SearchWorker
|
|
|
|
worker = SearchWorker("test query")
|
|
mock_signal = mock.MagicMock()
|
|
worker.results_ready = mock_signal
|
|
worker.run()
|
|
|
|
mock_signal.emit.assert_called_once()
|
|
data = mock_signal.emit.call_args[0][0]
|
|
song = data["songs"][0]
|
|
assert "videoId" in song
|
|
assert "title" in song
|
|
assert "artists" in song
|
|
assert len(song["artists"]) == 1
|
|
assert song["artists"][0]["name"] == "Artist 0"
|
|
|
|
def test_sets_query_correctly(self):
|
|
"""SearchWorker stores the query."""
|
|
from player import SearchWorker
|
|
worker = SearchWorker("my search query")
|
|
assert worker._query == "my search query"
|