Dymas/kaizoku
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace legacy branding with Kaizoku

Browse Source
This commit is contained in:
Codex
2026-08-09 13:42:51 +02:00
parent 95c9eca99f
commit fa7a922d13
15 changed files with 223 additions and 459 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_app.py
+119 -307
View File
@@ -22,12 +22,13 @@ if str(ROOT) not in sys.path:
import queue_jobs
import http_handler
import provider_bridge
import provider_downloader
os.environ.setdefault("ANI_CLI_WEB_DISABLE_WORKER", "1")
os.environ["ANI_CLI_WEB_STATE_ROOT"] = TEMP_STATE.name
os.environ.setdefault("KAIZOKU_DISABLE_WORKER", "1")
os.environ["KAIZOKU_STATE_ROOT"] = TEMP_STATE.name
SPEC = importlib.util.spec_from_file_location("ani_cli_web_app", MODULE_PATH)
SPEC = importlib.util.spec_from_file_location("kaizoku_app", MODULE_PATH)
APP = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
SPEC.loader.exec_module(APP)
@@ -119,7 +120,7 @@ class NetworkGuardTests(unittest.TestCase):
handler = DummyHandler("/api/config")
handler.client_address = ("192.168.1.25", 8421)
with mock.patch.dict(os.environ, {"ANI_CLI_WEB_ALLOW_REMOTE": "1"}, clear=False):
with mock.patch.dict(os.environ, {"KAIZOKU_ALLOW_REMOTE": "1"}, clear=False):
with self.assertRaises(PermissionError) as ctx:
APP.Handler.ensure_client_access(handler)
@@ -134,9 +135,9 @@ class NetworkGuardTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -148,14 +149,14 @@ class NetworkGuardTests(unittest.TestCase):
session_value = http_handler.remote_access_session_value(
{"auth_username": "demo", "auth_password": "secret-pass"}
)
handler.headers["Cookie"] = f"ani_cli_web_session={session_value}"
handler.headers["Cookie"] = f"kaizoku_session={session_value}"
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -169,9 +170,9 @@ class NetworkGuardTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -199,9 +200,9 @@ class NetworkGuardTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -220,9 +221,9 @@ class NetworkGuardTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -231,7 +232,7 @@ class NetworkGuardTests(unittest.TestCase):
headers = dict(handler.response_headers)
self.assertEqual(handler.response_status, HTTPStatus.SEE_OTHER)
self.assertEqual(headers.get("Location"), "/watchlist")
self.assertIn("ani_cli_web_session=", headers.get("Set-Cookie", ""))
self.assertIn("kaizoku_session=", headers.get("Set-Cookie", ""))
def test_remote_session_cookie_rejects_missing_origin_on_post(self):
body = b'{"show_id":"show-1"}'
@@ -239,7 +240,7 @@ class NetworkGuardTests(unittest.TestCase):
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Host"] = "anime.example:8421"
handler.headers["Cookie"] = (
"ani_cli_web_session="
"kaizoku_session="
+ http_handler.remote_access_session_value({"auth_username": "demo", "auth_password": "secret-pass"})
)
handler.handler_context = mock.Mock(remove_from_watchlist=mock.Mock(return_value={"ok": True}), http_error=APP.HttpError)
@@ -247,9 +248,9 @@ class NetworkGuardTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -265,7 +266,7 @@ class NetworkGuardTests(unittest.TestCase):
handler.headers["Host"] = "anime.example:8421"
handler.headers["Origin"] = "https://elsewhere.example"
handler.headers["Cookie"] = (
"ani_cli_web_session="
"kaizoku_session="
+ http_handler.remote_access_session_value({"auth_username": "demo", "auth_password": "secret-pass"})
)
handler.handler_context = mock.Mock(remove_from_watchlist=mock.Mock(return_value={"ok": True}), http_error=APP.HttpError)
@@ -273,9 +274,9 @@ class NetworkGuardTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -291,7 +292,7 @@ class NetworkGuardTests(unittest.TestCase):
handler.headers["Host"] = "anime.example:8421"
handler.headers["Origin"] = "https://anime.example:8421"
handler.headers["Cookie"] = (
"ani_cli_web_session="
"kaizoku_session="
+ http_handler.remote_access_session_value({"auth_username": "demo", "auth_password": "secret-pass"})
)
remove_from_watchlist = mock.Mock(return_value={"ok": True})
@@ -300,9 +301,9 @@ class NetworkGuardTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -331,7 +332,7 @@ class NetworkGuardTests(unittest.TestCase):
handler.client_address = ("127.0.0.1", 8421)
handler.headers["X-Forwarded-For"] = "203.0.113.10"
with mock.patch.dict(os.environ, {"ANI_CLI_WEB_ALLOW_REMOTE": "1"}, clear=False):
with mock.patch.dict(os.environ, {"KAIZOKU_ALLOW_REMOTE": "1"}, clear=False):
with self.assertRaises(PermissionError) as ctx:
APP.Handler.ensure_client_access(handler)
@@ -392,13 +393,13 @@ class DownloadQueueCancelTests(unittest.TestCase):
queue = object.__new__(APP.DownloadQueue)
queue.lock = threading.RLock()
queue._save_job_locked = lambda _job: None
job = {"id": "job-1", "download_backend": "animdl", "log": [], "updated_at": APP.now_iso()}
job = {"id": "job-1", "download_backend": "kaizoku", "log": [], "updated_at": APP.now_iso()}
with mock.patch.dict(os.environ, {"ANI_CLI_WEB_JOB_STDOUT": "true"}, clear=False), mock.patch("builtins.print") as print_mock:
with mock.patch.dict(os.environ, {"KAIZOKU_JOB_STDOUT": "true"}, clear=False), mock.patch("builtins.print") as print_mock:
APP.DownloadQueue._append_log(queue, job, "\x1b[32mDownloading episode 1\x1b[0m\n")
print_mock.assert_called_once_with(
"[download:job-1:animdl] Downloading episode 1",
"[download:job-1:kaizoku] Downloading episode 1",
flush=True,
)
self.assertEqual(job["log"], ["Downloading episode 1"])
@@ -848,66 +849,6 @@ class QueueApiTests(unittest.TestCase):
self.assertTrue(Path(moved[0]).exists())
self.assertTrue(Path(moved[1]).exists())
def test_tv_finalizer_uses_requested_episode_numbers_for_anipy_fallback(self):
with tempfile.TemporaryDirectory() as temp_root:
job = APP.app_support.build_job(
{
"query": "Fallback Show",
"title": "Fallback Show",
"anime_name": "Fallback Show",
"media_type": "tv",
"mode": "sub",
"quality": "best",
"episodes": "11",
"download_dir": temp_root,
"season": "2",
},
{"mode": "sub", "quality": "best", "download_dir": temp_root},
)
job["download_backend"] = "anipy-cli"
staging_dir = APP.app_support.job_staging_dir(job)
staging_dir.mkdir(parents=True, exist_ok=True)
source = staging_dir / "Fallback Show - S02E01.mp4"
source.write_bytes(b"fallback")
moved = APP.app_support.finalize_library_files(job)
self.assertEqual(
moved,
[f"{temp_root}/tv/Fallback Show/Season 02/Fallback Show - S02E11.mp4"],
)
self.assertTrue(Path(moved[0]).exists())
def test_tv_finalizer_uses_requested_episode_numbers_for_animdl_fallback(self):
with tempfile.TemporaryDirectory() as temp_root:
job = APP.app_support.build_job(
{
"query": "Fallback Show",
"title": "Fallback Show",
"anime_name": "Fallback Show",
"media_type": "tv",
"mode": "sub",
"quality": "best",
"episodes": "12",
"download_dir": temp_root,
"season": "2",
},
{"mode": "sub", "quality": "best", "download_dir": temp_root},
)
job["download_backend"] = "animdl"
staging_dir = APP.app_support.job_staging_dir(job)
staging_dir.mkdir(parents=True, exist_ok=True)
source = staging_dir / "Fallback Show - S02E01.mp4"
source.write_bytes(b"fallback")
moved = APP.app_support.finalize_library_files(job)
self.assertEqual(
moved,
[f"{temp_root}/tv/Fallback Show/Season 02/Fallback Show - S02E12.mp4"],
)
self.assertTrue(Path(moved[0]).exists())
def test_shutdown_wait_cancels_active_process_for_deterministic_teardown(self):
queue = object.__new__(APP.DownloadQueue)
queue.lock = threading.RLock()
@@ -937,7 +878,7 @@ class DownloadQueueWorkerFailureTests(unittest.TestCase):
job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1"})
staging_dir = queue_jobs.job_staging_dir(job)
with mock.patch.object(queue_jobs.subprocess, "Popen", side_effect=FileNotFoundError("ani-cli missing")):
with mock.patch.object(queue_jobs.subprocess, "Popen", side_effect=FileNotFoundError("kaizoku downloader missing")):
queue._run_job(job)
stored = queue._find(job["id"])
@@ -1090,58 +1031,13 @@ class DownloadQueueWorkerFailureTests(unittest.TestCase):
self.assertIn("Watchlist sync skipped for detached job.", stored["log"])
self.assertNotIn("Watchlist download state synced.", stored["log"])
def test_failed_ani_cli_retries_with_anipy_when_enabled(self):
def test_selected_unavailable_kaizoku_fails_without_start_attempt(self):
queue = APP.DownloadQueue(
lambda: {
"mode": "sub",
"quality": "best",
"download_dir": "/tmp/example",
"anipy_cli_fallback_enabled": True,
},
start_worker=False,
)
job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1"})
commands = []
class FakeProc:
def __init__(self, pid, stdout, exit_code):
self.pid = pid
self.stdout = stdout
self._exit_code = exit_code
def wait(self):
return self._exit_code
processes = [
FakeProc(4321, ["ani-cli attempt\n"], 1),
FakeProc(5432, ["anipy-cli attempt\n"], 0),
]
def fake_popen(command, **kwargs):
commands.append(command)
return processes.pop(0)
with mock.patch.object(queue_jobs, "cli_executable_exists", return_value=True), mock.patch.object(
queue_jobs.subprocess, "Popen", side_effect=fake_popen
), mock.patch.object(
queue_jobs, "finalize_library_files", return_value=["/tmp/example/Queue Show - S01E01.mp4"]
):
queue._run_job(job)
stored = queue._find(job["id"])
self.assertEqual(stored["status"], "done")
self.assertTrue(stored["fallback_used"])
self.assertEqual(commands[0][0], APP.app_support.ANI_CLI)
self.assertEqual(commands[1][0], APP.app_support.ANIPY_CLI)
self.assertIn("Fallback download completed with anipy-cli.", stored["log"])
def test_selected_unavailable_animdl_fails_without_start_attempt(self):
queue = APP.DownloadQueue(
lambda: {
"mode": "sub",
"quality": "best",
"download_dir": "/tmp/example",
"download_methods": ["animdl"],
"download_methods": ["kaizoku"],
},
start_worker=False,
)
@@ -1155,24 +1051,23 @@ class DownloadQueueWorkerFailureTests(unittest.TestCase):
stored = queue._find(job["id"])
self.assertEqual(stored["status"], "failed")
self.assertEqual(stored["download_backend"], "animdl")
self.assertIn("No selected download methods are installed or executable: animdl.", stored["log"])
self.assertEqual(stored["download_backend"], "kaizoku")
self.assertIn("No selected download methods are installed or executable: kaizoku.", stored["log"])
self.assertFalse(staging_dir.exists())
popen.assert_not_called()
def test_selected_download_methods_retry_through_animdl(self):
def test_selected_download_methods_runs_kaizoku(self):
queue = APP.DownloadQueue(
lambda: {
"mode": "sub",
"quality": "best",
"download_dir": "/tmp/example",
"download_methods": ["ani-cli", "anipy-cli", "animdl"],
"download_methods": ["kaizoku"],
},
start_worker=False,
)
job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1"})
commands = []
envs = []
class FakeProc:
def __init__(self, pid, stdout, exit_code):
@@ -1183,15 +1078,10 @@ class DownloadQueueWorkerFailureTests(unittest.TestCase):
def wait(self):
return self._exit_code
processes = [
FakeProc(4321, ["ani-cli attempt\n"], 1),
FakeProc(5432, ["anipy-cli attempt\n"], 1),
FakeProc(6543, ["animdl attempt\n"], 0),
]
processes = [FakeProc(4321, ["kaizoku attempt\n"], 0)]
def fake_popen(command, **kwargs):
commands.append(command)
envs.append(kwargs.get("env", {}))
return processes.pop(0)
with mock.patch.object(queue_jobs, "cli_executable_exists", return_value=True), mock.patch.object(
@@ -1203,10 +1093,9 @@ class DownloadQueueWorkerFailureTests(unittest.TestCase):
stored = queue._find(job["id"])
self.assertEqual(stored["status"], "done")
self.assertTrue(stored["fallback_used"])
self.assertEqual([command[0] for command in commands], [APP.app_support.ANI_CLI, APP.app_support.ANIPY_CLI, APP.app_support.ANIMDL])
self.assertEqual(envs[0]["ANI_CLI_PLAYER"], "download")
self.assertIn("Fallback download completed with animdl.", stored["log"])
self.assertFalse(stored["fallback_used"])
self.assertEqual([command[0] for command in commands], ["python3"])
self.assertIn("Download completed.", stored["log"])
class WatchlistRefreshAllTests(unittest.TestCase):
@@ -2051,7 +1940,7 @@ class WatchlistCompletionTests(unittest.TestCase):
self.assertNotIn("-S", APP.app_support.command_for_job(job))
def test_command_for_job_includes_result_switch_for_selected_search_result(self):
def test_command_for_job_ignores_result_index_for_provider_downloader(self):
command = APP.app_support.command_for_job(
{
"query": "Queue Show",
@@ -2059,11 +1948,13 @@ class WatchlistCompletionTests(unittest.TestCase):
"episodes": "1-10",
"mode": "dub",
"result_index": 2,
}
},
download_path="/tmp/kaizoku-staging",
)
self.assertIn("-S", command)
self.assertEqual(command[command.index("-S") + 1], "2")
self.assertNotIn("-S", command)
self.assertIn("--provider", command)
self.assertEqual(command[command.index("--provider") + 1], "anikoto")
def test_command_for_job_ignores_invalid_result_index(self):
command = APP.app_support.command_for_job(
@@ -2073,47 +1964,12 @@ class WatchlistCompletionTests(unittest.TestCase):
"episodes": "1-10",
"mode": "dub",
"result_index": "nope",
}
},
download_path="/tmp/kaizoku-staging",
)
self.assertNotIn("-S", command)
def test_command_for_job_builds_anipy_fallback_search_spec(self):
command = APP.app_support.command_for_job(
{
"query": "Re:Zero",
"quality": "720",
"episodes": "1-3",
"mode": "dub",
},
backend="anipy-cli",
download_path="/tmp/anipy-staging",
)
self.assertEqual(command[0], APP.app_support.ANIPY_CLI)
self.assertEqual(command[1:5], ["-D", "-s", "Re Zero:1-3:dub", "-q"])
self.assertIn("-l", command)
self.assertEqual(command[-1], "/tmp/anipy-staging")
def test_command_for_job_builds_animdl_download_command(self):
command = APP.app_support.command_for_job(
{
"query": "Re:Zero",
"quality": "720",
"episodes": "1-3",
"mode": "dub",
},
backend="animdl",
download_path="/tmp/animdl-staging",
)
self.assertEqual(command[0], APP.app_support.ANIMDL)
self.assertEqual(command[1:4], ["download", "Re:Zero", "-r"])
self.assertIn("-q", command)
self.assertIn("--index", command)
self.assertIn("-d", command)
self.assertEqual(command[-1], "/tmp/animdl-staging")
def test_download_watchlist_item_queues_full_episode_range_for_selected_mode(self):
item = {
"show_id": "show-42",
@@ -2135,6 +1991,7 @@ class WatchlistCompletionTests(unittest.TestCase):
runtime["download_queue"].add.assert_called_once_with(
{
"show_id": "show-42",
"provider": "anikoto",
"query": "Queue Show Source",
"title": "Queue Show",
"anime_name": "Queue Show Library",
@@ -2570,7 +2427,7 @@ class WatchlistBackupTests(unittest.TestCase):
backup = APP.export_watchlist_backup()
self.assertEqual(backup["format"], "ani-cli-web-watchlist-backup")
self.assertEqual(backup["format"], "kaizoku-watchlist-backup")
self.assertEqual(backup["format_version"], 1)
self.assertEqual(backup["tables"]["watchlist"]["rows"][0]["show_id"], "show-backup-1")
self.assertEqual(backup["tables"]["watchlist"]["rows"][0]["downloaded_dub_episodes_json"], APP.encode_episode_values(["1", "2"]))
@@ -2777,8 +2634,7 @@ class ConfigSnapshotTests(unittest.TestCase):
self.assertFalse(config["watchlist_auto_refresh_enabled"])
self.assertEqual(config["watchlist_auto_refresh_minutes"], 60)
self.assertEqual(config["watchlist_refresh_delay_seconds"], 5)
self.assertFalse(config["anipy_cli_fallback_enabled"])
self.assertEqual(config["download_methods"], ["ani-cli"])
self.assertEqual(config["download_methods"], ["kaizoku"])
self.assertFalse(config["auto_download_enabled"])
self.assertEqual(config["auto_download_mode"], "dub")
self.assertEqual(config["auto_download_quality"], "best")
@@ -2791,7 +2647,6 @@ class ConfigSnapshotTests(unittest.TestCase):
def test_normalize_config_parses_watchlist_schedule_fields(self):
config = APP.normalize_config(
{
"anipy_cli_fallback_enabled": "true",
"watchlist_auto_refresh_enabled": "true",
"watchlist_auto_refresh_minutes": "15",
"watchlist_refresh_delay_seconds": "7",
@@ -2803,8 +2658,7 @@ class ConfigSnapshotTests(unittest.TestCase):
"jellyfin_movie_dir": "~/jellyfin/movies",
}
)
self.assertTrue(config["anipy_cli_fallback_enabled"])
self.assertEqual(config["download_methods"], ["ani-cli", "anipy-cli"])
self.assertEqual(config["download_methods"], ["kaizoku"])
self.assertTrue(config["watchlist_auto_refresh_enabled"])
self.assertEqual(config["watchlist_auto_refresh_minutes"], 15)
self.assertEqual(config["watchlist_refresh_delay_seconds"], 7)
@@ -2818,12 +2672,11 @@ class ConfigSnapshotTests(unittest.TestCase):
def test_normalize_config_filters_download_methods(self):
config = APP.normalize_config(
{
"download_methods": ["animdl", "bad-method", "anipy-cli", "animdl"],
"download_methods": ["kaizoku", "bad-method", "kaizoku"],
}
)
self.assertEqual(config["download_methods"], ["animdl", "anipy-cli"])
self.assertTrue(config["anipy_cli_fallback_enabled"])
self.assertEqual(config["download_methods"], ["kaizoku"])
def test_normalize_config_filters_discord_webhook_events(self):
config = APP.normalize_config(
@@ -3000,7 +2853,7 @@ class DiscordWebhookFormattingTests(unittest.TestCase):
)
self.assertTrue(result["sent"])
self.assertEqual(payloads[0]["content"], "ani-cli-web moved a finished library into Jellyfin.")
self.assertEqual(payloads[0]["content"], "Kaizoku moved a finished library into Jellyfin.")
self.assertIn("/srv/jellyfin/Jelly Show/Season 01/Jelly Show - S01E01.mp4", payloads[0]["embeds"][1]["description"])
def test_tick_resets_timer_when_schedule_changes(self):
@@ -3092,6 +2945,7 @@ class AutoDownloadQueueTests(unittest.TestCase):
download_queue.add.assert_called_once_with(
{
"show_id": "show-42",
"provider": "anikoto",
"query": "Queue Show Source",
"title": "Queue Show",
"anime_name": "Queue Show- Alt-Name",
@@ -3147,6 +3001,7 @@ class AutoDownloadQueueTests(unittest.TestCase):
download_queue.add.assert_called_once_with(
{
"show_id": "show-99",
"provider": "anikoto",
"query": "Backlog Show Search",
"title": "Backlog Show",
"anime_name": "Backlog Show",
@@ -3202,6 +3057,7 @@ class AutoDownloadQueueTests(unittest.TestCase):
download_queue.add.assert_called_once_with(
{
"show_id": "show-first-refresh",
"provider": "anikoto",
"query": "First Refresh Source",
"title": "First Refresh Show",
"anime_name": "First Refresh Show",
@@ -3256,6 +3112,7 @@ class AutoDownloadQueueTests(unittest.TestCase):
download_queue.add.assert_called_once_with(
{
"show_id": "show-77",
"provider": "anikoto",
"query": "Split Season Search",
"title": "Split Season Show",
"anime_name": "Split Season Show",
@@ -3394,7 +3251,7 @@ class HandlerRouteTests(unittest.TestCase):
handler = DummyHandler("/api/watchlist/refresh-status")
APP.reset_runtime(wait=True)
try:
with mock.patch.dict(os.environ, {"ANI_CLI_WEB_DISABLE_WORKER": "1"}, clear=False):
with mock.patch.dict(os.environ, {"KAIZOKU_DISABLE_WORKER": "1"}, clear=False):
APP.Handler.do_GET(handler)
self.assertFalse(APP.WATCHLIST.worker_enabled)
self.assertIsNone(APP.WATCHLIST.refresh_worker)
@@ -3407,12 +3264,11 @@ class HandlerRouteTests(unittest.TestCase):
APP.initialize_runtime(start_workers=False)
def test_config_post_accepts_watchlist_schedule_fields(self):
body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","anipy_cli_fallback_enabled":true,"watchlist_auto_refresh_enabled":true,"watchlist_auto_refresh_minutes":"25","watchlist_refresh_delay_seconds":"9","auto_download_enabled":true,"auto_download_mode":"dub","auto_download_quality":"720","jellyfin_sync_enabled":true,"jellyfin_tv_dir":"/tmp/jellyfin-tv","jellyfin_movie_dir":"/tmp/jellyfin-movies"}'
body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","watchlist_auto_refresh_enabled":true,"watchlist_auto_refresh_minutes":"25","watchlist_refresh_delay_seconds":"9","auto_download_enabled":true,"auto_download_mode":"dub","auto_download_quality":"720","jellyfin_sync_enabled":true,"jellyfin_tv_dir":"/tmp/jellyfin-tv","jellyfin_movie_dir":"/tmp/jellyfin-movies"}'
handler = DummyHandler("/api/config", body=body, content_length=len(body))
APP.Handler.do_POST(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertTrue(handler.json_payload["anipy_cli_fallback_enabled"])
self.assertEqual(handler.json_payload["download_methods"], ["ani-cli", "anipy-cli"])
self.assertEqual(handler.json_payload["download_methods"], ["kaizoku"])
self.assertTrue(handler.json_payload["watchlist_auto_refresh_enabled"])
self.assertEqual(handler.json_payload["watchlist_auto_refresh_minutes"], 25)
self.assertEqual(handler.json_payload["watchlist_refresh_delay_seconds"], 9)
@@ -3424,73 +3280,42 @@ class HandlerRouteTests(unittest.TestCase):
self.assertEqual(handler.json_payload["jellyfin_movie_dir"], "/tmp/jellyfin-movies")
def test_config_post_accepts_download_methods(self):
body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","download_methods":["ani-cli","animdl"]}'
body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","download_methods":["kaizoku"]}'
handler = DummyHandler("/api/config", body=body, content_length=len(body))
APP.Handler.do_POST(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertEqual(handler.json_payload["download_methods"], ["ani-cli", "animdl"])
self.assertTrue(handler.json_payload["anipy_cli_fallback_enabled"])
self.assertEqual(handler.json_payload["download_methods"], ["kaizoku"])
def test_version_route_includes_cli_versions(self):
def test_version_route_reports_kaizoku_runtime(self):
handler = DummyHandler("/api/version")
APP.Handler.do_GET(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertEqual(handler.json_payload["name"], APP.APP_NAME)
self.assertEqual(handler.json_payload["version"], APP.VERSION)
self.assertIn("ani_cli_version", handler.json_payload)
self.assertIn("anipy_cli_version", handler.json_payload)
self.assertIn("animdl_version", handler.json_payload)
self.assertEqual(handler.json_payload["providers"], list(provider_bridge.PROVIDERS))
def test_ani_cli_version_probe_uses_download_player(self):
http_handler.installed_ani_cli_version.cache_clear()
calls = []
def fake_run(command, **kwargs):
calls.append((command, kwargs))
return mock.Mock(stdout="ani-cli 4.9.0\n", stderr="")
try:
with mock.patch.object(http_handler.subprocess, "run", side_effect=fake_run):
self.assertEqual(http_handler.installed_ani_cli_version(), "ani-cli 4.9.0")
finally:
http_handler.installed_ani_cli_version.cache_clear()
self.assertEqual(calls[0][0], [http_handler.ANI_CLI, "--version"])
self.assertEqual(calls[0][1]["env"]["ANI_CLI_PLAYER"], "download")
def test_dependencies_route_reports_fallback_tool_status(self):
def test_dependencies_route_reports_provider_runtime_status(self):
handler = DummyHandler("/api/dependencies")
APP.Handler.do_GET(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertIn("curl-impersonate", handler.json_payload)
self.assertIn("anipy-cli", handler.json_payload)
self.assertIn("animdl", handler.json_payload)
self.assertIn("node", handler.json_payload)
self.assertIn("python3", handler.json_payload)
self.assertIn("ffmpeg", handler.json_payload)
self.assertIn("provider-bridge", handler.json_payload)
def test_dependency_status_checks_docker_fallback_tool_paths(self):
with mock.patch.object(http_handler, "cli_executable_exists", return_value=True), mock.patch.object(
http_handler, "cli_executable_exists_any", return_value=True
) as exists_any:
status = http_handler.dependency_status()
def test_dependency_status_checks_provider_runtime(self):
with tempfile.TemporaryDirectory() as temp_dir:
bridge_path = Path(temp_dir) / "bridge.js"
bridge_path.write_text("", encoding="utf-8")
with mock.patch.object(http_handler.shutil, "which", return_value="/usr/bin/tool"), mock.patch.object(
http_handler.provider_bridge, "BRIDGE", bridge_path
):
status = http_handler.dependency_status()
self.assertTrue(status["animdl"])
exists_any.assert_any_call(
"curl_firefox135",
"curl_chrome136",
"curl_chrome116",
"curl_ff117",
)
exists_any.assert_any_call(
http_handler.ANIMDL,
"animdl",
"/usr/local/bin/animdl",
"/usr/bin/animdl",
)
exists_any.assert_any_call(
http_handler.ANIPY_CLI,
"anipy-cli",
"/usr/local/bin/anipy-cli",
"/usr/bin/anipy-cli",
)
self.assertTrue(status["node"])
self.assertTrue(status["python3"])
self.assertTrue(status["ffmpeg"])
self.assertTrue(status["provider-bridge"])
def test_clear_failed_route_removes_failed_jobs(self):
with APP.DOWNLOAD_QUEUE._connect() as conn:
@@ -3554,9 +3379,9 @@ class HandlerRouteTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -3577,9 +3402,9 @@ class HandlerRouteTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -3597,7 +3422,7 @@ class HandlerRouteTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_REMOTE_PATH_ROOTS": extra_root,
"KAIZOKU_REMOTE_PATH_ROOTS": extra_root,
},
clear=False,
):
@@ -3617,10 +3442,10 @@ class HandlerRouteTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"ANI_CLI_WEB_REMOTE_PATH_ROOTS": extra_root,
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_REMOTE_PATH_ROOTS": extra_root,
},
clear=False,
):
@@ -3685,9 +3510,9 @@ class HandlerRouteTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
@@ -3713,10 +3538,10 @@ class HandlerRouteTests(unittest.TestCase):
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"ANI_CLI_WEB_REMOTE_PATH_ROOTS": allowed_root,
"KAIZOKU_ALLOW_REMOTE": "1",
"KAIZOKU_AUTH_USERNAME": "demo",
"KAIZOKU_AUTH_PASSWORD": "secret-pass",
"KAIZOKU_REMOTE_PATH_ROOTS": allowed_root,
},
clear=False,
):
@@ -4005,7 +3830,7 @@ class HandlerRouteTests(unittest.TestCase):
def test_watchlist_export_route_returns_json_attachment(self):
handler = DummyHandler("/api/config/watchlist/export")
payload = {"format": "ani-cli-web-watchlist-backup", "tables": {"watchlist": {"rows": []}}}
payload = {"format": "kaizoku-watchlist-backup", "tables": {"watchlist": {"rows": []}}}
handler.handler_context = mock.Mock(
ensure_runtime=mock.Mock(),
runtime_state=mock.Mock(return_value={}),
@@ -4021,7 +3846,7 @@ class HandlerRouteTests(unittest.TestCase):
self.assertEqual(json.loads(handler.wfile.getvalue().decode("utf-8")), payload)
def test_watchlist_import_route_returns_import_summary(self):
body = b'{"format":"ani-cli-web-watchlist-backup","format_version":1,"tables":{"watchlist":{"rows":[]}}}'
body = b'{"format":"kaizoku-watchlist-backup","format_version":1,"tables":{"watchlist":{"rows":[]}}}'
handler = DummyHandler("/api/config/watchlist/import", body=body, content_length=len(body))
payload = {"message": "Imported 0 watchlist entries from backup.", "imported": 0, "removals": 0}
handler.handler_context = mock.Mock(import_watchlist_backup=mock.Mock(return_value=payload))
@@ -4108,13 +3933,8 @@ class TemplateHelperTests(unittest.TestCase):
self.assertLess(APP.CONFIG_HTML.index('<section class="deps" id="deps"></section>'), APP.CONFIG_HTML.index('<div class="settings-grid">'))
self.assertIn("@media (max-width: 1080px)", APP.CONFIG_HTML)
self.assertIn("justify-items: stretch;", APP.CONFIG_HTML)
self.assertIn('id="aniCliVersionLine"', APP.CONFIG_HTML)
self.assertIn('id="anipyCliVersionLine"', APP.CONFIG_HTML)
self.assertIn('id="animdlVersionLine"', APP.CONFIG_HTML)
self.assertIn('ani-cli ${data.ani_cli_version || "Unavailable"}', APP.CONFIG_HTML)
self.assertIn('anipy-cli ${data.anipy_cli_version || "Unavailable"}', APP.CONFIG_HTML)
self.assertIn('animdl ${data.animdl_version || "Unavailable"}', APP.CONFIG_HTML)
self.assertIn('class="download-method" type="checkbox" value="animdl"', 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)
@@ -4158,24 +3978,16 @@ class TemplateHelperTests(unittest.TestCase):
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(title)", 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 = {
"shows": {
"edges": [
{
"_id": "show-1",
"name": "Example Show",
"availableEpisodes": {"sub": 12},
}
]
}
}
with mock.patch.object(APP, "graph_request", return_value=payload), mock.patch.object(
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"}],
):