mirror of
https://github.com/Dodelidoo-Labs/sonobarr.git
synced 2026-04-17 20:58:02 -04:00
354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""Core unit tests for DataHandler helper and orchestration logic."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from sonobarr_app.services.data_handler import DataHandler, FAILED_TO_ADD_STATUS, SessionState
|
|
|
|
|
|
class _FakeSocketIO:
|
|
"""Socket.IO test double capturing all emitted events."""
|
|
|
|
def __init__(self):
|
|
self.events = []
|
|
self.tasks = []
|
|
|
|
def emit(self, event, payload=None, room=None):
|
|
self.events.append((event, payload, room))
|
|
|
|
def start_background_task(self, func, *args):
|
|
self.tasks.append((func, args))
|
|
return None
|
|
|
|
|
|
class _Response:
|
|
"""HTTP response test double for Lidarr-related helper tests."""
|
|
|
|
def __init__(self, text="", payload=None):
|
|
self.text = text
|
|
self._payload = payload
|
|
|
|
def json(self):
|
|
if isinstance(self._payload, Exception):
|
|
raise self._payload
|
|
return self._payload
|
|
|
|
|
|
def _make_handler(tmp_path: Path) -> tuple[DataHandler, _FakeSocketIO]:
|
|
"""Construct a DataHandler bound to isolated temporary config paths."""
|
|
|
|
socketio = _FakeSocketIO()
|
|
app_config = {
|
|
"CONFIG_DIR": str(tmp_path / "config"),
|
|
"SETTINGS_FILE": str(tmp_path / "config" / "settings.json"),
|
|
"APP_VERSION": "test",
|
|
}
|
|
handler = DataHandler(socketio=socketio, logger=logging.getLogger("test-data-handler"), app_config=app_config)
|
|
return handler, socketio
|
|
|
|
|
|
def test_session_state_lifecycle_and_session_management(tmp_path):
|
|
"""Session state should reset and stop correctly through helper methods."""
|
|
|
|
handler, _ = _make_handler(tmp_path)
|
|
session = handler.ensure_session("sid-1", user_id=1, is_admin=True, auto_approve_artist_requests=True)
|
|
|
|
assert session.stop_event.is_set()
|
|
session.prepare_for_search()
|
|
assert session.running is True
|
|
assert not session.stop_event.is_set()
|
|
|
|
session.mark_stopped()
|
|
assert session.running is False
|
|
assert session.stop_event.is_set()
|
|
|
|
handler.remove_session("sid-1")
|
|
assert handler.get_session_if_exists("sid-1") is None
|
|
|
|
|
|
def test_coercion_and_parse_helpers(tmp_path):
|
|
"""Type coercion helpers should normalize booleans, numeric values, and monitor settings."""
|
|
|
|
handler, _ = _make_handler(tmp_path)
|
|
|
|
assert handler._coerce_bool(" yes ") is True
|
|
assert handler._coerce_bool("off") is False
|
|
assert handler._coerce_bool("maybe") is None
|
|
assert handler._coerce_int("-4", minimum=1) == 1
|
|
assert handler._coerce_int("", minimum=1) is None
|
|
assert handler._coerce_float("-2.5", minimum=0.0) == 0.0
|
|
assert handler._normalize_monitor_option("future") == "future"
|
|
assert handler._normalize_monitor_option("invalid") == ""
|
|
assert handler._normalize_monitor_new_items("new") == "new"
|
|
assert handler._parse_albums_to_monitor(" one, two\nthree ") == ["one", "two", "three"]
|
|
|
|
|
|
def test_update_settings_applies_values_and_clamps(tmp_path, monkeypatch):
|
|
"""Settings update should parse payload values, clamp invalid minima, and refresh integrations."""
|
|
|
|
handler, _ = _make_handler(tmp_path)
|
|
refresh_calls = []
|
|
|
|
monkeypatch.setattr(handler, "_configure_openai_client", lambda: refresh_calls.append("openai"))
|
|
monkeypatch.setattr(handler, "_configure_listening_services", lambda: refresh_calls.append("listening"))
|
|
monkeypatch.setattr(handler, "save_config_to_file", lambda: refresh_calls.append("save"))
|
|
monkeypatch.setattr(handler, "broadcast_personal_sources_state", lambda: refresh_calls.append("broadcast"))
|
|
handler._flask_app = SimpleNamespace(config={})
|
|
|
|
handler.update_settings(
|
|
{
|
|
"lidarr_address": " http://lidarr.local ",
|
|
"quality_profile_id": "0",
|
|
"metadata_profile_id": "2",
|
|
"similar_artist_batch_size": "-8",
|
|
"openai_max_seed_artists": "0",
|
|
"lidarr_api_timeout": "-1",
|
|
"auto_start_delay": "-15",
|
|
"auto_start": "true",
|
|
"lidarr_monitored": "false",
|
|
"openai_extra_headers": {"X-Env": "test"},
|
|
"lidarr_monitor_option": "all",
|
|
"lidarr_monitor_new_items": "new",
|
|
"lidarr_albums_to_monitor": "album-a, album-b",
|
|
"api_key": "api-test",
|
|
}
|
|
)
|
|
|
|
assert handler.lidarr_address == "http://lidarr.local"
|
|
assert handler.quality_profile_id == 1
|
|
assert handler.metadata_profile_id == 2
|
|
assert handler.similar_artist_batch_size == 1
|
|
assert handler.openai_max_seed_artists == 1
|
|
assert handler.lidarr_api_timeout == 1.0
|
|
assert handler.auto_start_delay == 0
|
|
assert handler.auto_start is True
|
|
assert handler.lidarr_monitored is False
|
|
assert handler.openai_extra_headers == "{'X-Env': 'test'}"
|
|
assert handler.lidarr_monitor_option == "all"
|
|
assert handler.lidarr_monitor_new_items == "new"
|
|
assert handler.lidarr_albums_to_monitor == ["album-a", "album-b"]
|
|
assert handler._flask_app.config["API_KEY"] == "api-test"
|
|
assert refresh_calls == ["openai", "listening", "save", "broadcast"]
|
|
|
|
|
|
def test_personal_source_state_and_filtering_helpers(tmp_path):
|
|
"""Personal discovery payload helpers should emit expected state and dedupe behavior."""
|
|
|
|
handler, socketio = _make_handler(tmp_path)
|
|
session = handler.ensure_session("sid-state", user_id=7)
|
|
handler.last_fm_user_service = object()
|
|
handler.listenbrainz_user_service = object()
|
|
|
|
handler._resolve_user = lambda user_id: SimpleNamespace(
|
|
id=user_id,
|
|
lastfm_username="lfm-user",
|
|
listenbrainz_username="",
|
|
)
|
|
|
|
handler.emit_personal_sources_state("sid-state")
|
|
|
|
emitted = [entry for entry in socketio.events if entry[0] == "personal_sources_state"][-1]
|
|
payload = emitted[1]
|
|
assert payload["lastfm"]["enabled"] is True
|
|
assert payload["listenbrainz"]["enabled"] is False
|
|
|
|
deduped = handler._dedupe_names(["Beyonce", "Beyoncé", " ", "Bjork"])
|
|
assert deduped == ["Beyonce", "Bjork"]
|
|
|
|
filtered, skipped = handler._filter_existing_seed_artists(["A", "B"], {"a"})
|
|
assert filtered == ["B"]
|
|
assert skipped == ["A"]
|
|
assert handler._format_skipped_seed_message(["A"], "AI suggestion") == "A is already in your Lidarr library."
|
|
|
|
session.recommended_artists = [{"Name": "A", "Status": ""}]
|
|
handler._refresh_recommended_artist_status(session, "sid-state", "A", "Added")
|
|
assert session.recommended_artists[0]["Status"] == "Added"
|
|
|
|
|
|
def test_lidarr_add_payload_and_failure_mapping(tmp_path):
|
|
"""Lidarr payload and failure mapping helpers should produce stable frontend statuses."""
|
|
|
|
handler, _ = _make_handler(tmp_path)
|
|
handler.root_folder_path = "/music"
|
|
handler.quality_profile_id = 11
|
|
handler.metadata_profile_id = 22
|
|
handler.lidarr_monitored = True
|
|
handler.search_for_missing_albums = False
|
|
handler.lidarr_monitor_option = "future"
|
|
handler.lidarr_albums_to_monitor = ["a1", "a2"]
|
|
handler.lidarr_monitor_new_items = "new"
|
|
|
|
payload = handler._build_lidarr_add_payload("Artist", "Artist", "mbid-1")
|
|
assert payload["qualityProfileId"] == 11
|
|
assert payload["metadataProfileId"] == 22
|
|
assert payload["addOptions"]["monitor"] == "future"
|
|
assert payload["monitorNewItems"] == "new"
|
|
|
|
handler.dry_run_adding_to_lidarr = True
|
|
body, parsed, message = handler._extract_lidarr_error_message(None)
|
|
assert "Dry-run mode" in body
|
|
assert parsed is None
|
|
assert "Dry-run mode" in message
|
|
|
|
handler.dry_run_adding_to_lidarr = False
|
|
status_already = handler._resolve_lidarr_add_failure_status(
|
|
"Artist",
|
|
"Artist",
|
|
400,
|
|
_Response(text="err", payload=[{"errorMessage": "already been added"}]),
|
|
)
|
|
status_invalid = handler._resolve_lidarr_add_failure_status(
|
|
"Artist",
|
|
"Artist",
|
|
400,
|
|
_Response(text="err", payload={"message": "Invalid Path"}),
|
|
)
|
|
status_unknown = handler._resolve_lidarr_add_failure_status(
|
|
"Artist",
|
|
"Artist",
|
|
500,
|
|
_Response(text="unknown", payload=ValueError("bad json")),
|
|
)
|
|
|
|
assert status_already == "Already in Lidarr"
|
|
assert status_invalid == "Invalid Path"
|
|
assert status_unknown == FAILED_TO_ADD_STATUS
|
|
|
|
|
|
def test_artist_permission_validation_and_recording(tmp_path):
|
|
"""Permission checks and cache updates should synchronize session and global state."""
|
|
|
|
handler, socketio = _make_handler(tmp_path)
|
|
session = handler.ensure_session("sid-add")
|
|
session.recommended_artists = [{"Name": "Artist", "Status": ""}]
|
|
|
|
assert handler._validate_artist_add_permissions(session, "sid-add", "Artist", FAILED_TO_ADD_STATUS) is False
|
|
|
|
session.user_id = 123
|
|
handler._can_add_without_approval = lambda sess: False
|
|
assert handler._validate_artist_add_permissions(session, "sid-add", "Artist", FAILED_TO_ADD_STATUS) is False
|
|
|
|
handler._can_add_without_approval = lambda sess: True
|
|
assert handler._validate_artist_add_permissions(session, "sid-add", "Artist", FAILED_TO_ADD_STATUS) is True
|
|
|
|
handler._record_added_artist(session, "New Artist")
|
|
assert {"name": "New Artist", "checked": False} in session.lidarr_items
|
|
assert "new artist" in session.cleaned_lidarr_items
|
|
|
|
toast_events = [event for event in socketio.events if event[0] == "new_toast_msg"]
|
|
assert len(toast_events) >= 2
|
|
|
|
|
|
def test_stream_seed_artists_success_and_failure(tmp_path, monkeypatch):
|
|
"""Seed-streaming helper should emit acknowledgements and final load state for both outcomes."""
|
|
|
|
handler, socketio = _make_handler(tmp_path)
|
|
session = handler.ensure_session("sid-stream", user_id=1)
|
|
|
|
session.recommended_artists = []
|
|
monkeypatch.setattr(handler, "_iter_artist_payloads_from_names", lambda names, missing=None: iter([
|
|
{"Name": "Artist A", "Status": ""},
|
|
{"Name": "Artist B", "Status": ""},
|
|
]))
|
|
monkeypatch.setattr(handler, "prepare_similar_artist_candidates", lambda s: setattr(s, "similar_artist_candidates", [1]))
|
|
|
|
ok = handler._stream_seed_artists(
|
|
session,
|
|
"sid-stream",
|
|
["Artist A", "Artist B"],
|
|
ack_event="ack",
|
|
ack_payload={"seeds": ["Artist A"]},
|
|
error_event="err",
|
|
error_message="failed",
|
|
missing_title="Missing",
|
|
missing_message="missing",
|
|
source_log_label="AI",
|
|
)
|
|
|
|
assert ok is True
|
|
assert any(event[0] == "ack" for event in socketio.events)
|
|
assert any(event[0] == "initial_load_complete" for event in socketio.events)
|
|
|
|
socketio.events.clear()
|
|
session.recommended_artists.clear()
|
|
monkeypatch.setattr(handler, "_iter_artist_payloads_from_names", lambda names, missing=None: iter([]))
|
|
|
|
failed = handler._stream_seed_artists(
|
|
session,
|
|
"sid-stream",
|
|
["Unknown"],
|
|
ack_event="ack",
|
|
ack_payload={"seeds": ["Unknown"]},
|
|
error_event="err",
|
|
error_message="failed",
|
|
missing_title="Missing",
|
|
missing_message="missing",
|
|
source_log_label="AI",
|
|
)
|
|
|
|
assert failed is False
|
|
assert any(event[0] == "err" for event in socketio.events)
|
|
|
|
|
|
def test_openai_header_parsing_and_settings_normalization(tmp_path):
|
|
"""Header parsing and loaded-settings normalization should enforce valid runtime values."""
|
|
|
|
handler, _ = _make_handler(tmp_path)
|
|
|
|
assert handler._normalize_openai_headers_field({"X-A": "1"}) == '{"X-A": "1"}'
|
|
assert handler._normalize_openai_headers_field(None) == ""
|
|
assert handler._normalize_openai_headers_field(123) == "123"
|
|
|
|
handler.openai_extra_headers = '{"X-Token": "abc", "X-Null": null}'
|
|
assert handler._parse_openai_extra_headers() == {"X-Token": "abc"}
|
|
|
|
handler.openai_extra_headers = "[]"
|
|
assert handler._parse_openai_extra_headers() == {}
|
|
|
|
defaults = handler._default_settings()
|
|
handler.lidarr_monitored = "not-bool"
|
|
handler.lidarr_albums_to_monitor = "one,two"
|
|
handler.similar_artist_batch_size = "0"
|
|
handler.openai_max_seed_artists = "bad"
|
|
handler.lidarr_api_timeout = "bad"
|
|
handler._normalize_loaded_settings(defaults)
|
|
|
|
assert handler.lidarr_monitored is True
|
|
assert handler.lidarr_albums_to_monitor == ["one", "two"]
|
|
assert handler.similar_artist_batch_size == defaults["similar_artist_batch_size"]
|
|
assert handler.openai_max_seed_artists == defaults["openai_max_seed_artists"]
|
|
assert handler.lidarr_api_timeout == float(defaults["lidarr_api_timeout"])
|
|
|
|
|
|
def test_misc_helpers_for_counts_and_env_overrides(tmp_path, monkeypatch):
|
|
"""Formatting and environment helper wrappers should return predictable types."""
|
|
|
|
handler, _ = _make_handler(tmp_path)
|
|
|
|
assert handler.format_numbers(999) == "999"
|
|
assert handler.format_numbers(12_300) == "12.3K"
|
|
assert handler.format_numbers(2_000_000) == "2.0M"
|
|
|
|
values = {
|
|
"flag_true": "true",
|
|
"flag_bad": "bad",
|
|
"i_good": "4",
|
|
"i_bad": "x",
|
|
"f_good": "2.5",
|
|
"f_bad": "x",
|
|
}
|
|
handler._env = lambda key: values.get(key, "")
|
|
|
|
assert handler._env_bool_or_empty("flag_true") is True
|
|
assert handler._env_bool_or_empty("flag_bad") == ""
|
|
assert handler._env_int_or_empty("i_good") == 4
|
|
assert handler._env_int_or_empty("i_bad") == ""
|
|
assert handler._env_float_or_empty("f_good") == 2.5
|
|
assert handler._env_float_or_empty("f_bad") == ""
|