', APP.CONFIG_HTML)
self.assertLess(APP.CONFIG_HTML.index('
'), APP.CONFIG_HTML.index('
'))
self.assertIn("@media (max-width: 1080px)", APP.CONFIG_HTML)
self.assertIn("justify-items: stretch;", APP.CONFIG_HTML)
self.assertIn('id="versionLine"', APP.CONFIG_HTML)
self.assertIn('id="providerLine"', APP.CONFIG_HTML)
self.assertIn('id="openChangelogBtn"', APP.CONFIG_HTML)
self.assertIn('id="changelogOverlay"', APP.CONFIG_HTML)
self.assertIn('const data = await api("/api/changelog");', APP.CONFIG_HTML)
self.assertIn('changelogContent.textContent = data.content || "Changelog unavailable.";', APP.CONFIG_HTML)
def test_config_page_includes_host_path_browser_controls(self):
self.assertIn('class="ghost browse-path-btn"', APP.CONFIG_HTML)
self.assertIn('id="pathBrowserOverlay"', APP.CONFIG_HTML)
self.assertIn('/api/fs/browse?path=${encodeURIComponent(path || state.pathBrowser.path || "")}', APP.CONFIG_HTML)
self.assertIn('rootGroup.className = "browser-roots";', APP.CONFIG_HTML)
self.assertIn('const active = data.path === rootPath || String(data.path || "").startsWith(`${rootPath}/`);', APP.CONFIG_HTML)
def test_search_page_sends_watchlist_auto_download_series(self):
self.assertIn('auto_download_series: $("seasonInput").value || "1"', APP.INDEX_HTML)
def test_search_page_closes_selection_modal_after_successful_actions(self):
self.assertIn('setNotice(data.message || "Added to watchlist.");\n closeSelectionModal();', APP.INDEX_HTML)
self.assertIn('setNotice("Added to queue. Open Queue to monitor progress.");\n closeSelectionModal();', APP.INDEX_HTML)
def test_search_page_selection_modal_controls_download_mode(self):
self.assertIn('select id="selectionMode"', APP.INDEX_HTML)
self.assertIn('auto_download_mode: $("selectionMode").value', APP.INDEX_HTML)
self.assertIn('mode: $("selectionMode").value', APP.INDEX_HTML)
self.assertIn('loadSelectedEpisodes($("selectionMode").value)', APP.INDEX_HTML)
def test_queue_page_includes_clear_failed_bulk_action(self):
self.assertIn('id="clearFailedBtn"', APP.QUEUE_HTML)
self.assertIn('/api/queue/clear-failed', APP.QUEUE_HTML)
self.assertIn('Removed failed jobs', APP.QUEUE_HTML)
def test_queue_and_watchlist_sidebars_do_not_show_help_panels(self):
self.assertNotIn("What you can do here", APP.QUEUE_HTML)
self.assertNotIn("How it works", APP.WATCHLIST_HTML)
def test_docker_uses_locked_node_dependencies(self):
package = json.loads((ROOT / "package.json").read_text(encoding="utf-8"))
lock = json.loads((ROOT / "package-lock.json").read_text(encoding="utf-8"))
dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8")
self.assertEqual(package["dependencies"]["cheerio"], "1.0.0")
self.assertEqual(lock["packages"]["node_modules/cheerio"]["version"], "1.0.0")
self.assertIn("COPY package.json package-lock.json /app/", dockerfile)
self.assertIn("npm ci --omit=dev", dockerfile)
self.assertIn("FROM lexiforest/curl-impersonate:v2.2.2 AS curl_impersonate", dockerfile)
self.assertIn("COPY --from=curl_impersonate /usr/local/bin/ /usr/local/bin/", dockerfile)
self.assertIn("LD_LIBRARY_PATH=/usr/local/lib", dockerfile)
def test_docker_compose_exposes_jellyfin_browse_roots(self):
compose = (ROOT / "docker-compose.yaml").read_text(encoding="utf-8")
entrypoint = (ROOT / "docker-entrypoint.sh").read_text(encoding="utf-8")
dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8")
self.assertIn("KAIZOKU_REMOTE_PATH_ROOTS: ${KAIZOKU_REMOTE_PATH_ROOTS:-/downloads,/jellyfin/tv,/jellyfin/movies}", compose)
self.assertIn("${JELLYFIN_TV_DIR:-./jellyfin/tv}:/jellyfin/tv", compose)
self.assertIn("${JELLYFIN_MOVIE_DIR:-./jellyfin/movies}:/jellyfin/movies", compose)
self.assertIn("mkdir -p /downloads /jellyfin/tv /jellyfin/movies /app/.kaizoku", dockerfile)
self.assertIn("chown_path()", entrypoint)
self.assertIn("Warning: could not change ownership", entrypoint)
self.assertIn("for path in /downloads /jellyfin/tv /jellyfin/movies /app/.kaizoku /app/.kaizoku/home", entrypoint)
def test_active_code_does_not_reference_retired_metadata_source(self):
active_files = ("app.py", "app_support.py", "http_handler.py", "provider_bridge.py", "provider_downloader.py")
active_text = "\n".join((ROOT / filename).read_text(encoding="utf-8") for filename in active_files)
self.assertNotIn("allmanga", active_text.lower())
self.assertNotIn("allanime", active_text.lower())
self.assertNotIn("METADATA_API", active_text)
def test_queue_page_formats_jellyfin_handoff_jobs(self):
self.assertIn('job.job_type === "jellyfin_handoff"', APP.QUEUE_HTML)
self.assertIn('const progress = `${job.completed || 0}/${job.total || 0} entries`;', APP.QUEUE_HTML)
self.assertIn('const moved = `${job.moved || 0} moved`;', APP.QUEUE_HTML)
def test_queue_page_renders_download_progress_bar(self):
self.assertIn('class="progress-wrap"', APP.QUEUE_HTML)
self.assertIn('class="progress-fill"', APP.QUEUE_HTML)
self.assertIn("function progressPercent(job)", APP.QUEUE_HTML)
self.assertIn('item.querySelector(".progress-fill").style.width = `${percent}%`;', APP.QUEUE_HTML)
self.assertIn("segment ${progress.segment}/${progress.segment_total}", APP.QUEUE_HTML)
def test_search_page_queues_selected_title_with_result_index(self):
self.assertIn('query: state.selected.title,', APP.INDEX_HTML)
self.assertIn('result_index: state.selected.index,', APP.INDEX_HTML)
def test_search_page_renders_poster_results_under_selection_panel(self):
self.assertIn('class="results-grid" id="results"', APP.INDEX_HTML)
self.assertIn("Select results", APP.INDEX_HTML)
self.assertIn("resultThumbnailUrl(item)", APP.INDEX_HTML)
self.assertIn('/api/search/thumb?title=', APP.INDEX_HTML)
self.assertIn("result-subtitle", APP.INDEX_HTML)
self.assertIn("Also known as:", APP.INDEX_HTML)
def test_search_anime_includes_alternative_titles(self):
payload = {"results": [{"id": "show-1", "title": "Example Show", "provider": "anikoto"}]}
with mock.patch.object(provider_bridge, "search", return_value=payload), mock.patch.object(
provider_bridge, "episodes", return_value=[{"number": "1", "langs": ["sub"]}],
), mock.patch.object(
APP, "extract_anidb_alternative_titles_for_title",
return_value=[{"label": "English", "value": "Example Show English"}],
):
results = APP.search_anime("Example Show", "sub")
self.assertEqual(results[0]["alternative_titles"], [{"label": "English", "value": "Example Show English"}])
def test_pages_share_serial_polling_helper(self):
self.assertIn("function startSerialPoll(callback, intervalMs)", APP.QUEUE_HTML)
self.assertIn("function startSerialPoll(callback, intervalMs)", APP.WATCHLIST_HTML)
self.assertIn("startSerialPoll(() => pollVisibleJobs(), 2500)", APP.QUEUE_HTML)
self.assertIn("startSerialPoll(async () => {", APP.WATCHLIST_HTML)
self.assertIn("window.clearTimeout(timer);", APP.QUEUE_HTML)
self.assertIn('window.addEventListener("pagehide", stop, { once: true });', APP.QUEUE_HTML)
self.assertIn('window.addEventListener("beforeunload", stop, { once: true });', APP.WATCHLIST_HTML)
def test_watchlist_page_polls_when_queued_items_exist(self):
self.assertIn("queuedCount: 0", APP.WATCHLIST_HTML)
self.assertIn("else if (state.queuedCount > 0)", APP.WATCHLIST_HTML)
self.assertIn("fetchWatchlist(state.watchlistPage, { silent: true })", APP.WATCHLIST_HTML)
def test_watchlist_page_renders_single_refresh_progress_bar(self):
self.assertIn('class="refresh-progress" id="refreshJobStatus"', APP.WATCHLIST_HTML)
self.assertIn('class="refresh-progress-fill" id="refreshJobFill"', APP.WATCHLIST_HTML)
self.assertIn("linear-gradient(90deg, var(--accent), var(--accent-strong))", APP.WATCHLIST_HTML)
self.assertNotIn("var(--accent-2)", APP.WATCHLIST_HTML)
self.assertIn('role="progressbar"', APP.WATCHLIST_HTML)
self.assertIn("function refreshJobPercent(job)", APP.WATCHLIST_HTML)
self.assertIn('$("refreshJobFill").style.transform = `scaleX(${percent / 100})`;', APP.WATCHLIST_HTML)
self.assertIn('setAttribute("aria-valuenow", String(percent))', APP.WATCHLIST_HTML)
self.assertIn('`Refreshing ${job.current_title}`', APP.WATCHLIST_HTML)
self.assertEqual(APP.WATCHLIST_HTML.count('id="refreshJobStatus"'), 1)
self.assertNotIn("refreshed${job.current_title", APP.WATCHLIST_HTML)
def test_watchlist_page_highlights_fully_ready_cards(self):
self.assertIn(".anime-card.fully-ready", APP.WATCHLIST_HTML)
self.assertIn("const isFullyReady = item.sub_complete && item.dub_complete;", APP.WATCHLIST_HTML)
self.assertIn("card.className = watchlistCardClassName(item);", APP.WATCHLIST_HTML)
def test_watchlist_page_renders_provider_source_tag(self):
self.assertIn("function providerLabel(provider)", APP.WATCHLIST_HTML)
self.assertIn('appendStatusTag(statusRow, `Source: ${item.provider_label || providerLabel(item.provider)}`);', APP.WATCHLIST_HTML)
def test_watchlist_page_download_all_replaces_open_action(self):
self.assertIn("Download all", APP.WATCHLIST_HTML)
self.assertIn('api("/api/watchlist/download-all"', APP.WATCHLIST_HTML)
self.assertIn('class="download-picker"', APP.WATCHLIST_HTML)
self.assertIn('downloadPicker.classList.toggle("visible")', APP.WATCHLIST_HTML)
self.assertIn('downloadWatchlistItem(item, "sub"', APP.WATCHLIST_HTML)
self.assertIn('downloadWatchlistItem(item, "dub"', APP.WATCHLIST_HTML)
self.assertIn("grid-template-columns: repeat(2, minmax(0, 1fr));", APP.WATCHLIST_HTML)
self.assertLess(APP.WATCHLIST_HTML.index(''), APP.WATCHLIST_HTML.index(''))
self.assertLess(APP.WATCHLIST_HTML.index(''), APP.WATCHLIST_HTML.index(''))
def test_watchlist_page_renders_alternative_titles_toggle(self):
self.assertIn("Alternative titles", APP.WATCHLIST_HTML)
self.assertIn('class="ghost alt-titles-toggle"', APP.WATCHLIST_HTML)
self.assertIn('id="altTitlesOverlay"', APP.WATCHLIST_HTML)
self.assertIn("openAltTitlesOverlay(item)", APP.WATCHLIST_HTML)
self.assertIn("closeAltTitlesOverlay()", APP.WATCHLIST_HTML)
self.assertIn("AniDB English alternative titles were not available for this show.", APP.WATCHLIST_HTML)
def test_watchlist_page_renders_auto_download_editor(self):
self.assertIn("Auto-download", APP.WATCHLIST_HTML)
self.assertIn('id="autoDownloadOverlay"', APP.WATCHLIST_HTML)
self.assertIn('api("/api/watchlist/update-auto-download"', APP.WATCHLIST_HTML)
self.assertIn("openAutoDownloadOverlay(item)", APP.WATCHLIST_HTML)
self.assertIn("closeAutoDownloadOverlay()", APP.WATCHLIST_HTML)
self.assertIn("Save auto-download", APP.WATCHLIST_HTML)
self.assertIn("Manual add does not trigger it", APP.WATCHLIST_HTML)
self.assertIn("Source name", APP.WATCHLIST_HTML)
self.assertIn('id="autoDownloadSourceName"', APP.WATCHLIST_HTML)
self.assertIn("Name source", APP.WATCHLIST_HTML)
self.assertIn('class="wide-half">Name source', APP.WATCHLIST_HTML)
self.assertIn('class="wide-half">Name', APP.WATCHLIST_HTML)
self.assertIn("Custom", APP.WATCHLIST_HTML)
self.assertIn("Series", APP.WATCHLIST_HTML)
self.assertIn("Episode offset", APP.WATCHLIST_HTML)
def test_watchlist_thumbnail_warmup_skips_repeated_attempts(self):
self.assertIn("thumbnailWarmupPending: new Set()", APP.WATCHLIST_HTML)
self.assertIn("function thumbnailWarmupToken(item)", APP.WATCHLIST_HTML)
self.assertIn("if (state.thumbnailWarmupTokens[item.show_id] === token) continue;", APP.WATCHLIST_HTML)
self.assertIn("state.thumbnailWarmupPending.delete(showId);", APP.WATCHLIST_HTML)
class AniDbCacheTests(unittest.TestCase):
def test_get_cached_anidb_titles_reuses_cached_entries(self):
original_cache = dict(APP.ANIDB_TITLES_CACHE)
try:
cached_entries = [{"aid": "1", "title": "Example", "normalized_title": "example"}]
APP.ANIDB_TITLES_CACHE.update(
{
"mtime": 123.0,
"entries": cached_entries,
"title_index": {"example": [0]},
"token_index": {"example": [0]},
"match_cache": {},
}
)
with mock.patch.object(APP, "ensure_anidb_titles_dump") as ensure_dump:
ensure_dump.return_value = mock.Mock(exists=lambda: True, stat=lambda: mock.Mock(st_mtime=123.0))
result = APP.get_cached_anidb_titles()
self.assertIs(result, cached_entries)
finally:
APP.ANIDB_TITLES_CACHE.clear()
APP.ANIDB_TITLES_CACHE.update(original_cache)
class ProviderDownloaderTests(unittest.TestCase):
def test_episode_key_keeps_episode_ten_distinct_from_one(self):
self.assertEqual(provider_downloader.episode_key(1), "1")
self.assertEqual(provider_downloader.episode_key(10), "10")
self.assertEqual(provider_downloader.episode_key("10.0"), "10")
self.assertEqual(provider_downloader.episode_key("10.5"), "10.5")
def test_wanted_episodes_preserves_range_order_and_existing_ids(self):
episodes = [
{"number": 10, "id": "ep10"},
{"number": 2, "id": "ep2"},
{"number": 11, "id": "ep11"},
{"number": 12, "id": "ep12"},
]
wanted = provider_downloader.wanted_episodes("10-12", episodes)
self.assertEqual([episode["number"] for episode in wanted], [10, 11, 12])
self.assertEqual([episode["id"] for episode in wanted], ["ep10", "ep11", "ep12"])
def test_best_hls_variant_url_selects_highest_bandwidth(self):
master = """#EXTM3U
#EXT-X-STREAM-INF:BANDWIDTH=800000,RESOLUTION=640x360
low/index.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=5500000,RESOLUTION=1920x1080
high/index.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2800000,RESOLUTION=1280x720
mid/index.m3u8
"""
self.assertEqual(
provider_downloader.best_hls_variant_url("https://cdn.example/show/master.m3u8", master),
"https://cdn.example/show/high/index.m3u8",
)
def test_parse_hls_segments_resolves_plain_and_encrypted_segments(self):
playlist = """#EXTM3U
#EXT-X-MEDIA-SEQUENCE:7
#EXTINF:4.0,
seg-1
#EXT-X-KEY:METHOD=AES-128,URI="keys/key.bin"
#EXTINF:4.0,
https://cdn.example/raw-segment
#EXTINF:4.0,
seg-3.ts
"""
segments = provider_downloader.parse_hls_segments("https://cdn.example/path/index.m3u8", playlist)
self.assertEqual(segments[0], {"url": "https://cdn.example/path/seg-1"})
self.assertEqual(
segments[1],
{
"url": "https://cdn.example/raw-segment",
"key_url": "https://cdn.example/path/keys/key.bin",
"iv": "8",
},
)
self.assertEqual(segments[2]["iv"], "9")
def test_strip_png_header_removes_short_wrapper(self):
payload = b"\x89PNG\r\n\x1a\njunkIENDabcdVIDEO"
self.assertEqual(provider_downloader.strip_png_header(payload), b"VIDEO")
def test_request_headers_keep_referer_without_synthetic_origin(self):
headers = provider_downloader.request_headers({"Referer": "https://megaplay.example/"})
self.assertEqual(headers["Referer"], "https://megaplay.example/")
self.assertNotIn("Origin", headers)
self.assertEqual(headers["Sec-Fetch-Dest"], "video")
def test_curl_fetch_bytes_returns_body_before_status_marker(self):
proc = mock.Mock(returncode=0, stdout=b"video-bytes\nKAIZOKU_HTTP_STATUS:200", stderr=b"")
with tempfile.TemporaryDirectory() as temp_root, mock.patch.object(
provider_downloader, "curl_cookie_file", return_value=Path(temp_root) / "cookies.txt"
), mock.patch.object(provider_downloader, "curl_candidates", return_value=["/usr/bin/curl"]), mock.patch.object(
provider_downloader.subprocess, "run", return_value=proc
) as run:
data = provider_downloader.curl_fetch_bytes("https://cdn.example/segment.ts", headers={"Referer": "https://player.example/"})
self.assertEqual(data, b"video-bytes")
self.assertIn("--header", run.call_args.args[0])
self.assertIn("Referer: https://player.example/", run.call_args.args[0])
self.assertIn("--cookie-jar", run.call_args.args[0])
def test_media_request_preserves_playlist_cookie_for_segments(self):
jar = http.cookiejar.CookieJar()
cookie = http.cookiejar.Cookie(
version=0,
name="cdn_session",
value="allowed",
port=None,
port_specified=False,
domain="cdn.example",
domain_specified=True,
domain_initial_dot=False,
path="/",
path_specified=True,
secure=True,
expires=None,
discard=True,
comment=None,
comment_url=None,
rest={},
rfc2109=False,
)
jar.set_cookie(cookie)
with mock.patch.object(provider_downloader, "MEDIA_COOKIE_JAR", jar):
request = provider_downloader.media_request("https://cdn.example/video/segment.ts")
self.assertEqual(request.get_header("Cookie"), "cdn_session=allowed")
def test_fetch_bytes_keeps_set_cookie_for_next_media_request(self):
class Response:
def __init__(self, url, body, cookie=""):
self.url = url
self.body = body
self.headers = Message()
if cookie:
self.headers.add_header("Set-Cookie", cookie)
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def info(self):
return self.headers
def geturl(self):
return self.url
def read(self):
return self.body
requests = []
responses = iter(
[
Response("https://cdn.example/index.m3u8", b"playlist", "cdn_session=allowed; Path=/; Secure"),
Response("https://cdn.example/video/segment.ts", b"video"),
]
)
def fake_urlopen(request, timeout):
requests.append(request)
return next(responses)
with mock.patch.object(provider_downloader, "MEDIA_COOKIE_JAR", http.cookiejar.CookieJar()), mock.patch.object(
provider_downloader.urllib.request, "urlopen", side_effect=fake_urlopen
):
provider_downloader.fetch_bytes("https://cdn.example/index.m3u8")
provider_downloader.fetch_bytes("https://cdn.example/video/segment.ts")
self.assertEqual(requests[1].get_header("Cookie"), "cdn_session=allowed")
def test_curl_fetch_uses_python_media_session_cookie(self):
proc = mock.Mock(returncode=0, stdout=b"video\nKAIZOKU_HTTP_STATUS:200", stderr=b"")
with tempfile.TemporaryDirectory() as temp_root, mock.patch.object(
provider_downloader, "curl_cookie_file", return_value=Path(temp_root) / "cookies.txt"
), mock.patch.object(provider_downloader, "curl_candidates", return_value=["/usr/bin/curl"]), mock.patch.object(
provider_downloader, "media_cookie_header", return_value="cdn_session=allowed"
), mock.patch.object(provider_downloader.subprocess, "run", return_value=proc) as run:
provider_downloader.curl_fetch_bytes("https://cdn.example/segment.ts")
self.assertIn("Cookie: cdn_session=allowed", run.call_args.args[0])
def test_curl_request_headers_preserve_browser_wrapper_defaults(self):
headers = provider_downloader.curl_request_headers(
"/usr/local/bin/curl_chrome142",
{"Referer": "https://player.example/"},
)
self.assertEqual(headers, {"Referer": "https://player.example/"})
def test_curl_candidates_prefer_new_browser_wrappers(self):
def fake_which(name):
if name in {"curl_chrome142", "curl_chrome120", "curl"}:
return f"/usr/local/bin/{name}"
return None
with mock.patch.object(provider_downloader.shutil, "which", side_effect=fake_which), mock.patch.object(
provider_downloader.os, "access", return_value=False
):
candidates = provider_downloader.curl_candidates()
self.assertEqual(candidates[:3], ["/usr/local/bin/curl_chrome142", "/usr/local/bin/curl_chrome120", "/usr/local/bin/curl"])
def test_fetch_bytes_retries_forbidden_media_with_curl_fallback(self):
error = urllib.error.HTTPError(
"https://cdn.example/segment.ts",
403,
"Forbidden",
hdrs={},
fp=None,
)
try:
with mock.patch.object(provider_downloader.urllib.request, "urlopen", side_effect=error), mock.patch.object(
provider_downloader, "curl_candidates", return_value=["/usr/bin/curl"]
), mock.patch.object(provider_downloader, "curl_fetch_bytes", return_value=b"video") as curl_fetch:
data = provider_downloader.fetch_bytes("https://cdn.example/segment.ts")
finally:
error.close()
self.assertEqual(data, b"video")
curl_fetch.assert_called_once()
def test_hls_segments_need_native_download_for_extensionless_provider_urls(self):
self.assertTrue(
provider_downloader.hls_segments_need_native_download(
[{"url": "https://p16-ad-sg.ibyteimg.com/obj/ad-site-i18n/202606025d0da6f666bd3d7b4e7ab879"}]
)
)
self.assertFalse(
provider_downloader.hls_segments_need_native_download(
[{"url": "https://cdn.example/video/segment-1.ts"}]
)
)
def test_hls_segments_need_native_download_for_snapcdn_disguised_segments(self):
self.assertTrue(
provider_downloader.hls_segments_need_native_download(
[{"url": "https://shard-102.snapcdn.top/anime/show/episode/seg-f1-00176.jpg"}]
)
)
self.assertTrue(
provider_downloader.hls_segments_need_native_download(
[{"url": "https://cdn.example/video/seg-f1-00170.js"}]
)
)
def test_hls_preflight_failure_does_not_fall_through_to_direct_ffmpeg(self):
with tempfile.TemporaryDirectory() as temp_root:
target = Path(temp_root) / "Episode.mp4"
stream = {"url": "https://cdn.example/master.m3u8", "isM3U8": True, "headers": {}}
with mock.patch.object(provider_downloader, "resolve_hls_input_url", return_value=stream["url"]), mock.patch.object(
provider_downloader, "hls_media_playlist", side_effect=RuntimeError("playlist blocked")
), mock.patch.object(provider_downloader, "download_hls_segments", return_value=1) as download_segments, mock.patch.object(
provider_downloader, "run_ffmpeg"
) as run_ffmpeg:
code = provider_downloader.download_episode(stream, target, episode_number=1)
self.assertEqual(code, 1)
download_segments.assert_called_once()
run_ffmpeg.assert_not_called()
def test_native_segment_failure_does_not_retry_same_source_from_preflight(self):
playlist = """#EXTM3U
#EXTINF:4.0,
https://shard-102.snapcdn.top/anime/show/episode/seg-f1-00083.css
"""
with tempfile.TemporaryDirectory() as temp_root:
target = Path(temp_root) / "Episode.mp4"
stream = {"url": "https://cdn.example/master.m3u8", "isM3U8": True, "headers": {}}
with mock.patch.object(provider_downloader, "resolve_hls_input_url", return_value=stream["url"]), mock.patch.object(
provider_downloader, "hls_media_playlist", return_value=(stream["url"], playlist)
), mock.patch.object(
provider_downloader, "download_hls_segments", side_effect=RuntimeError("segment forbidden")
) as download_segments, mock.patch.object(provider_downloader, "run_ffmpeg") as run_ffmpeg:
code = provider_downloader.download_episode(stream, target, episode_number=1)
self.assertEqual(code, 1)
download_segments.assert_called_once()
run_ffmpeg.assert_not_called()
def test_fetch_bytes_with_retries_recovers_from_segment_rate_limit(self):
error = urllib.error.HTTPError(
"https://cdn.example/video/raw-segment",
429,
"Too Many Requests",
hdrs={},
fp=None,
)
try:
with mock.patch.object(
provider_downloader,
"fetch_bytes",
side_effect=[error, b"video"],
) as fetch_bytes, mock.patch.object(provider_downloader.time, "sleep") as sleep:
data = provider_downloader.fetch_bytes_with_retries(
"https://cdn.example/video/raw-segment",
retries=1,
retry_label="segment",
)
finally:
error.close()
self.assertEqual(data, b"video")
self.assertEqual(fetch_bytes.call_count, 2)
sleep.assert_called_once()
def test_provider_stream_candidates_returns_resolved_sources(self):
with mock.patch.object(
provider_downloader,
"bridge",
return_value={
"sources": [
{"url": "https://cdn.example/one.m3u8", "server": "HD-1"},
{"server": "broken"},
{"url": "https://cdn.example/two.m3u8", "server": "HD-2"},
]
},
):
sources = provider_downloader.provider_stream_candidates("anikoto", "ep1", "dub", "best")
self.assertEqual([source["server"] for source in sources], ["HD-1", "HD-2"])
def test_provider_downloader_tries_next_same_provider_source_after_failure(self):
calls = []
def fake_bridge(command, provider, *args):
if command == "info":
return {"title": "Retry Source Show"}
if command == "episodes":
return {"episodes": [{"number": 1, "id": "ep1"}]}
if command == "resolve-all":
self.assertEqual(args[1], "dub")
return {
"sources": [
{"url": "https://cdn.example/first.m3u8", "server": "HD-1", "type": "dub"},
{"url": "https://cdn.example/second.m3u8", "server": "HD-2", "type": "dub"},
]
}
raise AssertionError(f"unexpected bridge command {command}")
def fake_download(stream, target, episode_number=None, episode_index=None, episode_total=None):
calls.append(stream["server"])
return 1 if stream["server"] == "HD-1" else 0
argv = [
"provider_downloader.py",
"--provider",
"anikoto",
"--show-id",
"anikoto:show-1",
"--title",
"Retry Source Show",
"--episodes",
"1",
"--mode",
"dub",
"--quality",
"best",
"--output-dir",
"/tmp/kaizoku-provider-test",
]
with mock.patch.object(provider_downloader, "bridge", side_effect=fake_bridge), mock.patch.object(
provider_downloader, "download_episode", side_effect=fake_download
), mock.patch.object(provider_downloader, "download_subtitles", return_value=[]), mock.patch.object(
provider_downloader.sys, "argv", argv
):
with self.assertRaises(SystemExit) as raised:
provider_downloader.main()
self.assertEqual(raised.exception.code, 0)
self.assertEqual(calls, ["HD-1", "HD-2"])
def test_provider_downloader_tries_next_source_after_playlist_exception(self):
calls = []
def fake_bridge(command, provider, *args):
if command == "info":
return {"title": "Retry Exception Show"}
if command == "episodes":
return {"episodes": [{"number": 1, "id": "ep1"}]}
if command == "resolve-all":
return {
"sources": [
{"url": "https://bad.example/master.m3u8", "server": "StreamHG", "type": "dub"},
{"url": "https://good.example/master.m3u8", "server": "Earnvids", "type": "dub"},
]
}
raise AssertionError(f"unexpected bridge command {command}")
def fake_download(stream, target, episode_number=None, episode_index=None, episode_total=None):
calls.append(stream["server"])
if stream["server"] == "StreamHG":
raise RuntimeError("HTTP Error 502: Bad Gateway")
return 0
argv = [
"provider_downloader.py",
"--provider",
"anikoto",
"--show-id",
"anikoto:show-1",
"--title",
"Retry Exception Show",
"--episodes",
"1",
"--mode",
"dub",
"--quality",
"best",
"--output-dir",
"/tmp/kaizoku-provider-test",
]
with mock.patch.object(provider_downloader, "bridge", side_effect=fake_bridge), mock.patch.object(
provider_downloader, "download_episode", side_effect=fake_download
), mock.patch.object(provider_downloader, "download_subtitles", return_value=[]), mock.patch.object(
provider_downloader.sys, "argv", argv
):
with self.assertRaises(SystemExit) as raised:
provider_downloader.main()
self.assertEqual(raised.exception.code, 0)
self.assertEqual(calls, ["StreamHG", "Earnvids"])
def test_provider_bridge_resolve_does_not_fallback_from_dub_to_sub(self):
bridge_js = (ROOT / "providers" / "bridge.js").read_text(encoding="utf-8")
self.assertNotIn('mode === "dub"', bridge_js)
self.assertNotIn('fetchEpisodeSources(episodeId, "sub")', bridge_js)
class DockerfilePackagingTests(unittest.TestCase):
def test_provider_updates_module_is_copied_into_image(self):
dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8")
self.assertIn("provider_updates.py", dockerfile)
if __name__ == "__main__":
unittest.main()