import json
from unittest.mock import MagicMock, patch
from lib.api.tmdbv3api.as_obj import AsObj
from lib.search import (
SearchVariant,
_build_title_fallback_queries,
_check_search_caches,
_is_source_enabled,
run_search_entry,
search_client,
show_source_select,
)
from lib.utils.general.utils import Indexer
def test_search_variant_values():
assert SearchVariant.DEFAULT == "default"
assert SearchVariant.TITLE_YEAR == "title_year"
assert SearchVariant.ORIGINAL_TITLE == "original_title"
assert SearchVariant.ORIGINAL_TITLE_YEAR == "original_title_year"
def test_build_title_fallback_queries_default_variant_uses_tmdb_titles():
details = AsObj(
{
"original_name": "Shingeki no Kyojin",
"translations": {
"translations": [
{"iso_639_1": "en", "data": {"name": "Attack on Titan"}},
]
},
}
)
with patch("lib.clients.tmdb.utils.utils.get_tmdb_media_details", return_value=details):
queries = _build_title_fallback_queries(
"Ataque a los Titanes",
{"tmdb_id": "1429"},
"tv",
)
assert queries == ["Ataque a los Titanes", "Attack on Titan", "Shingeki no Kyojin"]
def test_build_title_fallback_queries_supports_variants_and_year():
details = AsObj(
{
"original_name": "Shingeki no Kyojin",
"translations": {
"translations": [
{"iso_639_1": "en", "data": {"name": "Attack on Titan"}},
]
},
}
)
with patch("lib.clients.tmdb.utils.utils.get_tmdb_media_details", return_value=details):
title_year_queries = _build_title_fallback_queries(
"Ataque a los Titanes",
{"tmdb_id": "1429"},
"tv",
variant=SearchVariant.TITLE_YEAR,
year=2013,
)
original_title_queries = _build_title_fallback_queries(
"Ataque a los Titanes",
{"tmdb_id": "1429"},
"tv",
variant=SearchVariant.ORIGINAL_TITLE,
)
original_title_year_queries = _build_title_fallback_queries(
"Ataque a los Titanes",
{"tmdb_id": "1429"},
"tv",
variant=SearchVariant.ORIGINAL_TITLE_YEAR,
year=2013,
)
assert title_year_queries == [
"Ataque a los Titanes 2013",
"Attack on Titan 2013",
"Shingeki no Kyojin 2013",
]
assert original_title_queries == ["Shingeki no Kyojin"]
assert original_title_year_queries == ["Shingeki no Kyojin 2013"]
def test_show_source_select_passes_year_from_media_metadata():
results = [object()]
metadata = {"year": 2010, "title": "Inception", "original_title": "Inception"}
with patch("lib.search.build_media_metadata", return_value=metadata), patch(
"lib.search.source_select", return_value=True
) as source_select_mock:
show_source_select(
results,
"movies",
{"tmdb_id": "27205"},
{},
"Inception",
"movie",
False,
)
item_info = source_select_mock.call_args[0][0]
assert item_info["year"] == 2010
assert item_info["query"] == "Inception"
def test_show_source_select_playnext_context_sets_direct_without_autoplay():
results = [object()]
with patch("lib.search.build_media_metadata", return_value={}), patch(
"lib.search.source_select", return_value=True
) as source_select_mock:
show_source_select(
results,
"tv",
{"tmdb_id": "123"},
{"season": 1, "episode": 2},
"Show",
"tv",
False,
autoplay_context="1",
)
item_info = source_select_mock.call_args[0][0]
assert item_info["playnext_context"] is True
assert item_info["direct_play"] is True
assert "autoplay" not in item_info
def test_run_search_entry_source_select_cancel_skipped_on_back():
params = {
"query": "Show",
"mode": "tv",
"media_type": "tv",
"ids": json.dumps({"tmdb_id": "123"}),
"tv_data": json.dumps({"season": 1, "episode": 2}),
"skip_cancel_on_back": True,
}
with patch("lib.search._handle_super_quick_play", return_value=False), patch(
"lib.search.search_client", return_value=[object()]
), patch(
"lib.search._process_search_results", return_value=[object()]
), patch("lib.search.set_content_type"), patch(
"lib.search.set_watched_title"
), patch("lib.search.auto_play_enabled", return_value=False), patch(
"lib.search.show_source_select", return_value=False
), patch("lib.search.cancel_playback") as cancel_mock:
run_search_entry(params)
cancel_mock.assert_not_called()
def test_run_search_entry_source_select_cancel_on_back_when_not_skipped():
params = {
"query": "Show",
"mode": "tv",
"media_type": "tv",
"ids": json.dumps({"tmdb_id": "123"}),
"tv_data": json.dumps({"season": 1, "episode": 2}),
"skip_cancel_on_back": False,
}
with patch("lib.search._handle_super_quick_play", return_value=False), patch(
"lib.search.search_client", return_value=[object()]
), patch(
"lib.search._process_search_results", return_value=[object()]
), patch("lib.search.set_content_type"), patch(
"lib.search.set_watched_title"
), patch("lib.search.auto_play_enabled", return_value=False), patch(
"lib.search.show_source_select", return_value=False
), patch("lib.search.cancel_playback") as cancel_mock:
run_search_entry(params)
cancel_mock.assert_called_once_with()
def test_run_search_entry_passes_decoded_episode_name_to_result_processing():
params = {
"query": "Show",
"mode": "tv",
"media_type": "tv",
"ids": json.dumps({"tmdb_id": "123"}),
"tv_data": json.dumps(
{
"name": "The%20Scales%20%26%20the%20Sword",
"season": "2",
"episode": "3",
}
),
"skip_cancel_on_back": True,
}
with patch("lib.search._handle_super_quick_play", return_value=False), patch(
"lib.search.search_client", return_value=[object()]
), patch(
"lib.search._process_search_results", return_value=[object()]
) as process_results_mock, patch("lib.search.set_content_type"), patch(
"lib.search.set_watched_title"
), patch("lib.search.auto_play_enabled", return_value=False), patch(
"lib.search.show_source_select", return_value=False
):
run_search_entry(params)
process_args = process_results_mock.call_args[0]
assert process_args[2] == "The Scales & the Sword"
assert process_args[3] == 3
assert process_args[4] == 2
def test_is_source_enabled_returns_true_when_cache_empty():
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = None
assert _is_source_enabled(Indexer.JACKETT) is True
assert _is_source_enabled(Indexer.STREMIO, "some_addon") is True
def test_is_source_enabled_returns_true_when_cache_invalid():
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = "not-json"
assert _is_source_enabled(Indexer.JACKETT) is True
@patch("lib.search.get_setting")
def test_is_source_enabled_checks_builtin_by_indexer_name(mock_get_setting):
def setting_side_effect(key):
return key in ("jackett_enabled", "prowlarr_enabled")
mock_get_setting.side_effect = setting_side_effect
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = json.dumps(["Jackett", "Prowlarr"])
assert _is_source_enabled(Indexer.JACKETT) is True
assert _is_source_enabled(Indexer.BURST) is False
assert _is_source_enabled(Indexer.PROWLARR) is True
def test_is_source_enabled_checks_stremio_addon_key():
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = json.dumps(["Stremio:addon1|url1", "Jackett"])
assert _is_source_enabled(Indexer.STREMIO, "addon1|url1") is True
assert _is_source_enabled(Indexer.STREMIO, "addon2|url2") is False
@patch("lib.search.get_setting")
def test_is_source_enabled_strict_whitelist_ignores_settings(mock_get_setting):
"""A source enabled in settings but deselected in Manage Sources must not search."""
def setting_side_effect(key):
return key in ("jackett_enabled", "prowlarr_enabled")
mock_get_setting.side_effect = setting_side_effect
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = json.dumps(["Jackett"])
assert _is_source_enabled(Indexer.JACKETT) is True
# Prowlarr is enabled in settings but was deselected in Manage Sources
assert _is_source_enabled(Indexer.PROWLARR) is False
# Burst is not enabled in settings and not selected
assert _is_source_enabled(Indexer.BURST) is False
@patch("lib.search.get_setting")
def test_is_source_enabled_external_scraper_respects_selection(mock_get_setting):
"""External scraper must respect Manage Sources selection by module name."""
def setting_side_effect(key):
if key == "external_scraper_enabled":
return True
if key == "external_scraper_module_name":
return "cocoscrapers"
return False
mock_get_setting.side_effect = setting_side_effect
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = json.dumps(["Jackett"])
assert _is_source_enabled(Indexer.EXTERNAL_SCRAPER) is False
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = json.dumps(["cocoscrapers"])
assert _is_source_enabled(Indexer.EXTERNAL_SCRAPER) is True
# Fallback literal supports older caches
with patch("lib.search.cache") as mock_cache:
mock_cache.get.return_value = json.dumps(["External Scraper"])
assert _is_source_enabled(Indexer.EXTERNAL_SCRAPER) is True
@patch("lib.search.get_setting")
@patch("lib.search.cache")
@patch("lib.search._perform_search_with_title_fallback")
@patch("lib.search._perform_search")
def test_submit_search_tasks_skips_disabled_sources(
mock_perform_search,
mock_perform_fallback,
mock_cache,
mock_get_setting,
):
def setting_side_effect(key):
if key == "jackett_enabled":
return True
return False
mock_get_setting.side_effect = setting_side_effect
mock_cache.get.return_value = json.dumps(["Jackett"])
executor = MagicMock()
tasks = []
dialog = MagicMock()
from lib.search import _submit_search_tasks
_submit_search_tasks(
executor,
tasks,
dialog,
"query",
"movies",
"movie",
1,
1,
{"imdb_id": "tt123"},
"",
"123",
"tt123",
True,
)
# Only Jackett task should be added.
# Burst/Prowlarr/Jackgram/Easynews/Stremio/ExternalScraper are disabled.
assert executor.submit.call_count == 1
@patch("lib.search.get_setting")
@patch("lib.search.cache")
@patch("lib.search.get_selected_stream_addons")
@patch("lib.search.get_addon_display_name")
def test_submit_search_tasks_managed_filters_stremio_addons(
mock_get_display_name,
mock_get_addons,
mock_cache,
mock_get_setting,
):
mock_get_setting.return_value = True
mock_cache.get.return_value = json.dumps(["Stremio:addon1|url1"])
mock_get_display_name.side_effect = lambda addon: addon.manifest.name
addon1 = MagicMock()
addon1.key.return_value = "addon1|url1"
addon1.manifest.name = "Addon One"
addon1.url.return_value = "http://a.com"
addon2 = MagicMock()
addon2.key.return_value = "addon2|url2"
addon2.manifest.name = "Addon Two"
addon2.url.return_value = "http://b.com"
mock_get_addons.return_value = [addon1, addon2]
manager = MagicMock()
dialog = MagicMock()
from lib.search import _submit_search_tasks_managed
_submit_search_tasks_managed(
manager,
dialog,
"query",
"tv",
"tv",
1,
1,
{"imdb_id": "tt123"},
"",
"123",
"tt123",
)
# Only addon1 should be submitted
stremio_calls = [
call for call in manager.submit_task.call_args_list if call[0][1] == Indexer.STREMIO
]
assert len(stremio_calls) == 1
assert stremio_calls[0][1]["scoped_addon_url"] == "http://a.com"
# ---------------------------------------------------------------------------
# _check_search_caches
# ---------------------------------------------------------------------------
class TestCheckSearchCaches:
"""Cover all branches of _check_search_caches."""
def test_standard_cache_hit_returns_results(self):
"""Standard cache hit returns cached results immediately (even empty list)."""
cached = [MagicMock(), MagicMock()]
with patch("lib.search.get_cached_results", return_value=cached):
result = _check_search_caches("q", {}, "movies", "movie", 0, 0, "scope")
assert result == cached
def test_standard_cache_hit_empty_list_returns_empty(self):
"""Standard cache hit with ``[]`` returns empty list, not None."""
with patch("lib.search.get_cached_results", return_value=[]):
result = _check_search_caches("q", {}, "movies", "movie", 0, 0, "scope")
assert result == []
def test_cache_miss_non_tv_skips_autoscrape(self):
"""Standard cache miss + non-TV mode = no autoscrape fallback."""
with patch("lib.search.get_cached_results", return_value=None):
result = _check_search_caches("q", {}, "movies", "movie", 0, 0, "scope")
assert result is None
def test_cache_miss_tv_no_ids_skips_autoscrape(self):
"""TV mode but no id_value → skip autoscrape fallback."""
with patch("lib.search.get_cached_results", return_value=None):
result = _check_search_caches("q", {}, "tv", "tv", 0, 0, "scope")
assert result is None
def test_cache_miss_tv_no_autoscrape_data_returns_none(self):
"""Autoscrape key exists but cache miss → return None."""
ids = {"tmdb_id": "123"}
with (
patch("lib.search.get_cached_results", return_value=None),
patch("lib.db.cached.cache.get", return_value=None),
patch("lib.utils.player.utils.get_autoscrape_results_cache_key", return_value="as:123_1_2"),
):
result = _check_search_caches("q", ids, "tv", "tv", 2, 1, "scope")
assert result is None
def test_cache_miss_tv_autoscrape_hit_migrates_and_returns(self):
"""Autoscrape hit → migrate to standard cache + return results."""
ids = {"imdb_id": "tt999"}
autoscrape_results = [MagicMock(), MagicMock()]
with (
patch("lib.search.get_cached_results", return_value=None),
patch("lib.db.cached.cache.get", return_value=autoscrape_results),
patch("lib.utils.player.utils.get_autoscrape_results_cache_key", return_value="as:tt999_2_3"),
patch("lib.search.cache_results") as mock_cache_results,
):
result = _check_search_caches("q", ids, "tv", "tv", 3, 2, "scope")
assert result == autoscrape_results
mock_cache_results.assert_called_once_with(
autoscrape_results, "q", "tv", "tv", 3, cache_scope="scope",
)
# ---------------------------------------------------------------------------
# search_client (orchestration)
# ---------------------------------------------------------------------------
class TestSearchClient:
"""Cover orchestration logic of search_client (branching, cache, search dispatch)."""
def test_rescrape_skips_cache_checks(self):
"""rescrape=True → skip _check_search_caches entirely."""
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year", return_value=2020),
patch("lib.search._build_title_fallback_queries", return_value=["q 2020"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches") as mock_check,
patch("lib.search.get_setting", return_value="0"),
patch("lib.search._run_simple_search", return_value=[]),
patch("lib.search.cache_results"),
):
search_client("q", {}, "movies", "movie", rescrape=True, season=0, episode=0)
mock_check.assert_not_called()
def test_not_rescrape_cache_hit_returns_early(self):
"""Not rescrape + cache hit → return cached, no search."""
cached = [MagicMock()]
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year", return_value=2020),
patch("lib.search._build_title_fallback_queries", return_value=["q 2020"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches", return_value=cached),
patch("lib.search.get_setting") as mock_get_setting,
patch("lib.search._run_simple_search") as mock_simple,
patch("lib.search._run_detailed_search") as mock_detailed,
patch("lib.search.cache_results") as mock_cache_results,
):
result = search_client("q", {}, "movies", "movie", rescrape=False, season=0, episode=0)
assert result == cached
mock_simple.assert_not_called()
mock_detailed.assert_not_called()
mock_cache_results.assert_not_called()
def test_cache_miss_detailed_dialog_branch(self):
"""Cache miss + search_dialog_style=1 → calls _run_detailed_search."""
expected = [MagicMock()]
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year", return_value=2020),
patch("lib.search._build_title_fallback_queries", return_value=["q 2020"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches", return_value=None),
patch("lib.search.get_setting", return_value="1"),
patch("lib.search._run_detailed_search", return_value=expected) as mock_detailed,
patch("lib.search._run_simple_search") as mock_simple,
patch("lib.search.cache_results") as mock_cache,
):
result = search_client("q", {}, "movies", "movie", rescrape=False, season=0, episode=0)
assert result == expected
mock_detailed.assert_called_once()
mock_simple.assert_not_called()
mock_cache.assert_called_once()
def test_cache_miss_simple_dialog_branch(self):
"""Cache miss + search_dialog_style=0 → calls _run_simple_search."""
expected = [MagicMock()]
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year", return_value=2020),
patch("lib.search._build_title_fallback_queries", return_value=["q 2020"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches", return_value=None),
patch("lib.search.get_setting", return_value="0"),
patch("lib.search._run_detailed_search") as mock_detailed,
patch("lib.search._run_simple_search", return_value=expected) as mock_simple,
patch("lib.search.cache_results"),
):
result = search_client("q", {}, "movies", "movie", rescrape=False, season=0, episode=0)
assert result == expected
mock_simple.assert_called_once()
mock_detailed.assert_not_called()
def test_show_dialog_false_uses_simple_branch(self):
"""show_dialog=False → simple branch regardless of search_dialog_style."""
expected = [MagicMock()]
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year", return_value=2020),
patch("lib.search._build_title_fallback_queries", return_value=["q 2020"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches", return_value=None),
patch("lib.search.get_setting", return_value="1"), # would trigger detailed, but show_dialog=False
patch("lib.search._run_detailed_search") as mock_detailed,
patch("lib.search._run_simple_search", return_value=expected) as mock_simple,
patch("lib.search.cache_results"),
):
result = search_client(
"q", {}, "movies", "movie", rescrape=False,
season=0, episode=0, show_dialog=False,
)
assert result == expected
mock_simple.assert_called_once()
mock_detailed.assert_not_called()
def test_detailed_search_failure_falls_back_to_simple_search(self):
"""Detailed XML UI failure → search continues through simple progress."""
expected = [MagicMock()]
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year", return_value=2020),
patch("lib.search._build_title_fallback_queries", return_value=["q 2020"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches", return_value=None),
patch("lib.search.get_setting", return_value="1"),
patch("lib.search._run_detailed_search", side_effect=RuntimeError("xml failed")) as mock_detailed,
patch("lib.search._run_simple_search", return_value=expected) as mock_simple,
patch("lib.search.cache_results") as mock_cache,
):
result = search_client("q", {}, "movies", "movie", rescrape=False, season=0, episode=0)
assert result == expected
mock_detailed.assert_called_once()
mock_simple.assert_called_once()
assert mock_simple.call_args[0][9] is True
mock_cache.assert_called_once_with(expected, "q", "movies", "movie", 0, cache_scope="scope")
def test_passes_year_to_infer_when_none(self):
"""year=None → _infer_tmdb_year is called."""
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year", return_value=2020) as mock_infer,
patch("lib.search._build_title_fallback_queries", return_value=["q"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches", return_value=None),
patch("lib.search.get_setting", return_value="0"),
patch("lib.search._run_simple_search", return_value=[]),
patch("lib.search.cache_results"),
):
search_client("q", {"tmdb_id": "123"}, "movies", "movie", rescrape=False, season=0, episode=0)
mock_infer.assert_called_once_with({"tmdb_id": "123"}, "movies")
def test_does_not_infer_year_when_provided(self):
"""year=2010 → _infer_tmdb_year is NOT called."""
with (
patch("lib.search.close_busy_dialog"),
patch("lib.search._infer_tmdb_year") as mock_infer,
patch("lib.search._build_title_fallback_queries", return_value=["q"]),
patch("lib.search._build_search_cache_scope", return_value="scope"),
patch("lib.search._check_search_caches", return_value=None),
patch("lib.search.get_setting", return_value="0"),
patch("lib.search._run_simple_search", return_value=[]),
patch("lib.search.cache_results"),
):
search_client("q", {}, "movies", "movie", rescrape=False, season=0, episode=0, year=2010)
mock_infer.assert_not_called()