from unittest.mock import MagicMock, patch from lib.clients.debrid.torbox import TorboxHelper from lib.nav import debrid as debrid_navigation from lib.utils.debrid import debrid_utils from lib.utils.general.utils import DebridType from lib.utils.torrent import torrserver_utils def test_add_source_to_torrserver_uses_magnet_when_available(): api = MagicMock() api.add_magnet.return_value = "abc123" with patch.object(torrserver_utils, "JACKTORR_ADDON", True), patch.object( torrserver_utils, "get_torrserver_api", return_value=api ), patch.object(torrserver_utils, "notification") as notification: result = torrserver_utils.add_source_to_torrserver( magnet="magnet:?xt=urn:btih:abc123", title="Test" ) assert result == "abc123" api.add_magnet.assert_called_once_with("magnet:?xt=urn:btih:abc123", title="Test", poster="") notification.assert_called_once() def test_add_source_to_torrserver_saves_metadata_under_returned_and_source_hashes(): api = MagicMock() api.add_magnet.return_value = "RETURNEDHASH" data = '{"title": "Test Movie", "ids": {"imdb_id": "tt123"}}' with patch.object(torrserver_utils, "JACKTORR_ADDON", True), patch.object( torrserver_utils, "get_torrserver_api", return_value=api ), patch.object(torrserver_utils, "notification"), patch.object( torrserver_utils, "get_info_hash_from_magnet", return_value="MAGNETHASH" ), patch.object(torrserver_utils, "save_torrent_meta") as mock_save: result = torrserver_utils.add_source_to_torrserver( magnet="magnet:?xt=urn:btih:MAGNETHASH", info_hash="SOURCEHASH", title="Test", data=data, ) assert result == "RETURNEDHASH" saved_hashes = [call.args[0] for call in mock_save.call_args_list] assert saved_hashes == ["returnedhash", "sourcehash", "magnethash"] assert all(call.args[1]["ids"]["imdb_id"] == "tt123" for call in mock_save.call_args_list) def test_add_source_to_torrserver_uploads_torrent_file_from_url(): api = MagicMock() api.add_torrent_obj.return_value = "def456" response = MagicMock() response.content = b"torrent-bytes" response.raise_for_status.return_value = None with patch.object(torrserver_utils, "JACKTORR_ADDON", True), patch.object( torrserver_utils, "get_torrserver_api", return_value=api ), patch.object(torrserver_utils.requests, "get", return_value=response) as http_get: result = torrserver_utils.add_source_to_torrserver(url="https://example.com/test.torrent") assert result == "def456" http_get.assert_called_once() api.add_torrent_obj.assert_called_once() def test_add_source_to_torrserver_prefers_torrent_url_over_magnet(): api = MagicMock() api.add_torrent_obj.return_value = "urlhash" response = MagicMock() response.content = b"torrent-bytes" response.raise_for_status.return_value = None with patch.object(torrserver_utils, "JACKTORR_ADDON", True), patch.object( torrserver_utils, "get_torrserver_api", return_value=api ), patch.object(torrserver_utils.requests, "get", return_value=response): result = torrserver_utils.add_source_to_torrserver( url="https://filelist.io/download.php?id=1", magnet="magnet:?xt=urn:btih:abc123", ) assert result == "urlhash" api.add_torrent_obj.assert_called_once() api.add_magnet.assert_not_called() def test_add_source_to_torrserver_falls_back_to_magnet_when_url_fails(): api = MagicMock() api.add_magnet.return_value = "abc123" with patch.object(torrserver_utils, "JACKTORR_ADDON", True), patch.object( torrserver_utils, "get_torrserver_api", return_value=api ), patch.object(torrserver_utils.requests, "get", side_effect=Exception("network")): result = torrserver_utils.add_source_to_torrserver( url="https://example.com/details/1", magnet="magnet:?xt=urn:btih:abc123", ) assert result == "abc123" api.add_magnet.assert_called_once_with("magnet:?xt=urn:btih:abc123", title="", poster="") def test_add_source_to_debrid_uses_preferred_enabled_service(): helper = MagicMock() with patch.object( debrid_utils, "get_enabled_cloud_transfer_debrids", return_value=[DebridType.RD] ), patch.object(debrid_utils, "RealDebridHelper", return_value=helper), patch.object( debrid_utils, "notification" ) as notification: result = debrid_utils.add_source_to_debrid("abc123", DebridType.RD) assert result == DebridType.RD helper.add_magnet.assert_called_once_with("abc123") notification.assert_called_once() def test_choose_debrid_for_transfer_prompts_when_multiple_supported_services_enabled(): dialog = MagicMock() dialog.select.return_value = 1 with patch.object( debrid_utils, "get_enabled_cloud_transfer_debrids", return_value=[DebridType.RD, DebridType.AD], ), patch.object(debrid_utils, "Dialog", return_value=dialog): selected = debrid_utils.choose_debrid_for_transfer() assert selected == DebridType.AD dialog.select.assert_called_once() def test_offcloud_is_supported_for_pack_direct_and_cloud_transfer(): assert DebridType.OC in debrid_utils.PACK_DIRECT_DEBRID_TYPES assert DebridType.OC in debrid_utils.SUPPORTED_CLOUD_TRANSFER_DEBRIDS def test_torbox_helper_get_cloud_downloads_returns_playable_items(): helper = TorboxHelper() helper.client = MagicMock() helper.client.get_user_torrent_list.return_value = { "data": [ { "id": "torrent-1", "name": "Torrent One", "download_present": True, "files": [ {"id": "file-1", "name": "sample.txt", "size": 1}, {"id": "file-2", "name": "video.mkv", "size": 10}, ], } ] } downloads = helper.get_cloud_downloads() assert downloads == [ { "name": "video.mkv", "torrent_id": "torrent-1", "file_id": "file-2", "info_hash": "", "created_at": "", "updated_at": "", } ] def test_torbox_helper_get_link_uses_cloud_ids_when_present(): helper = TorboxHelper() helper.client = MagicMock() helper.client.create_download_link.return_value = {"data": "https://download"} with patch("lib.clients.debrid.torbox.get_public_ip", return_value="127.0.0.1"): data = helper.get_link( "", {"torrent_id": "torrent-1", "file_id": "file-2", "title": "video.mkv"}, ) assert data is not None assert data["url"] == "https://download" helper.client.create_download_link.assert_called_once_with("torrent-1", "file-2", "127.0.0.1") def test_get_tb_downloads_builds_playable_cloud_entries(): with patch.object( debrid_navigation.TorboxHelper, "get_cloud_downloads", return_value=[ { "name": "older.mkv", "torrent_id": "torrent-1", "file_id": "file-1", "info_hash": "oldhash", "created_at": "2026-03-22T10:00:00Z", "updated_at": "2026-03-22T10:00:00Z", }, { "name": "video.mkv", "torrent_id": "torrent-2", "file_id": "file-2", "info_hash": "abc123", "created_at": "2026-03-23T10:00:00Z", "updated_at": "2026-03-23T10:00:00Z", }, ], ), patch.object(debrid_navigation.cache, "get", return_value=None), patch.object( debrid_navigation.cache, "set" ), patch.object(debrid_navigation, "addDirectoryItem") as add_directory_item, patch.object( debrid_navigation, "end_of_directory" ) as end_of_directory: debrid_navigation.get_tb_downloads({}) assert add_directory_item.call_count == 2 first_url = add_directory_item.call_args_list[0].args[1] second_url = add_directory_item.call_args_list[1].args[1] assert "action=play_media" in first_url assert "mode%22%3A+%22movie%22" in first_url assert "torrent-2" in first_url assert "torrent-1" in second_url end_of_directory.assert_called_once()