✅ | Add tests
Some checks failed
SonarQube Code Quality Scan / SonarQube Scan (push) Failing after 2m37s
Some checks failed
SonarQube Code Quality Scan / SonarQube Scan (push) Failing after 2m37s
This commit is contained in:
16
.coveragerc
Normal file
16
.coveragerc
Normal file
@@ -0,0 +1,16 @@
|
||||
[run]
|
||||
relative_files = True
|
||||
source = .
|
||||
omit =
|
||||
tests/*
|
||||
.venv/*
|
||||
docs/*
|
||||
.gitea/*
|
||||
__pycache__/*
|
||||
*.pyc
|
||||
shibokensupport/*
|
||||
PySide6/*
|
||||
shiboken2/*
|
||||
|
||||
[xml]
|
||||
output = coverage.xml
|
||||
@@ -18,6 +18,28 @@ jobs:
|
||||
with:
|
||||
fetch-depth: 0 # Required for advanced SonarQube features like blame info
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install system dependencies
|
||||
run: |
|
||||
sudo apt-get update -qq
|
||||
sudo apt-get install -y -qq libegl1 libxkbcommon0 libglib2.0-0t64 libgl1-mesa-glx || true
|
||||
|
||||
- name: Install project dependencies
|
||||
run: |
|
||||
python -m venv .venv
|
||||
.venv/bin/pip install --upgrade pip
|
||||
.venv/bin/pip install -r requirements.txt
|
||||
.venv/bin/pip install -r requirements-dev.txt
|
||||
|
||||
- name: Run tests with coverage
|
||||
run: |
|
||||
.venv/bin/coverage run -m pytest --junitxml=pytest-report.xml tests/
|
||||
.venv/bin/coverage xml
|
||||
|
||||
- name: Run SonarQube Scanner
|
||||
uses: sonarsource/sonarqube-scan-action@v4
|
||||
env:
|
||||
|
||||
1
.gitignore
vendored
1
.gitignore
vendored
@@ -46,6 +46,7 @@ htmlcov/
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
pytest-report.xml
|
||||
*.cover
|
||||
*.py,cover
|
||||
.hypothesis/
|
||||
|
||||
4
requirements-dev.txt
Normal file
4
requirements-dev.txt
Normal file
@@ -0,0 +1,4 @@
|
||||
# Test dependencies
|
||||
pytest>=7.0.0
|
||||
pytest-cov>=4.0.0
|
||||
coverage>=7.0.0
|
||||
@@ -2,3 +2,15 @@ sonar.projectKey=Tunetti
|
||||
sonar.projectName=Tunetti
|
||||
sonar.sources=.
|
||||
sonar.language=py
|
||||
sonar.sourceEncoding=UTF-8
|
||||
|
||||
# Test directories & coverage
|
||||
sonar.tests=tests
|
||||
sonar.python.coverage.reportPaths=coverage.xml
|
||||
sonar.python.xunit.reportPath=pytest-report.xml
|
||||
sonar.test.exclusions=tests/**
|
||||
|
||||
# Exclude generated / third-party code
|
||||
sonar.exclusions=.venv/**,docs/**,.gitea/**,__pycache__/**,*.pyc
|
||||
sonar.python.pylint.reportPaths=
|
||||
sonar.python.bandit.reportPaths=
|
||||
|
||||
1
tests/__init__.py
Normal file
1
tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Tests for Tunetti."""
|
||||
125
tests/test_config.py
Normal file
125
tests/test_config.py
Normal file
@@ -0,0 +1,125 @@
|
||||
"""Tests for the config module (config.py)."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# We import the module under test after patching CONFIG_DIR so that
|
||||
# the module-level SETTINGS dict is loaded from a temp location.
|
||||
@pytest.fixture(autouse=True)
|
||||
def _patch_config_paths(monkeypatch, tmp_path):
|
||||
"""Redirect CONFIG_DIR to a tmp_path so tests don't touch ~/.config."""
|
||||
config_dir = tmp_path / ".config" / "tunetti"
|
||||
monkeypatch.setattr("config.CONFIG_DIR", config_dir)
|
||||
monkeypatch.setattr("config.CONFIG_FILE", config_dir / "config.json")
|
||||
monkeypatch.setattr("config._LEGACY_FILE", tmp_path / "tunetti_config.json")
|
||||
|
||||
# Also reset SETTINGS and DEFAULT_CONFIG after patching
|
||||
import config as cfg
|
||||
cfg.SETTINGS = cfg.load_config()
|
||||
return cfg
|
||||
|
||||
|
||||
# ── Tests ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestLoadConfig:
|
||||
def test_first_launch_creates_defaults(self, _patch_config_paths):
|
||||
"""On first launch, no config exists → defaults are written to disk."""
|
||||
cfg = _patch_config_paths
|
||||
assert cfg.CONFIG_FILE.exists()
|
||||
assert cfg.SETTINGS["volume"] == 50
|
||||
assert cfg.SETTINGS["max_history"] == 5000
|
||||
assert cfg.SETTINGS["discord_rpc_enabled"] is True
|
||||
|
||||
def test_loads_existing_config(self, _patch_config_paths):
|
||||
"""Loading an existing config merges user values over defaults."""
|
||||
cfg = _patch_config_paths
|
||||
cfg.CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
user_config = {"volume": 75, "max_history": 100}
|
||||
cfg.CONFIG_FILE.write_text(json.dumps(user_config))
|
||||
|
||||
loaded = cfg.load_config()
|
||||
assert loaded["volume"] == 75
|
||||
assert loaded["max_history"] == 100
|
||||
# default keys still present
|
||||
assert loaded["discord_rpc_enabled"] is True
|
||||
|
||||
def test_corrupt_config_falls_back_to_defaults(self, _patch_config_paths):
|
||||
"""Corrupt JSON should be ignored and defaults returned."""
|
||||
cfg = _patch_config_paths
|
||||
cfg.CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
cfg.CONFIG_FILE.write_text("not valid json")
|
||||
|
||||
loaded = cfg.load_config()
|
||||
assert loaded["volume"] == 50 # default
|
||||
# The corrupt file is not overwritten, but defaults are returned
|
||||
|
||||
def test_legacy_migration(self, _patch_config_paths):
|
||||
"""Legacy config in project dir is migrated to XDG location."""
|
||||
cfg = _patch_config_paths
|
||||
# Remove the config that was created by the fixture so migration runs
|
||||
cfg.CONFIG_FILE.unlink(missing_ok=True)
|
||||
cfg.SETTINGS = dict(cfg.DEFAULT_CONFIG)
|
||||
|
||||
legacy = {**cfg.DEFAULT_CONFIG, "volume": 90}
|
||||
# Legacy file location is the parent of the "real" config dir (tmp_path)
|
||||
cfg._LEGACY_FILE.write_text(json.dumps(legacy))
|
||||
|
||||
loaded = cfg.load_config()
|
||||
assert loaded["volume"] == 90
|
||||
# Legacy file should be renamed
|
||||
assert not cfg._LEGACY_FILE.exists()
|
||||
assert cfg._LEGACY_FILE.with_suffix(".json.migrated").exists()
|
||||
|
||||
|
||||
class TestSaveConfig:
|
||||
def test_save_config_persists(self, _patch_config_paths):
|
||||
"""save_config writes merged config to disk."""
|
||||
cfg = _patch_config_paths
|
||||
cfg.save_config({"volume": 30, "visualizer_enabled": False})
|
||||
data = json.loads(cfg.CONFIG_FILE.read_text())
|
||||
assert data["volume"] == 30
|
||||
assert data["visualizer_enabled"] is False
|
||||
|
||||
def test_save_volume_clamps(self, _patch_config_paths):
|
||||
"""save_volume clamps values between 0 and 100."""
|
||||
cfg = _patch_config_paths
|
||||
cfg.save_volume(-10)
|
||||
assert cfg.SETTINGS["volume"] == 0
|
||||
cfg.save_volume(150)
|
||||
assert cfg.SETTINGS["volume"] == 100
|
||||
cfg.save_volume(42)
|
||||
assert cfg.SETTINGS["volume"] == 42
|
||||
|
||||
def test_save_setting_updates_memory_and_disk(self, _patch_config_paths):
|
||||
"""save_setting updates both in-memory SETTINGS and persisted config."""
|
||||
cfg = _patch_config_paths
|
||||
cfg.save_setting("max_history", 999)
|
||||
assert cfg.SETTINGS["max_history"] == 999
|
||||
data = json.loads(cfg.CONFIG_FILE.read_text())
|
||||
assert data["max_history"] == 999
|
||||
|
||||
|
||||
class TestGetSetting:
|
||||
def test_returns_set_value(self, _patch_config_paths):
|
||||
"""get_setting returns the current in-memory value."""
|
||||
cfg = _patch_config_paths
|
||||
cfg.SETTINGS["volume"] = 88
|
||||
assert cfg.get_setting("volume") == 88
|
||||
|
||||
def test_returns_default_for_missing(self, _patch_config_paths):
|
||||
"""get_setting falls back to DEFAULT_CONFIG for missing keys."""
|
||||
cfg = _patch_config_paths
|
||||
# Remove from SETTINGS
|
||||
cfg.SETTINGS.pop("volume", None)
|
||||
assert cfg.get_setting("volume") == cfg.DEFAULT_CONFIG["volume"]
|
||||
|
||||
def test_nonexistent_key_returns_none(self, _patch_config_paths):
|
||||
"""get_setting returns None for keys not in either dict."""
|
||||
cfg = _patch_config_paths
|
||||
assert cfg.get_setting("nonexistent_key") is None
|
||||
177
tests/test_discord_rpc.py
Normal file
177
tests/test_discord_rpc.py
Normal file
@@ -0,0 +1,177 @@
|
||||
"""Tests for Discord RPC module (discord_rpc.py)."""
|
||||
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ── _safe_artists ─────────────────────────────────────────────────────────────
|
||||
|
||||
class TestSafeArtists:
|
||||
"""Tests for the _safe_artists helper."""
|
||||
|
||||
def _import(self):
|
||||
from discord_rpc import _safe_artists
|
||||
return _safe_artists
|
||||
|
||||
def test_none_song(self):
|
||||
fn = self._import()
|
||||
assert fn(None) == ""
|
||||
|
||||
def test_empty_dict(self):
|
||||
fn = self._import()
|
||||
assert fn({}) == ""
|
||||
|
||||
def test_missing_artists_key(self):
|
||||
fn = self._import()
|
||||
assert fn({"title": "Test"}) == ""
|
||||
|
||||
def test_artists_not_list(self):
|
||||
fn = self._import()
|
||||
assert fn({"artists": "not a list"}) == ""
|
||||
|
||||
def test_empty_artists_list(self):
|
||||
fn = self._import()
|
||||
assert fn({"artists": []}) == ""
|
||||
|
||||
def test_single_artist(self):
|
||||
fn = self._import()
|
||||
song = {"artists": [{"name": "Test Artist", "id": "x"}]}
|
||||
assert fn(song) == "Test Artist"
|
||||
|
||||
def test_multiple_artists(self):
|
||||
fn = self._import()
|
||||
song = {
|
||||
"artists": [
|
||||
{"name": "Artist One", "id": "1"},
|
||||
{"name": "Artist Two", "id": "2"},
|
||||
]
|
||||
}
|
||||
assert fn(song) == "Artist One, Artist Two"
|
||||
|
||||
def test_artist_without_name(self):
|
||||
fn = self._import()
|
||||
song = {"artists": [{"id": "x"}, {"name": "Real Artist", "id": "y"}]}
|
||||
assert fn(song) == "Real Artist"
|
||||
|
||||
def test_artist_is_not_dict(self):
|
||||
fn = self._import()
|
||||
song = {"artists": ["just a string"]}
|
||||
assert fn(song) == ""
|
||||
|
||||
def test_artist_with_empty_name(self):
|
||||
fn = self._import()
|
||||
song = {"artists": [{"name": "", "id": "x"}]}
|
||||
assert fn(song) == ""
|
||||
|
||||
|
||||
# ── DiscordRPC ────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestDiscordRPC:
|
||||
"""Tests for DiscordRPC class with mocked pypresence."""
|
||||
|
||||
@pytest.fixture
|
||||
def rpc(self):
|
||||
with mock.patch("discord_rpc.Presence") as mock_presence:
|
||||
# Make the mock Presence instance return a mock
|
||||
mock_instance = mock.MagicMock()
|
||||
mock_presence.return_value = mock_instance
|
||||
|
||||
from discord_rpc import DiscordRPC
|
||||
rpc = DiscordRPC(client_id="test_client_123")
|
||||
yield rpc, mock_instance
|
||||
rpc.stop()
|
||||
|
||||
def test_constructor(self, rpc):
|
||||
drpc, _ = rpc
|
||||
assert drpc._client_id == "test_client_123"
|
||||
assert drpc._connected is False
|
||||
assert drpc._song is None
|
||||
|
||||
def test_start_and_connect(self, rpc):
|
||||
drpc, mock_presence = rpc
|
||||
drpc.start()
|
||||
# Give the thread a moment to connect
|
||||
import time
|
||||
time.sleep(0.1)
|
||||
assert drpc._connected is True
|
||||
mock_presence.connect.assert_called_once()
|
||||
|
||||
def test_update_song_sets_song_and_resets_start(self, rpc):
|
||||
drpc, _ = rpc
|
||||
song = {"title": "Test", "artists": [{"name": "A"}], "duration": 200}
|
||||
drpc.update_song(song, reset_start=True)
|
||||
assert drpc._song == song
|
||||
assert drpc._start_ts is not None
|
||||
|
||||
def test_update_song_none(self, rpc):
|
||||
drpc, _ = rpc
|
||||
drpc.update_song(None)
|
||||
assert drpc._song is None
|
||||
assert drpc._start_ts is None
|
||||
|
||||
def test_update_song_no_reset(self, rpc):
|
||||
drpc, _ = rpc
|
||||
song = {"title": "Test", "artists": [{"name": "A"}], "duration": 200}
|
||||
drpc.update_song(song, reset_start=True)
|
||||
old_start = drpc._start_ts
|
||||
time.sleep(0.01)
|
||||
# Without reset, start_ts should stay the same
|
||||
drpc.update_song(song, reset_start=False)
|
||||
assert drpc._start_ts == old_start
|
||||
|
||||
def test_clear_presence(self, rpc):
|
||||
drpc, _ = rpc
|
||||
drpc.update_song({"title": "Test"})
|
||||
assert drpc._song is not None
|
||||
drpc.clear()
|
||||
assert drpc._song is None
|
||||
assert drpc._start_ts is None
|
||||
|
||||
def test_seek_to_adjusts_start_ts(self, rpc):
|
||||
drpc, _ = rpc
|
||||
drpc.update_song({"title": "Test"}, reset_start=True)
|
||||
original_start = drpc._start_ts
|
||||
time.sleep(0.01)
|
||||
drpc.seek_to(position_ms=30_000) # seek to 30s
|
||||
# Now the diff between now and start_ts should be ~30s
|
||||
diff = int(time.time()) - drpc._start_ts
|
||||
assert 28 <= diff <= 32 # allow some fuzz
|
||||
|
||||
def test_stop_disconnects_and_joins(self, rpc):
|
||||
drpc, mock_presence = rpc
|
||||
drpc.start()
|
||||
time.sleep(0.1)
|
||||
drpc.stop()
|
||||
assert drpc._rpc is None
|
||||
assert drpc._connected is False
|
||||
|
||||
def test_send_presence_clears_when_no_song(self, rpc):
|
||||
drpc, mock_presence = rpc
|
||||
drpc.start()
|
||||
time.sleep(0.1)
|
||||
drpc.clear()
|
||||
time.sleep(0.1)
|
||||
mock_presence.clear.assert_called()
|
||||
|
||||
def test_double_start_is_idempotent(self, rpc):
|
||||
drpc, _ = rpc
|
||||
drpc.start()
|
||||
drpc.start() # should not crash
|
||||
assert True
|
||||
|
||||
@mock.patch("discord_rpc.Presence")
|
||||
def test_connect_failure_graceful(self, mock_presence_cls):
|
||||
"""When Discord is not running, connect failure is handled gracefully."""
|
||||
from discord_rpc import DiscordRPC
|
||||
mock_instance = mock.MagicMock()
|
||||
mock_instance.connect.side_effect = Exception("Discord not found")
|
||||
mock_presence_cls.return_value = mock_instance
|
||||
|
||||
drpc = DiscordRPC(client_id="test")
|
||||
drpc.start()
|
||||
time.sleep(0.1)
|
||||
# Should not crash, just log a warning
|
||||
assert drpc._connected is False
|
||||
drpc.stop()
|
||||
164
tests/test_gui_helpers.py
Normal file
164
tests/test_gui_helpers.py
Normal file
@@ -0,0 +1,164 @@
|
||||
"""Tests for helper functions in the GUI module (gui.py)."""
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ── _fmt_ms ───────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestFmtMs:
|
||||
"""Tests for the _fmt_ms helper."""
|
||||
|
||||
def _import(self):
|
||||
from gui import _fmt_ms
|
||||
return _fmt_ms
|
||||
|
||||
def test_zero(self):
|
||||
fn = self._import()
|
||||
assert fn(0) == "0:00"
|
||||
|
||||
def test_seconds_only(self):
|
||||
fn = self._import()
|
||||
assert fn(5_000) == "0:05"
|
||||
assert fn(59_000) == "0:59"
|
||||
|
||||
def test_minutes(self):
|
||||
fn = self._import()
|
||||
assert fn(60_000) == "1:00"
|
||||
assert fn(90_000) == "1:30"
|
||||
assert fn(3599_000) == "59:59"
|
||||
|
||||
def test_hours_display(self):
|
||||
fn = self._import()
|
||||
assert fn(3600_000) == "60:00"
|
||||
assert fn(3661_000) == "61:01"
|
||||
|
||||
def test_zero_ms(self):
|
||||
fn = self._import()
|
||||
assert fn(0) == "0:00"
|
||||
|
||||
def test_large_duration(self):
|
||||
fn = self._import()
|
||||
assert fn(5999_000) == "99:59"
|
||||
|
||||
|
||||
# ── _artists_str ─────────────────────────────────────────────────────────────
|
||||
|
||||
class TestArtistsStr:
|
||||
"""Tests for the _artists_str helper."""
|
||||
|
||||
def _import(self):
|
||||
from gui import _artists_str
|
||||
return _artists_str
|
||||
|
||||
def test_none(self):
|
||||
fn = self._import()
|
||||
assert fn(None) == ""
|
||||
|
||||
def test_empty_list(self):
|
||||
fn = self._import()
|
||||
assert fn([]) == ""
|
||||
|
||||
def test_not_a_list(self):
|
||||
fn = self._import()
|
||||
assert fn("str") == ""
|
||||
assert fn(42) == ""
|
||||
|
||||
def test_single_artist(self):
|
||||
fn = self._import()
|
||||
assert fn([{"name": "Artist One", "id": "a1"}]) == "Artist One"
|
||||
|
||||
def test_multiple_artists(self):
|
||||
fn = self._import()
|
||||
artists = [
|
||||
{"name": "Artist One", "id": "a1"},
|
||||
{"name": "Artist Two", "id": "a2"},
|
||||
]
|
||||
assert fn(artists) == "Artist One, Artist Two"
|
||||
|
||||
def test_artist_without_name_key(self):
|
||||
fn = self._import()
|
||||
assert fn([{"id": "a1"}]) == ""
|
||||
|
||||
def test_mixed_valid_and_invalid(self):
|
||||
fn = self._import()
|
||||
artists = [
|
||||
{"name": "Valid", "id": "v1"},
|
||||
{"id": "no_name"},
|
||||
{"name": "Also Valid", "id": "v2"},
|
||||
]
|
||||
assert fn(artists) == "Valid, Also Valid"
|
||||
|
||||
|
||||
# ── _norm_artists ────────────────────────────────────────────────────────────
|
||||
|
||||
class TestNormArtists:
|
||||
"""Tests for the _norm_artists helper."""
|
||||
|
||||
def _import(self):
|
||||
from gui import _norm_artists
|
||||
return _norm_artists
|
||||
|
||||
def test_none(self):
|
||||
fn = self._import()
|
||||
assert fn(None) == []
|
||||
|
||||
def test_not_a_list(self):
|
||||
fn = self._import()
|
||||
assert fn("not a list") == []
|
||||
assert fn(42) == []
|
||||
|
||||
def test_empty_list(self):
|
||||
fn = self._import()
|
||||
assert fn([]) == []
|
||||
|
||||
def test_filters_non_dicts(self):
|
||||
fn = self._import()
|
||||
artists = [{"name": "Real"}, "just a string", 123, {"name": "Another"}]
|
||||
result = fn(artists)
|
||||
assert len(result) == 2
|
||||
assert result[0]["name"] == "Real"
|
||||
assert result[1]["name"] == "Another"
|
||||
|
||||
def test_all_valid(self):
|
||||
fn = self._import()
|
||||
artists = [{"name": "A"}, {"name": "B"}]
|
||||
assert fn(artists) == artists
|
||||
|
||||
|
||||
# ── _artists_from_json ───────────────────────────────────────────────────────
|
||||
|
||||
class TestArtistsFromJson:
|
||||
"""Tests for the _artists_from_json helper."""
|
||||
|
||||
def _import(self):
|
||||
from gui import _artists_from_json
|
||||
return _artists_from_json
|
||||
|
||||
def test_valid_json_array(self):
|
||||
fn = self._import()
|
||||
j = json.dumps([{"name": "Artist A", "id": "a1"}, {"name": "Artist B", "id": "a2"}])
|
||||
assert fn(j) == "Artist A, Artist B"
|
||||
|
||||
def test_single_artist_json(self):
|
||||
fn = self._import()
|
||||
j = json.dumps([{"name": "Solo Artist", "id": "s1"}])
|
||||
assert fn(j) == "Solo Artist"
|
||||
|
||||
def test_invalid_json(self):
|
||||
fn = self._import()
|
||||
assert fn("not json") == "not json"
|
||||
|
||||
def test_empty_array(self):
|
||||
fn = self._import()
|
||||
assert fn("[]") == ""
|
||||
|
||||
def test_plain_string_input(self):
|
||||
fn = self._import()
|
||||
assert fn("Just a name") == "Just a name"
|
||||
|
||||
def test_jagged_array_skips_non_dicts(self):
|
||||
fn = self._import()
|
||||
j = json.dumps([{"name": "Real"}, 42, "str", {"name": "Also Real"}])
|
||||
assert fn(j) == "Real, Also Real"
|
||||
246
tests/test_music_db.py
Normal file
246
tests/test_music_db.py
Normal file
@@ -0,0 +1,246 @@
|
||||
"""Tests for the music database module (music_db.py)."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
# We'll create a fresh MusicDB for each test using a temporary file.
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def music_db():
|
||||
"""Yield a MusicDB backed by a temporary SQLite file."""
|
||||
from music_db import MusicDB
|
||||
|
||||
tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
||||
tmp.close()
|
||||
db = MusicDB(tmp.name)
|
||||
yield db
|
||||
db.close()
|
||||
os.unlink(tmp.name)
|
||||
|
||||
|
||||
# ── Helper ────────────────────────────────────────────────────────────────────
|
||||
|
||||
_SONG = dict(
|
||||
video_id="test_video_001",
|
||||
title="Test Song",
|
||||
artists=[{"name": "Test Artist", "id": "artist_001"}],
|
||||
album={"name": "Test Album", "id": "album_001"},
|
||||
duration=240,
|
||||
thumbnail="https://example.com/thumb.jpg",
|
||||
)
|
||||
|
||||
|
||||
def _record_play(db, **overrides):
|
||||
kwargs = dict(_SONG)
|
||||
kwargs.update(overrides)
|
||||
db.record_play(**kwargs)
|
||||
|
||||
|
||||
# ── Tests ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestRecordPlay:
|
||||
def test_records_play_and_history(self, music_db):
|
||||
"""record_play creates a song entry and a history row."""
|
||||
_record_play(music_db)
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 1
|
||||
assert stats["total_plays"] == 1
|
||||
|
||||
history = music_db.get_history(limit=10)
|
||||
assert len(history) == 1
|
||||
assert history[0]["video_id"] == _SONG["video_id"]
|
||||
|
||||
def test_increments_play_count(self, music_db):
|
||||
"""Playing the same song multiple times increments play_count."""
|
||||
_record_play(music_db)
|
||||
_record_play(music_db)
|
||||
_record_play(music_db)
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 1
|
||||
assert stats["total_plays"] == 3
|
||||
|
||||
def test_updates_last_played(self, music_db):
|
||||
"""Re-playing a song updates last_played timestamp."""
|
||||
_record_play(music_db)
|
||||
import time
|
||||
time.sleep(0.01)
|
||||
_record_play(music_db)
|
||||
history = music_db.get_history(limit=10)
|
||||
# The latest play should be the one returned
|
||||
assert history[0]["video_id"] == _SONG["video_id"]
|
||||
|
||||
def test_without_album(self, music_db):
|
||||
"""record_play works without album data."""
|
||||
_record_play(music_db, album=None)
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 1
|
||||
|
||||
def test_without_thumbnail(self, music_db):
|
||||
"""record_play works without thumbnail."""
|
||||
_record_play(music_db, thumbnail=None)
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 1
|
||||
|
||||
|
||||
class TestFavourites:
|
||||
def test_toggle_favourite(self, music_db):
|
||||
"""toggle_favourite flips the favourite flag."""
|
||||
_record_play(music_db)
|
||||
# Start unfavoured
|
||||
assert music_db.is_favourite(_SONG["video_id"]) is False
|
||||
# Toggle on
|
||||
assert music_db.toggle_favourite(_SONG["video_id"]) is True
|
||||
assert music_db.is_favourite(_SONG["video_id"]) is True
|
||||
# Toggle off
|
||||
assert music_db.toggle_favourite(_SONG["video_id"]) is False
|
||||
assert music_db.is_favourite(_SONG["video_id"]) is False
|
||||
|
||||
def test_toggle_favourite_nonexistent_song(self, music_db):
|
||||
"""toggle_favourite on a non-existent song returns False."""
|
||||
assert music_db.toggle_favourite("nonexistent") is False
|
||||
|
||||
def test_set_favourite(self, music_db):
|
||||
"""set_favourite explicitly sets the favourite state."""
|
||||
_record_play(music_db)
|
||||
vid = _SONG["video_id"]
|
||||
assert music_db.set_favourite(vid, True) is True
|
||||
assert music_db.is_favourite(vid) is True
|
||||
assert music_db.set_favourite(vid, False) is True
|
||||
assert music_db.is_favourite(vid) is False
|
||||
|
||||
def test_set_favourite_nonexistent(self, music_db):
|
||||
"""set_favourite on a non-existent song returns False."""
|
||||
assert music_db.set_favourite("nonexistent", True) is False
|
||||
|
||||
def test_get_favourites_returns_favourited_songs(self, music_db):
|
||||
"""get_favourites returns only favourited songs."""
|
||||
# Record two songs, favourite one
|
||||
_record_play(music_db, video_id="song_a", title="A")
|
||||
_record_play(music_db, video_id="song_b", title="B")
|
||||
music_db.toggle_favourite("song_a")
|
||||
favs = music_db.get_favourites()
|
||||
assert len(favs) == 1
|
||||
assert favs[0]["video_id"] == "song_a"
|
||||
|
||||
|
||||
class TestHistory:
|
||||
def test_history_ordered_by_recent(self, music_db):
|
||||
"""get_history returns songs ordered by most recent play."""
|
||||
_record_play(music_db, video_id="song_a", title="A")
|
||||
import time
|
||||
time.sleep(0.01)
|
||||
_record_play(music_db, video_id="song_b", title="B")
|
||||
history = music_db.get_history(limit=10)
|
||||
# Most recent (song_b) should be first
|
||||
assert history[0]["video_id"] == "song_b"
|
||||
assert history[1]["video_id"] == "song_a"
|
||||
|
||||
def test_history_deduplicates_by_video_id(self, music_db):
|
||||
"""get_history shows each song at most once (most recent play)."""
|
||||
_record_play(music_db, video_id="song_a", title="A")
|
||||
_record_play(music_db, video_id="song_b", title="B")
|
||||
_record_play(music_db, video_id="song_a", title="A")
|
||||
history = music_db.get_history(limit=10)
|
||||
# song_a appears once, but song_b is in there too
|
||||
assert len(history) == 2
|
||||
# song_a is most recent since it was played last
|
||||
assert history[0]["video_id"] == "song_a"
|
||||
|
||||
def test_history_respects_limit(self, music_db):
|
||||
"""get_history respects the limit parameter."""
|
||||
for i in range(5):
|
||||
_record_play(music_db, video_id=f"song_{i}", title=f"S{i}")
|
||||
history = music_db.get_history(limit=3)
|
||||
assert len(history) == 3
|
||||
|
||||
def test_empty_history(self, music_db):
|
||||
"""get_history on empty DB returns empty list."""
|
||||
assert music_db.get_history() == []
|
||||
|
||||
|
||||
class TestStats:
|
||||
def test_get_stats_empty(self, music_db):
|
||||
"""get_stats on empty DB returns zero counts."""
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 0
|
||||
assert stats["total_plays"] is None or stats["total_plays"] == 0
|
||||
assert stats["favourites"] == 0
|
||||
assert stats["top_played"] == []
|
||||
assert stats["recent"] == []
|
||||
assert stats["plays_by_day"] == []
|
||||
|
||||
def test_get_stats_with_data(self, music_db):
|
||||
"""get_stats returns accurate aggregate data."""
|
||||
_record_play(music_db, video_id="song_a", title="A", duration=120)
|
||||
_record_play(music_db, video_id="song_b", title="B", duration=180)
|
||||
_record_play(music_db, video_id="song_a", title="A", duration=120)
|
||||
music_db.toggle_favourite("song_b")
|
||||
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 2
|
||||
assert stats["total_plays"] == 3
|
||||
assert stats["favourites"] == 1
|
||||
assert stats["top_played"][0]["video_id"] == "song_a"
|
||||
assert stats["top_played"][0]["play_count"] == 2
|
||||
assert len(stats["recent"]) == 3
|
||||
assert len(stats["plays_by_day"]) >= 1 # today at least
|
||||
|
||||
def test_top_played_limit_returns_up_to_ten(self, music_db):
|
||||
"""top_played in stats returns at most 10 entries."""
|
||||
for i in range(15):
|
||||
_record_play(music_db, video_id=f"song_{i}", title=f"S{i}")
|
||||
for _ in range(i + 1):
|
||||
_record_play(music_db, video_id=f"song_{i}", title=f"S{i}")
|
||||
stats = music_db.get_stats()
|
||||
assert len(stats["top_played"]) == 10
|
||||
|
||||
|
||||
class TestEdgeCases:
|
||||
def test_unicode_song_title(self, music_db):
|
||||
"""record_play handles Unicode characters in titles and artists."""
|
||||
_record_play(music_db, title="🎵 Джаз", artists=[{"name": "José", "id": "j"}])
|
||||
history = music_db.get_history(limit=10)
|
||||
assert history[0]["title"] == "🎵 Джаз"
|
||||
|
||||
def test_empty_artists_list(self, music_db):
|
||||
"""record_play works with empty artists list."""
|
||||
_record_play(music_db, artists=[])
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 1
|
||||
|
||||
def test_very_long_title(self, music_db):
|
||||
"""record_play handles very long titles."""
|
||||
long_title = "A" * 1000
|
||||
_record_play(music_db, title=long_title)
|
||||
history = music_db.get_history(limit=10)
|
||||
assert history[0]["title"] == long_title
|
||||
|
||||
def test_concurrent_access(self, music_db):
|
||||
"""Multiple rapid calls don't cause errors."""
|
||||
import threading
|
||||
|
||||
errors = []
|
||||
|
||||
def worker(vid):
|
||||
try:
|
||||
_record_play(music_db, video_id=vid, title=f"S{vid}")
|
||||
music_db.toggle_favourite(vid)
|
||||
music_db.is_favourite(vid)
|
||||
music_db.get_stats()
|
||||
except Exception as e:
|
||||
errors.append(e)
|
||||
|
||||
threads = [threading.Thread(target=worker, args=(f"vid_{i}",)) for i in range(20)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join(timeout=5)
|
||||
|
||||
assert not errors, f"Concurrent errors: {errors}"
|
||||
stats = music_db.get_stats()
|
||||
assert stats["unique_songs"] == 20
|
||||
277
tests/test_player_helpers.py
Normal file
277
tests/test_player_helpers.py
Normal file
@@ -0,0 +1,277 @@
|
||||
"""Tests for helper functions in the player module (player.py)."""
|
||||
|
||||
import os
|
||||
import struct
|
||||
import math
|
||||
import tempfile
|
||||
|
||||
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)
|
||||
|
||||
with open(os.path.join(td, "vid123.mp3"), "w") as f:
|
||||
pass # empty
|
||||
|
||||
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"] == ""
|
||||
Reference in New Issue
Block a user