import importlib
import sys
from unittest.mock import MagicMock, patch
from lib.domain.torrent import TorrentStream
from lib.utils.general.utils import DebridType, IndexerType
def _load_source_select_module():
if "lib.gui.source_select" in sys.modules:
return importlib.reload(sys.modules["lib.gui.source_select"])
return importlib.import_module("lib.gui.source_select")
def test_download_file_uses_episode_filename_when_tv_data_is_present():
source_select_module = _load_source_select_module()
source_select = source_select_module.SourceSelect.__new__(source_select_module.SourceSelect)
source_select.item_information = {
"title": "Game of Thrones",
"tv_data": {"season": "1", "episode": "1"},
}
source_select.playback_info = None
selected_source = MagicMock()
playback_info = {
"title": "Game of Thrones",
"url": "https://example.com/files/game.of.thrones.mkv",
"tv_data": {"season": "1", "episode": "1"},
"mode": "movie",
}
with patch.object(
source_select_module.SourceSelect,
"_ensure_playback_info",
return_value=playback_info,
), patch(
"lib.downloader.get_destination_path", return_value="/downloads"
) as mock_get_dest, patch.object(
source_select_module, "action_url_run", return_value="builtin"
) as action_url_run, patch.object(
source_select_module.xbmc, "executebuiltin"
) as executebuiltin:
source_select._download_file(selected_source)
mock_get_dest.assert_called_once()
dest_call_data = mock_get_dest.call_args[0][0]
assert dest_call_data["title"] == "Game of Thrones"
assert dest_call_data["mode"] == "movies"
action_url_run.assert_called_once_with(
"handle_download_file",
file_name="Game of Thrones - S01E01",
url="https://example.com/files/game.of.thrones.mkv",
destination="/downloads",
)
executebuiltin.assert_called_once_with("builtin")
def test_download_file_uses_year_for_movie_filename():
source_select_module = _load_source_select_module()
source_select = source_select_module.SourceSelect.__new__(source_select_module.SourceSelect)
source_select.item_information = {
"title": "Project Hail Mary",
"year": 2025,
}
source_select.playback_info = None
selected_source = MagicMock()
playback_info = {
"title": "Project Hail Mary",
"url": "https://example.com/files/project.hail.mary.mkv",
"mode": "movie",
}
with patch.object(
source_select_module.SourceSelect,
"_ensure_playback_info",
return_value=playback_info,
), patch(
"lib.downloader.get_destination_path", return_value="/downloads"
) as mock_get_dest, patch.object(
source_select_module, "action_url_run", return_value="builtin"
) as action_url_run, patch.object(
source_select_module.xbmc, "executebuiltin"
) as executebuiltin:
source_select._download_file(selected_source)
mock_get_dest.assert_called_once()
dest_call_data = mock_get_dest.call_args[0][0]
assert dest_call_data["title"] == "Project Hail Mary"
assert dest_call_data["mode"] == "movies"
action_url_run.assert_called_once_with(
"handle_download_file",
file_name="Project Hail Mary (2025)",
url="https://example.com/files/project.hail.mary.mkv",
destination="/downloads",
)
executebuiltin.assert_called_once_with("builtin")
def test_populate_sources_list_shows_debrid_type_for_stremio_debrid_sources():
source_select_module = _load_source_select_module()
source_select = source_select_module.SourceSelect.__new__(source_select_module.SourceSelect)
class _FakeListItem:
def __init__(self, label=""):
self.label = label
self.properties = {}
def setProperty(self, key, value):
self.properties[key] = value
class _FakeDisplayList:
def __init__(self):
self.items = []
def reset(self):
self.items = []
def addItem(self, item):
self.items.append(item)
source_select.display_list = _FakeDisplayList()
source_select.filter_applied = False
source_select.filtered_sources = None
source_select.sources = [
TorrentStream(
title="Example.Source.1080p",
type=IndexerType.STREMIO_DEBRID,
debridType=DebridType.TB,
subindexer="Torrentio",
size=1024,
provider="ProviderX",
quality="1080p",
)
]
with patch.object(source_select_module.xbmcgui, "ListItem", _FakeListItem), patch.object(
source_select_module,
"parse_title_info",
return_value={
"clean_title": "Example Source",
"codec": "",
"audio": "",
"badges": "",
"release_group": "",
},
), patch.object(
source_select_module, "bytes_to_human_readable", return_value="1 KB"
), patch.object(
source_select_module, "get_provider_color", side_effect=lambda value: value
), patch.object(
source_select_module, "get_random_color", side_effect=lambda value: value
), patch.object(source_select_module, "get_colored_languages", return_value=""), patch.object(
source_select_module, "extract_publish_date", return_value=""
), patch.object(source_select_module, "get_source_status", return_value="[B]Cached[/B]"):
source_select.populate_sources_list()
assert source_select.display_list.items[0].properties["type"] == DebridType.TB
def test_populate_sources_list_uses_current_stremio_alias_for_cached_source():
source_select_module = _load_source_select_module()
source_select = source_select_module.SourceSelect.__new__(source_select_module.SourceSelect)
class _FakeListItem:
def __init__(self, label=""):
self.label = label
self.properties = {}
def setProperty(self, key, value):
self.properties[key] = value
class _FakeDisplayList:
def __init__(self):
self.items = []
def reset(self):
self.items = []
def addItem(self, item):
self.items.append(item)
addon_key = "org.example.addon|https://example.com"
source_select.display_list = _FakeDisplayList()
source_select.filter_applied = False
source_select.filtered_sources = None
source_select.sources = [
TorrentStream(
title="Example.Source.1080p",
type=IndexerType.STREMIO_DEBRID,
addonKey=addon_key,
addonName="Old Alias",
addonInstanceLabel="Old Alias (example.com, custom)",
indexer="Original",
size=1024,
quality="1080p",
),
TorrentStream(
title="Another.Source.1080p",
type=IndexerType.STREMIO_DEBRID,
addonKey="org.example.other|https://example.com",
addonInstanceLabel="Other Addon",
indexer="Original Other",
size=2048,
quality="1080p",
),
]
with patch.object(source_select_module.xbmcgui, "ListItem", _FakeListItem), patch.object(
source_select_module,
"parse_title_info",
return_value={
"clean_title": "Example Source",
"codec": "",
"audio": "",
"badges": "",
"release_group": "",
},
), patch.object(
source_select_module, "bytes_to_human_readable", return_value="1 KB"
), patch.object(
source_select_module, "get_provider_color", side_effect=lambda value: value
), patch.object(
source_select_module, "get_random_color", side_effect=lambda value: value
), patch.object(source_select_module, "get_colored_languages", return_value=""), patch.object(
source_select_module, "extract_publish_date", return_value=""
), patch.object(source_select_module, "get_source_status", return_value="[B]Cached[/B]"), patch.object(
source_select_module.cache,
"get",
return_value={addon_key: "New Alias"},
) as cache_get:
source_select.populate_sources_list()
cache_get.assert_called_once_with(source_select_module.STREMIO_ADDON_ALIASES_KEY)
assert (
source_select.display_list.items[0].properties["indexer"]
== "New Alias (example.com, custom)"
)
assert source_select.display_list.items[1].properties["indexer"] == "Other Addon"
def test_torrent_context_menu_does_not_include_download_video():
source_select_module = _load_source_select_module()
source_select = source_select_module.SourceSelect.__new__(source_select_module.SourceSelect)
source_select.position = 0
source_select.list_sources = [TorrentStream(type=IndexerType.TORRENT)]
display_list = MagicMock()
display_list.getSelectedPosition.return_value = 0
source_select.display_list = display_list
labels = {
90365: "Download to debrid",
90359: "Add to TorrServer",
90083: "Download video",
90082: "Download subtitles",
90744: "Upload subtitle",
}
dialog = MagicMock()
dialog.contextmenu.return_value = -1
with patch.object(
source_select_module,
"translation",
side_effect=lambda key: labels.get(key, str(key)),
), patch.object(source_select_module.xbmcgui, "Dialog", return_value=dialog):
source_select._handle_context_menu_action()
shown_items = dialog.contextmenu.call_args.args[0]
assert "Download video" not in shown_items
assert shown_items == [
"Download to debrid",
"Add to TorrServer",
"Download subtitles",
"Upload subtitle",
]
def test_torrent_context_menu_download_subtitles_resolves_for_subtitle_download():
source_select_module = _load_source_select_module()
source_select = source_select_module.SourceSelect.__new__(source_select_module.SourceSelect)
source_select.position = 0
source_select.item_information = {}
selected_source = TorrentStream(type=IndexerType.TORRENT)
source_select.list_sources = [selected_source]
display_list = MagicMock()
display_list.getSelectedPosition.return_value = 0
source_select.display_list = display_list
dialog = MagicMock()
dialog.contextmenu.return_value = 2
with patch.object(source_select_module.xbmcgui, "Dialog", return_value=dialog), patch.object(
source_select, "_resolve_item"
) as resolve_item:
source_select._handle_context_menu_action()
resolve_item.assert_called_once_with(selected_source, is_subtitle_download=True)