#!/usr/bin/env python3 import base64 import importlib.util import io import os import sys import tempfile import threading import unittest from http import HTTPStatus from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parent MODULE_PATH = ROOT / "app.py" TEMP_STATE = tempfile.TemporaryDirectory() if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) import queue_jobs import http_handler os.environ.setdefault("ANI_CLI_WEB_DISABLE_WORKER", "1") os.environ["ANI_CLI_WEB_STATE_ROOT"] = TEMP_STATE.name SPEC = importlib.util.spec_from_file_location("ani_cli_web_app", MODULE_PATH) APP = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None SPEC.loader.exec_module(APP) IMPORT_RUNTIME_WAS_DEFERRED = ( APP.CONFIG is None and APP.WATCHLIST is None and APP.DOWNLOAD_QUEUE is None and APP.WATCHLIST_REFRESH is None ) APP.initialize_runtime(start_workers=False) APP.reset_runtime(wait=True) APP.initialize_runtime(start_workers=False) unittest.addModuleCleanup(lambda: APP.reset_runtime(wait=True)) class DummyHandler: def __init__(self, path, body=b"{}", content_length=None): self.path = path self.client_address = ("127.0.0.1", 8421) self.handler_context = APP.Handler.handler_context self.headers = {} if content_length is not None: self.headers["Content-Length"] = str(content_length) self.rfile = io.BytesIO(body) self.json_payload = None self.json_status = None self.error_status = None self.error_message = None self.file_path = None self.response_status = None self.response_headers = [] self.wfile = io.BytesIO() def ensure_client_access(self): return APP.Handler.ensure_client_access(self) def body_json(self): return APP.Handler.body_json(self) def json(self, payload, status=HTTPStatus.OK): self.json_payload = payload self.json_status = status def error(self, status, message): self.error_status = status self.error_message = message def file(self, path): self.file_path = path def html(self, _content): raise AssertionError("HTML should not be served in this test") def exception(self, exc): return APP.Handler.exception(self, exc) def send_response(self, status): self.response_status = status def send_header(self, key, value): self.response_headers.append((key, value)) def end_headers(self): return None class NetworkGuardTests(unittest.TestCase): def test_runtime_bootstrap_is_deferred_until_initialize(self): self.assertTrue(IMPORT_RUNTIME_WAS_DEFERRED) def test_client_address_is_local_accepts_loopback(self): self.assertTrue(APP.client_address_is_local("127.0.0.1")) self.assertTrue(APP.client_address_is_local("::1")) self.assertTrue(APP.client_address_is_local("localhost")) def test_client_address_is_local_rejects_non_loopback(self): self.assertFalse(APP.client_address_is_local("192.168.1.25")) self.assertFalse(APP.client_address_is_local("10.0.0.8")) def test_remote_access_requires_credentials_for_non_local_clients(self): 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 self.assertRaises(PermissionError) as ctx: APP.Handler.ensure_client_access(handler) self.assertIn("AUTH_USERNAME", str(ctx.exception)) def test_remote_access_accepts_valid_basic_auth_for_non_local_clients(self): handler = DummyHandler("/api/config") handler.client_address = ("192.168.1.25", 8421) encoded = base64.b64encode(b"demo:secret-pass").decode("ascii") handler.headers["Authorization"] = f"Basic {encoded}" 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", }, clear=False, ): APP.Handler.ensure_client_access(handler) def test_remote_access_accepts_valid_session_cookie_for_non_local_clients(self): handler = DummyHandler("/api/config") handler.client_address = ("192.168.1.25", 8421) 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}" 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", }, clear=False, ): APP.Handler.ensure_client_access(handler) def test_remote_browser_get_without_session_renders_login_page(self): handler = DummyHandler("/") handler.client_address = ("192.168.1.25", 8421) handler.command = "GET" 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", }, clear=False, ): APP.Handler.do_GET(handler) body = handler.wfile.getvalue().decode("utf-8") self.assertEqual(handler.response_status, HTTPStatus.UNAUTHORIZED) self.assertIn('
', body) self.assertIn("Username", body) self.assertIn("Password", body) def test_remote_login_post_sets_cookie_and_redirects(self): body = b"username=demo&password=secret-pass&next=%2Fwatchlist" handler = DummyHandler("/auth/login", body=body, content_length=len(body)) handler.client_address = ("192.168.1.25", 8421) 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", }, clear=False, ): APP.Handler.do_POST(handler) 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", "")) def test_loopback_proxy_with_forwarded_remote_ip_requires_credentials(self): handler = DummyHandler("/api/config") 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 self.assertRaises(PermissionError) as ctx: APP.Handler.ensure_client_access(handler) self.assertIn("AUTH_USERNAME", str(ctx.exception)) class DownloadQueueCancelTests(unittest.TestCase): def setUp(self): queue = APP.DownloadQueue(lambda: {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, start_worker=False) with queue._connect() as conn: conn.execute("DELETE FROM jobs") def test_cancel_updates_inflight_job_state(self): queue = object.__new__(APP.DownloadQueue) queue.lock = threading.RLock() queue.current_job_id = "job-1" queue.current_process = mock.Mock(pid=4321) queue.current_job = {"id": "job-1", "status": "running", "cancel_requested": False} saved = [] queue._save_job_locked = lambda job: saved.append(dict(job)) queue._find = lambda job_id: {"id": job_id, "status": "running", "cancel_requested": False} with mock.patch.object(APP.os, "getpgid", return_value=4321), mock.patch.object(APP.os, "killpg") as killpg: result = APP.DownloadQueue.cancel(queue, "job-1") killpg.assert_called_once() self.assertTrue(result["cancel_requested"]) self.assertTrue(queue.current_job["cancel_requested"]) self.assertEqual(saved[-1]["id"], "job-1") self.assertTrue(saved[-1]["cancel_requested"]) def test_append_log_batches_persistence(self): queue = object.__new__(APP.DownloadQueue) queue.lock = threading.RLock() saved = [] def fake_save(job): saved.append(list(job.get("log", []))) job["_dirty_log_lines"] = 0 job["_last_persist_monotonic"] = 0.0 queue._save_job_locked = fake_save job = {"log": [], "updated_at": APP.now_iso()} with mock.patch.object(queue_jobs.time, "monotonic", return_value=0.0): for index in range(queue_jobs.LOG_PERSIST_LINE_BATCH - 1): APP.DownloadQueue._append_log(queue, job, f"line {index}") self.assertEqual(saved, []) APP.DownloadQueue._append_log(queue, job, "line flush") self.assertEqual(len(saved), 1) self.assertEqual(saved[0][-1], "line flush") class QueueApiTests(unittest.TestCase): def setUp(self): APP.reset_runtime(wait=True) APP.initialize_runtime(start_workers=False) self.queue = APP.DOWNLOAD_QUEUE with self.queue._connect() as conn: conn.execute("DELETE FROM jobs") def test_download_queue_get_returns_single_job(self): created = self.queue.add( { "query": "Example Show", "title": "Example Show", "anime_name": "Example Show", "mode": "sub", "quality": "best", "episodes": "1", "download_dir": "/tmp/example", "season": "1", } ) fetched = self.queue.get(created["id"]) self.assertEqual(fetched["id"], created["id"]) self.assertEqual(fetched["title"], "Example Show") def test_queue_item_get_route_returns_single_job_payload(self): created = self.queue.add( { "query": "Route Show", "title": "Route Show", "anime_name": "Route Show", "mode": "sub", "quality": "best", "episodes": "1", "download_dir": "/tmp/example", "season": "1", } ) handler = DummyHandler(f"/api/queue/{created['id']}") APP.Handler.do_GET(handler) self.assertEqual(handler.json_status, HTTPStatus.OK) self.assertEqual(handler.json_payload["id"], created["id"]) def test_retry_rejects_done_jobs(self): created = self.queue.add( { "query": "Done Show", "title": "Done Show", "anime_name": "Done Show", "mode": "sub", "quality": "best", "episodes": "1", "download_dir": "/tmp/example", "season": "1", } ) with self.queue.lock: job = self.queue._find(created["id"]) job["status"] = "done" self.queue._save_job_locked(job) with self.assertRaisesRegex(ValueError, "Only failed or canceled jobs can be retried"): self.queue.retry(created["id"]) def test_retry_allows_canceled_jobs(self): created = self.queue.add( { "query": "Canceled Show", "title": "Canceled Show", "anime_name": "Canceled Show", "mode": "sub", "quality": "best", "episodes": "1", "download_dir": "/tmp/example", "season": "1", } ) with self.queue.lock: job = self.queue._find(created["id"]) job["status"] = "canceled" job["cancel_requested"] = True self.queue._save_job_locked(job) retried = self.queue.retry(created["id"]) self.assertEqual(retried["status"], "pending") self.assertFalse(retried["cancel_requested"]) def test_detach_show_clears_queue_coverage_and_sync_binding(self): created = self.queue.add( { "show_id": "show-42", "query": "Queued Show", "title": "Queued Show", "anime_name": "Queued Show", "mode": "dub", "quality": "best", "episodes": "1-3", "download_dir": "/tmp/example", "season": "1", } ) result = self.queue.detach_show("show-42") detached = self.queue.get(created["id"]) self.assertEqual(result["count"], 1) self.assertEqual(detached["show_id"], "") self.assertTrue(detached["skip_watchlist_sync"]) self.assertEqual(self.queue.covered_episodes("show-42", "dub", ["1", "2", "3"]), set()) def test_save_job_splits_log_and_extra_payload_columns(self): queue = APP.DownloadQueue(lambda: {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, start_worker=False) job = { "id": "job-structured", "show_id": "show-1", "title": "Structured Show", "anime_name": "Structured Show", "media_type": "movie", "season": "1", "query": "Structured Show", "result_index": 1, "mode": "sub", "quality": "best", "episodes": "1-2", "download_dir": "/tmp/example", "status": "pending", "created_at": APP.now_iso(), "updated_at": APP.now_iso(), "log": ["line 1", "line 2"], "custom_flag": True, } with queue.lock: queue._save_job_locked(job) with queue._connect() as conn: row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job["id"],)).fetchone() self.assertEqual(row["title"], "Structured Show") self.assertEqual(row["media_type"], "movie") self.assertEqual(row["log_payload"], '["line 1", "line 2"]') self.assertEqual(APP.json.loads(row["payload"]), {"custom_flag": True}) restored = queue._job_from_row(row) self.assertEqual(restored["log"], ["line 1", "line 2"]) self.assertTrue(restored["custom_flag"]) self.assertEqual(restored["media_type"], "movie") def test_movie_finalizer_uses_movie_folder_and_single_file_name(self): with tempfile.TemporaryDirectory() as temp_root: job = APP.app_support.build_job( { "query": "Movie Show", "title": "Movie Show", "anime_name": "Movie Show", "media_type": "movie", "mode": "sub", "quality": "best", "episodes": "1", "download_dir": temp_root, "season": "1", }, {"mode": "sub", "quality": "best", "download_dir": temp_root}, ) staging_dir = APP.app_support.job_staging_dir(job) staging_dir.mkdir(parents=True, exist_ok=True) source = staging_dir / "Movie Show Episode 1.mp4" source.write_bytes(b"movie-bytes") moved = APP.app_support.finalize_library_files(job) self.assertEqual(moved, [f"{temp_root}/movie/Movie Show/Movie Show.mp4"]) self.assertFalse(source.exists()) self.assertTrue(Path(moved[0]).exists()) def test_tv_finalizer_applies_episode_offset_to_named_files(self): with tempfile.TemporaryDirectory() as temp_root: job = APP.app_support.build_job( { "query": "Split Season Show", "title": "Split Season Show", "anime_name": "Split Season Show", "media_type": "tv", "mode": "sub", "quality": "best", "episodes": "1-2", "download_dir": temp_root, "season": "2", "episode_offset": "13", }, {"mode": "sub", "quality": "best", "download_dir": temp_root}, ) staging_dir = APP.app_support.job_staging_dir(job) staging_dir.mkdir(parents=True, exist_ok=True) first = staging_dir / "Split Season Show Episode 1.mp4" second = staging_dir / "Split Season Show Episode 2.mp4" first.write_bytes(b"ep1") second.write_bytes(b"ep2") moved = APP.app_support.finalize_library_files(job) self.assertEqual( moved, [ f"{temp_root}/tv/Split Season Show/Season 02/Split Season Show - S02E13.mp4", f"{temp_root}/tv/Split Season Show/Season 02/Split Season Show - S02E14.mp4", ], ) self.assertTrue(Path(moved[0]).exists()) self.assertTrue(Path(moved[1]).exists()) def test_shutdown_wait_cancels_active_process_for_deterministic_teardown(self): queue = object.__new__(APP.DownloadQueue) queue.lock = threading.RLock() queue.stop_event = threading.Event() queue.wakeup = threading.Event() queue.current_process = mock.Mock(pid=4321) queue.current_job = {"id": "job-1", "cancel_requested": False} queue.worker = mock.Mock() queue.worker.is_alive.side_effect = [True, False, False] with mock.patch.object(APP.os, "getpgid", return_value=4321), mock.patch.object(APP.os, "killpg") as killpg: stopped = APP.DownloadQueue.shutdown(queue, wait=True, cancel_active=False) self.assertTrue(stopped) self.assertTrue(queue.current_job["cancel_requested"]) killpg.assert_called_once_with(4321, queue_jobs.signal.SIGTERM) class DownloadQueueWorkerFailureTests(unittest.TestCase): def setUp(self): queue = APP.DownloadQueue(lambda: {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, start_worker=False) with queue._connect() as conn: conn.execute("DELETE FROM jobs") def test_run_job_marks_startup_failure_and_cleans_staging(self): queue = APP.DownloadQueue(lambda: {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, start_worker=False) 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")): queue._run_job(job) stored = queue._find(job["id"]) self.assertEqual(stored["status"], "failed") self.assertEqual(stored["exit_code"], 1) self.assertIn("Could not start download", stored["log"][-1]) self.assertFalse(staging_dir.exists()) def test_failed_download_cleans_staging_dir(self): queue = APP.DownloadQueue(lambda: {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, start_worker=False) job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1"}) staging_dir = queue_jobs.job_staging_dir(job) staging_dir.mkdir(parents=True, exist_ok=True) (staging_dir / "partial.mp4").write_text("partial", encoding="utf-8") class FakeProc: pid = 4321 stdout = [] def wait(self): return 1 with mock.patch.object(queue_jobs.subprocess, "Popen", return_value=FakeProc()): queue._run_job(job) stored = queue._find(job["id"]) self.assertEqual(stored["status"], "failed") self.assertFalse(staging_dir.exists()) def test_zero_exit_without_downloaded_files_marks_job_failed_with_clear_message(self): queue = APP.DownloadQueue(lambda: {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, start_worker=False) job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1-10"}) staging_dir = queue_jobs.job_staging_dir(job) class FakeProc: pid = 4321 stdout = ["Checking dependencies...\n"] def wait(self): return 0 with mock.patch.object(queue_jobs.subprocess, "Popen", return_value=FakeProc()): queue._run_job(job) stored = queue._find(job["id"]) self.assertEqual(stored["status"], "failed") self.assertEqual(stored["exit_code"], 1) self.assertIn("without downloading any files", "\n".join(stored["log"])) self.assertFalse(staging_dir.exists()) def test_run_job_marks_post_start_exception_failed_without_killing_worker_state(self): queue = APP.DownloadQueue(lambda: {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, start_worker=False) job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1"}) staging_dir = queue_jobs.job_staging_dir(job) class BrokenStdout: def __iter__(self): return self def __next__(self): raise UnicodeDecodeError("utf-8", b"\xff", 0, 1, "bad output") class FakeProc: pid = 4321 stdout = BrokenStdout() def terminate(self): return None with mock.patch.object(queue_jobs.subprocess, "Popen", return_value=FakeProc()), mock.patch.object( queue_jobs.os, "getpgid", side_effect=OSError("gone") ): queue._run_job(job) stored = queue._find(job["id"]) self.assertEqual(stored["status"], "failed") self.assertEqual(stored["exit_code"], 1) self.assertIn("unexpectedly", stored["log"][-1].lower()) self.assertIsNone(queue.current_process) self.assertIsNone(queue.current_job_id) self.assertIsNone(queue.current_job) self.assertFalse(staging_dir.exists()) def test_successful_download_sends_completion_notification(self): queue = APP.DownloadQueue( lambda: { "mode": "sub", "quality": "best", "download_dir": "/tmp/example", "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["download_completed"], }, watchlist_sync_fn=lambda _job: { "show_id": "show-55", "title": "Queue Show", "category": "finished", "thumbnail_path": None, }, start_worker=False, ) job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1-2"}) class FakeProc: pid = 4321 stdout = [] def wait(self): return 0 with mock.patch.object(queue_jobs.subprocess, "Popen", return_value=FakeProc()), mock.patch.object( queue_jobs, "finalize_library_files", return_value=["/tmp/example/Queue Show - S01E01.mp4"] ), mock.patch.object(queue_jobs, "send_discord_webhook_event") as send_webhook: queue._run_job(job) stored = queue._find(job["id"]) self.assertEqual(stored["status"], "done") self.assertIn("Download completed.", stored["log"]) send_webhook.assert_called_once() self.assertEqual(send_webhook.call_args.args[1], "download_completed") self.assertEqual(send_webhook.call_args.args[2]["title"], "Queue Show") self.assertEqual(send_webhook.call_args.args[2]["category"], "finished") self.assertEqual(send_webhook.call_args.args[2]["moved_files"], ["/tmp/example/Queue Show - S01E01.mp4"]) def test_successful_detached_download_logs_sync_skipped(self): queue = APP.DownloadQueue( lambda: { "mode": "sub", "quality": "best", "download_dir": "/tmp/example", }, watchlist_sync_fn=lambda _job: None, start_worker=False, ) job = queue.add({"query": "Queue Show", "title": "Queue Show", "season": "1", "episodes": "1"}) job["skip_watchlist_sync"] = True class FakeProc: pid = 4321 stdout = [] def wait(self): return 0 with mock.patch.object(queue_jobs.subprocess, "Popen", return_value=FakeProc()), 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.assertIn("Watchlist sync skipped for detached job.", stored["log"]) self.assertNotIn("Watchlist download state synced.", stored["log"]) class WatchlistRefreshAllTests(unittest.TestCase): def test_refresh_all_iterates_only_watching_and_planned_show_ids(self): store = object.__new__(APP.WatchlistStore) calls = [] store.all_show_ids = lambda categories=None: ["show-a", "show-b"] store.refresh = lambda show_id, include_thumbnail=True: calls.append((show_id, include_thumbnail)) or {"show_id": show_id} result = APP.WatchlistStore.refresh_all(store) self.assertEqual(calls, [("show-a", False), ("show-b", False)]) self.assertEqual(result["count"], 2) self.assertEqual([item["show_id"] for item in result["items"]], ["show-a", "show-b"]) def test_all_show_ids_can_filter_by_category(self): with APP.WATCHLIST._connect() as conn: conn.execute("DELETE FROM watchlist") now = APP.now_iso() items = [ {"show_id": "show-w", "title": "Watching", "category": "watching"}, {"show_id": "show-p", "title": "Planned", "category": "planned"}, {"show_id": "show-f", "title": "Finished", "category": "finished"}, {"show_id": "show-d", "title": "Dropped", "category": "dropped"}, ] with APP.WATCHLIST.lock, APP.WATCHLIST._connect() as conn: for item in items: APP.WATCHLIST._upsert_conn( conn, { **APP.WATCHLIST._auto_download_defaults(item["title"]), "show_id": item["show_id"], "title": item["title"], "category": item["category"], "downloaded": False, "status": "tracked", "status_message": "seeded", "created_at": now, "updated_at": now, "last_checked": None, "sub_count": 0, "dub_count": 0, "expected_count": 0, "airing_status": None, "sub_latest_episode": None, "dub_latest_episode": None, "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": None, "anidb_title": None, "thumbnail_path": None, "thumbnail_checked_at": None, }, ) filtered = APP.WATCHLIST.all_show_ids(categories=("watching", "planned")) self.assertEqual(filtered, ["show-p", "show-w"]) class WatchlistRefreshJobTests(unittest.TestCase): def setUp(self): with APP.WatchlistRefreshJobs(APP.WATCHLIST, start_worker=False)._connect() as conn: conn.execute("DELETE FROM watchlist_refresh_jobs") def test_refresh_job_tracks_completion_and_errors(self): class FakeStore: def all_show_ids(self, categories=None): return ["show-a", "show-b"] def refresh(self, show_id, include_thumbnail=True): self.last_include_thumbnail = include_thumbnail if show_id == "show-a": return {"title": "Show A", "status": "updated"} return {"title": "Show B", "status": "error"} service = APP.WatchlistRefreshJobs(FakeStore(), start_worker=False) started = service.start() self.assertTrue(started["created"]) job = service._next_pending() self.assertIsNotNone(job) service._run_job(job) status = service.status()["job"] self.assertIsNotNone(status) self.assertEqual(status["status"], "done") self.assertEqual(status["completed"], 2) self.assertEqual(status["errors"], 1) self.assertFalse(service.store.last_include_thumbnail) def test_refresh_job_skips_removed_items_without_thumbnail_warmup(self): calls = [] class FakeStore: def all_show_ids(self, categories=None): return ["show-a", "show-b"] def refresh(self, show_id, include_thumbnail=True): calls.append((show_id, include_thumbnail)) if show_id == "show-a": return None return {"title": "Show B", "status": "updated"} service = APP.WatchlistRefreshJobs(FakeStore(), start_worker=False) started = service.start() self.assertTrue(started["created"]) job = service._next_pending() self.assertIsNotNone(job) service._run_job(job) status = service.status()["job"] self.assertEqual(status["status"], "done") self.assertEqual(status["completed"], 2) self.assertEqual(status["errors"], 0) self.assertCountEqual(calls, [("show-a", False), ("show-b", False)]) def test_refresh_job_marks_interrupted_when_shutdown_requested(self): service = APP.WatchlistRefreshJobs(mock.Mock(), start_worker=False) job = { "id": "refresh-1", "status": "running", "created_at": APP.now_iso(), "updated_at": APP.now_iso(), "completed": 0, "errors": 0, "current_title": None, "message": "", } service.store.all_show_ids = lambda categories=None: ["show-a"] service.store.refresh = mock.Mock(return_value={"title": "Show A", "status": "updated"}) service.stop_event.set() service._run_job(job) self.assertEqual(job["status"], "interrupted") self.assertIn("interrupted", job["message"].lower()) def test_refresh_start_reuses_pending_job(self): class FakeStore: def all_show_ids(self, categories=None): return [] def refresh(self, _show_id, include_thumbnail=True): return {"title": "Unused", "status": "updated"} service = APP.WatchlistRefreshJobs(FakeStore(), start_worker=False) first = service.start() second = service.start() self.assertTrue(first["created"]) self.assertFalse(second["created"]) self.assertEqual(second["job"]["id"], first["job"]["id"]) self.assertEqual(second["job"]["status"], "pending") def test_scheduled_refresh_logs_queue_and_completion(self): class FakeStore: def all_show_ids(self, categories=None): return ["show-a"] def refresh(self, _show_id, include_thumbnail=True): return {"title": "Show A", "status": "updated"} service = APP.WatchlistRefreshJobs(FakeStore(), start_worker=False) with mock.patch("builtins.print") as print_mock: started = service.start(source="scheduled") job = service._next_pending() service._run_job(job) self.assertTrue(started["created"]) logged = "\n".join(call.args[0] for call in print_mock.call_args_list if call.args) self.assertIn("Scheduled refresh queued", logged) self.assertIn("Scheduled refresh started", logged) self.assertIn("Scheduled refresh finished successfully", logged) def test_refresh_history_marks_pending_jobs_interrupted_after_restart(self): class FakeStore: def all_show_ids(self, categories=None): return [] def refresh(self, _show_id, include_thumbnail=True): return {"title": "Unused", "status": "updated"} service = APP.WatchlistRefreshJobs(FakeStore(), start_worker=False) started = service.start() restarted = APP.WatchlistRefreshJobs(FakeStore(), start_worker=False) status = restarted.status()["job"] self.assertEqual(status["id"], started["job"]["id"]) self.assertEqual(status["status"], "interrupted") self.assertIn("restart", status["message"].lower()) class WatchlistAlternativeTitleTests(unittest.TestCase): def setUp(self): APP.reset_runtime(wait=True) APP.initialize_runtime(start_workers=False) self.watchlist = APP.WATCHLIST with self.watchlist._connect() as conn: conn.execute("DELETE FROM watchlist") def test_watchlist_schema_includes_alternative_title_column(self): with self.watchlist._connect() as conn: columns = {row["name"] for row in conn.execute("PRAGMA table_info(watchlist)").fetchall()} self.assertIn("alternative_titles_json", columns) def test_refresh_caches_english_anidb_alternative_titles(self): self.watchlist.add({"show_id": "show-1", "title": "Attack on Titan", "category": "watching"}) with mock.patch.object( APP, "fetch_show_episode_snapshot", return_value={"show_id": "show-1", "title": "Attack on Titan", "episodes": {"sub": ["1", "2"], "dub": ["1"]}}, ), mock.patch.object( APP, "resolve_animeschedule_anime", return_value={"route": "attack-on-titan", "title": "Attack on Titan", "episodes": 2, "status": "Finished"}, ), mock.patch.object( APP, "resolve_anidb_aid", return_value={"aid": "16498", "title": "Attack on Titan"}, ), mock.patch.object( APP, "get_cached_anidb_titles", return_value=[ {"aid": "16498", "title": "Attack on Titan", "type": "official", "lang": "en", "normalized_title": "attack on titan"}, {"aid": "16498", "title": "AoT", "type": "syn", "lang": "en", "normalized_title": "aot"}, {"aid": "16498", "title": "Shingeki no Kyojin", "type": "official", "lang": "x-jat", "normalized_title": "shingeki no kyojin"}, {"aid": "16498", "title": "Attack Titan", "type": "short", "lang": "en", "normalized_title": "attack titan"}, ], ): item = self.watchlist.refresh("show-1", include_thumbnail=False) self.assertEqual(item["anidb_aid"], "16498") self.assertEqual( item["alternative_titles"], [ {"label": "English", "value": "AoT"}, {"label": "English", "value": "Attack Titan"}, ], ) self.assertTrue(item["has_alternative_titles"]) class ThumbnailEventRecoveryTests(unittest.TestCase): def test_ensure_thumbnails_skips_show_removed_during_batch(self): store = object.__new__(APP.WatchlistStore) item = { "show_id": "show-1", "thumbnail_ready": False, } store.get = mock.Mock(side_effect=[item, KeyError("Anime not found in watchlist.")]) store.ensure_thumbnail = mock.Mock(return_value=None) result = APP.WatchlistStore.ensure_thumbnails(store, ["show-1"]) self.assertEqual(result, {"items": []}) store.ensure_thumbnail.assert_called_once_with("show-1", force=False) def test_ensure_thumbnail_waiter_returns_none_after_show_is_removed(self): store = object.__new__(APP.WatchlistStore) store.lock = threading.RLock() inflight = threading.Event() inflight.set() store.thumbnail_events = {"show-1": inflight} store.has_show = lambda _show_id: False store.get = mock.Mock(return_value={ "show_id": "show-1", "title": "Show One", "thumbnail_path": None, "thumbnail_checked_at": None, "thumbnail_ready": False, "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": None, "anidb_title": None, }) result = APP.WatchlistStore.ensure_thumbnail(store, "show-1") self.assertIsNone(result) self.assertEqual(store.get.call_count, 1) def test_ensure_thumbnail_clears_inflight_event_after_update_failure(self): store = object.__new__(APP.WatchlistStore) store.lock = threading.RLock() store.thumbnail_events = {} store.get = lambda _show_id: { "show_id": "show-1", "title": "Show One", "thumbnail_path": None, "thumbnail_checked_at": None, "thumbnail_ready": False, "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": None, "anidb_title": None, } class BrokenConn: def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def execute(self, _sql, _params): raise RuntimeError("db write failed") store._connect = lambda: BrokenConn() with mock.patch.object(APP, "fetch_animeschedule_cover", return_value={"route": "route-1", "title": "Show One", "url": "https://example.com/cover.jpg"}), mock.patch.object( APP, "cache_thumbnail_image", return_value="show-1.jpg" ): with self.assertRaises(RuntimeError): APP.WatchlistStore.ensure_thumbnail(store, "show-1", force=False) self.assertEqual(store.thumbnail_events, {}) def test_ensure_thumbnail_records_failed_attempt_time_for_retry_backoff(self): now = APP.now_iso() item = { "show_id": "show-thumb-1", "title": "Broken Cover Show", "category": "watching", "downloaded": False, "status": "tracked", "status_message": "Waiting for thumbnail.", "created_at": now, "updated_at": now, "last_checked": None, "sub_count": 0, "dub_count": 0, "expected_count": 0, "airing_status": None, "sub_latest_episode": None, "dub_latest_episode": None, "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": None, "anidb_title": None, "media_type": "tv", "thumbnail_path": None, "thumbnail_checked_at": None, } with APP.WATCHLIST.lock, APP.WATCHLIST._connect() as conn: APP.WATCHLIST._upsert_conn(conn, item) with mock.patch.object(APP, "fetch_animeschedule_cover", side_effect=RuntimeError("lookup failed")) as fetch_cover, mock.patch.object( APP, "fetch_anidb_cover", side_effect=RuntimeError("fallback failed") ) as fetch_fallback: result = APP.WATCHLIST.ensure_thumbnail("show-thumb-1") self.assertIsNone(result) self.assertEqual(fetch_cover.call_count, 1) self.assertEqual(fetch_fallback.call_count, 1) failed_item = APP.WATCHLIST.get("show-thumb-1") self.assertTrue(failed_item["thumbnail_checked_at"]) self.assertFalse(APP.thumbnail_retry_due(failed_item["thumbnail_checked_at"])) with mock.patch.object(APP, "fetch_animeschedule_cover", side_effect=AssertionError("retry window should suppress another lookup")), mock.patch.object( APP, "fetch_anidb_cover", side_effect=AssertionError("retry window should suppress fallback lookup") ): second = APP.WATCHLIST.ensure_thumbnail("show-thumb-1") self.assertIsNone(second) def test_ensure_thumbnail_falls_back_from_stale_animeschedule_route(self): now = APP.now_iso() item = { "show_id": "show-thumb-2", "title": "Route Fallback Show", "category": "watching", "downloaded": False, "status": "tracked", "status_message": "Waiting for thumbnail.", "created_at": now, "updated_at": now, "last_checked": None, "sub_count": 0, "dub_count": 0, "expected_count": 0, "airing_status": None, "sub_latest_episode": None, "dub_latest_episode": None, "animeschedule_route": "stale-route", "animeschedule_title": "Route Fallback Show", "anidb_aid": None, "anidb_title": None, "thumbnail_path": None, "thumbnail_checked_at": None, } with APP.WATCHLIST.lock, APP.WATCHLIST._connect() as conn: APP.WATCHLIST._upsert_conn(conn, item) with mock.patch.object(APP, "fetch_animeschedule_cover_by_route", side_effect=RuntimeError("stale route")), mock.patch.object( APP, "fetch_animeschedule_cover", return_value={"route": "fresh-route", "title": "Route Fallback Show", "url": "https://example.com/cover.jpg"} ) as title_lookup, mock.patch.object(APP, "cache_thumbnail_image", return_value="show-thumb-2.jpg"): result = APP.WATCHLIST.ensure_thumbnail("show-thumb-2") title_lookup.assert_called_once_with("Route Fallback Show") self.assertEqual(result, APP.THUMBNAIL_ROOT / "show-thumb-2.jpg") refreshed = APP.WATCHLIST.get("show-thumb-2") self.assertEqual(refreshed["animeschedule_route"], "fresh-route") def test_ensure_thumbnail_falls_back_from_stale_anidb_aid(self): now = APP.now_iso() item = { "show_id": "show-thumb-3", "title": "AniDB Fallback Show", "category": "watching", "downloaded": False, "status": "tracked", "status_message": "Waiting for thumbnail.", "created_at": now, "updated_at": now, "last_checked": None, "sub_count": 0, "dub_count": 0, "expected_count": 0, "airing_status": None, "sub_latest_episode": None, "dub_latest_episode": None, "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": "12345", "anidb_title": "AniDB Fallback Show", "thumbnail_path": None, "thumbnail_checked_at": None, } with APP.WATCHLIST.lock, APP.WATCHLIST._connect() as conn: APP.WATCHLIST._upsert_conn(conn, item) with mock.patch.object(APP, "fetch_animeschedule_cover", side_effect=RuntimeError("animeschedule failed")), mock.patch.object( APP, "fetch_anidb_cover_by_aid", side_effect=RuntimeError("stale aid") ), mock.patch.object( APP, "fetch_anidb_cover", return_value={"aid": "67890", "title": "AniDB Fallback Show", "url": "https://example.com/cover.jpg"} ) as title_lookup, mock.patch.object(APP, "cache_thumbnail_image", return_value="show-thumb-3.jpg"): result = APP.WATCHLIST.ensure_thumbnail("show-thumb-3") title_lookup.assert_called_once_with("AniDB Fallback Show") self.assertEqual(result, APP.THUMBNAIL_ROOT / "show-thumb-3.jpg") refreshed = APP.WATCHLIST.get("show-thumb-3") self.assertEqual(refreshed["anidb_aid"], "67890") class WatchlistCompletionTests(unittest.TestCase): def setUp(self): with APP.WATCHLIST._connect() as conn: conn.execute("DELETE FROM watchlist") def seed_watchlist_item(self, **overrides): now = APP.now_iso() item = { "show_id": "show-1", "title": "Example Show", "category": "watching", "downloaded": False, "status": "tracked", "status_message": "Waiting for the first manual refresh.", "created_at": now, "updated_at": now, "last_checked": None, "sub_count": 0, "dub_count": 0, "expected_count": 0, "airing_status": None, "sub_latest_episode": None, "dub_latest_episode": None, "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": None, "anidb_title": None, "thumbnail_path": None, "thumbnail_checked_at": None, } item.update(overrides) with APP.WATCHLIST.lock, APP.WATCHLIST._connect() as conn: APP.WATCHLIST._upsert_conn(conn, item) def test_watchlist_status_message_marks_ready_modes(self): message = APP.watchlist_status_message(12, 10, expected_count=12, airing_status="finished") self.assertIn("Sub: 12/12 available.", message) self.assertIn("Dub: 10/12 available.", message) self.assertIn("AnimeSchedule: Finished.", message) self.assertIn("Sub ready.", message) self.assertNotIn("Dub ready.", message) def test_refresh_persists_expected_episode_total_and_ready_flags(self): self.seed_watchlist_item() snapshot = { "show_id": "show-1", "title": "Example Show", "episodes": { "sub": [str(index) for index in range(1, 13)], "dub": [str(index) for index in range(1, 13)], }, } schedule = { "route": "example-show", "title": "Example Show", "episodes": 12, "status": "Finished", } with mock.patch.object(APP, "fetch_show_episode_snapshot", return_value=snapshot), mock.patch.object( APP, "resolve_animeschedule_anime", return_value=schedule ), mock.patch.object(APP.WatchlistStore, "ensure_thumbnail", return_value=None): item = APP.WATCHLIST.refresh("show-1") self.assertEqual(item["expected_count"], 12) self.assertEqual(item["airing_status"], "Finished") self.assertTrue(item["sub_complete"]) self.assertTrue(item["dub_complete"]) self.assertEqual(item["sub_progress"], "12/12") self.assertEqual(item["dub_progress"], "12/12") self.assertEqual(item["animeschedule_route"], "example-show") self.assertIn("Sub ready.", item["status_message"]) self.assertIn("Dub ready.", item["status_message"]) def test_list_filters_by_category_and_reports_category_counts(self): self.seed_watchlist_item(show_id="show-1", title="Watching Show", category="watching", status="queued") self.seed_watchlist_item(show_id="show-2", title="Finished Show", category="finished", downloaded=True) listing = APP.WATCHLIST.list(category="watching") self.assertEqual(listing["active_category"], "watching") self.assertEqual(listing["summary"]["shows"], 2) self.assertEqual(listing["categories"]["watching"], 1) self.assertEqual(listing["categories"]["finished"], 1) self.assertEqual(listing["queued_count"], 1) self.assertEqual([item["show_id"] for item in listing["items"]], ["show-1"]) def test_list_clamps_out_of_range_page_before_querying_rows(self): self.seed_watchlist_item(show_id="show-1", title="Alpha Show", category="watching") self.seed_watchlist_item(show_id="show-2", title="Beta Show", category="watching") listing = APP.WATCHLIST.list(page=3, per_page=1, category="watching") self.assertEqual(listing["page"], 2) self.assertEqual(listing["pages"], 2) self.assertEqual([item["show_id"] for item in listing["items"]], ["show-2"]) def test_queue_refresh_marks_item_queued_without_sync_refresh(self): self.seed_watchlist_item(show_id="show-11", title="Refresh Show", category="watching", status="updated") with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)) as schedule_refresh, mock.patch.object( APP.WATCHLIST, "refresh", side_effect=AssertionError("queue_refresh should not refresh synchronously") ): item = APP.WATCHLIST.queue_refresh("show-11", status_message="Queued from test.") schedule_refresh.assert_called_once_with("show-11", source="manual") self.assertEqual(item["status"], "queued") self.assertEqual(item["status_message"], "Queued from test.") def test_mark_downloaded_keeps_existing_category_when_selected_mode_is_incomplete(self): self.seed_watchlist_item( show_id="show-9", title="Queue Show", category="planned", downloaded=False, expected_count=12, airing_status="Finished", dub_count=8, ) with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)), mock.patch.object( APP.WATCHLIST, "refresh", side_effect=AssertionError("mark_downloaded should not refresh synchronously") ), mock.patch.object(APP, "episode_list", return_value=["1", "2", "3", "4", "5", "6", "7", "8"]): item = APP.WATCHLIST.mark_downloaded("show-9", title="Queue Show", mode="dub", episodes="2-3") self.assertEqual(item["category"], "planned") self.assertTrue(item["downloaded"]) self.assertEqual(item["finished_badge"], "") self.assertEqual(item["status"], "queued") self.assertEqual(item["downloaded_dub_episodes"], ["2", "3"]) def test_mark_downloaded_moves_existing_category_to_finished_when_selected_mode_is_complete(self): self.seed_watchlist_item( show_id="show-9c", title="Queue Show Complete", category="watching", downloaded=False, expected_count=3, airing_status="Finished", dub_count=3, ) with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)), mock.patch.object( APP.WATCHLIST, "refresh", side_effect=AssertionError("mark_downloaded should not refresh synchronously") ), mock.patch.object(APP, "episode_list", return_value=["1", "2", "3"]): item = APP.WATCHLIST.mark_downloaded("show-9c", title="Queue Show Complete", mode="dub", episodes="1-3") self.assertEqual(item["category"], "finished") self.assertTrue(item["downloaded"]) self.assertEqual(item["finished_badge"], "downloaded") self.assertEqual(item["downloaded_dub_episodes"], ["1", "2", "3"]) def test_mark_downloaded_creates_missing_show_in_finished_category(self): with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)), mock.patch.object( APP.WATCHLIST, "refresh", side_effect=AssertionError("mark_downloaded should not refresh synchronously") ): item = APP.WATCHLIST.mark_downloaded("show-10", title="Downloaded Show") self.assertEqual(item["category"], "finished") self.assertTrue(item["downloaded"]) self.assertEqual(item["finished_badge"], "downloaded") self.assertEqual(item["status"], "queued") def test_mark_downloaded_does_not_recreate_removed_show(self): self.seed_watchlist_item(show_id="show-removed", title="Removed Show", category="watching") APP.WATCHLIST.remove("show-removed") with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=AssertionError("removed show should not be requeued")), mock.patch.object( APP.WATCHLIST, "refresh", side_effect=AssertionError("mark_downloaded should not refresh synchronously") ): item = APP.WATCHLIST.mark_downloaded("show-removed", title="Removed Show", mode="sub", episodes="1") self.assertIsNone(item) self.assertFalse(APP.WATCHLIST.has_show("show-removed")) def test_mark_downloaded_reconciles_unique_title_match_to_new_show_id(self): self.seed_watchlist_item( show_id="legacy-show-9", title="Queue Show", category="watching", downloaded=False, expected_count=3, airing_status="Finished", sub_count=3, ) with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)), mock.patch.object( APP.WATCHLIST, "refresh", side_effect=AssertionError("mark_downloaded should not refresh synchronously") ), mock.patch.object(APP, "episode_list", return_value=["1", "2", "3"]): item = APP.WATCHLIST.mark_downloaded("resolved-show-9", title="Queue Show", mode="sub", episodes="1-3") self.assertEqual(item["show_id"], "resolved-show-9") self.assertEqual(item["category"], "finished") self.assertTrue(item["downloaded"]) self.assertEqual(item["finished_badge"], "downloaded") self.assertEqual(item["status"], "queued") self.assertFalse(APP.WATCHLIST.has_show("legacy-show-9")) self.assertEqual(APP.WATCHLIST.all_show_ids(), ["resolved-show-9"]) def test_add_queues_refresh_without_sync_refresh(self): with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)) as schedule_refresh, mock.patch.object( APP.WATCHLIST, "refresh", side_effect=AssertionError("add should not refresh synchronously") ): result = APP.WATCHLIST.add({"show_id": "show-22", "title": "Queued Show", "category": "watching"}) schedule_refresh.assert_called_once_with("show-22", source="initial") self.assertTrue(result["created"]) self.assertEqual(result["item"]["status"], "queued") self.assertIn("Refresh queued", result["message"]) def test_add_uses_requested_auto_download_series(self): with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)): result = APP.WATCHLIST.add( {"show_id": "show-23", "title": "Queued Show", "category": "watching", "auto_download_series": "4"} ) self.assertEqual(result["item"]["auto_download_series"], "4") def test_add_uses_requested_auto_download_offset(self): with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)): result = APP.WATCHLIST.add( {"show_id": "show-23b", "title": "Queued Show", "category": "watching", "auto_download_offset": "13"} ) self.assertEqual(result["item"]["auto_download_offset"], "13") def test_add_uses_requested_media_type(self): with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id, source="manual": APP.WATCHLIST.get(show_id)): result = APP.WATCHLIST.add( {"show_id": "show-24", "title": "Queued Movie", "category": "watching", "media_type": "movie"} ) self.assertEqual(result["item"]["media_type"], "movie") self.assertEqual(result["item"]["media_type_label"], "Movie") def test_refresh_uses_bounded_request_timeout(self): self.seed_watchlist_item(show_id="show-1", title="Example Show") snapshot = { "show_id": "show-1", "title": "Example Show", "episodes": {"sub": ["1"], "dub": []}, } schedule = { "route": "example-show", "title": "Example Show", "episodes": 12, "status": "Finished", } with mock.patch.object(APP, "fetch_show_episode_snapshot", return_value=snapshot) as fetch_snapshot, mock.patch.object( APP, "resolve_animeschedule_anime", return_value=schedule ) as resolve_schedule, mock.patch.object(APP.WatchlistStore, "ensure_thumbnail", return_value=None): APP.WATCHLIST.refresh("show-1") fetch_snapshot.assert_called_once_with("show-1", timeout=APP.WATCHLIST_REFRESH_REQUEST_TIMEOUT_SECONDS) resolve_schedule.assert_called_once_with( "Example Show", route=None, timeout=APP.WATCHLIST_REFRESH_REQUEST_TIMEOUT_SECONDS, ) def test_sync_downloaded_job_to_watchlist_resolves_missing_show_id_from_search(self): job = {"query": "Queue Show", "mode": "sub", "result_index": 1, "title": "Queue Show"} with mock.patch.object(APP, "search_anime", return_value=[{"id": "show-42", "title": "Queue Show"}]), mock.patch.object( APP.WATCHLIST, "mark_downloaded", return_value={"show_id": "show-42"} ) as mark_downloaded: APP.sync_downloaded_job_to_watchlist(job) mark_downloaded.assert_called_once_with("show-42", title="Queue Show", mode="sub", episodes=None) def test_prepare_download_job_persists_resolved_show_id(self): job = {"query": "Queue Show", "mode": "sub", "result_index": 1, "title": "Queue Show", "show_id": ""} with mock.patch.object(APP, "search_anime", return_value=[{"id": "show-42", "title": "Queue Show"}]): prepared = APP.prepare_download_job(dict(job)) self.assertEqual(prepared["show_id"], "show-42") self.assertEqual(prepared["title"], "Queue Show") def test_prepare_download_job_accepts_equivalent_season_notation(self): job = {"query": "Queue Show 2nd Season", "mode": "sub", "result_index": 1, "title": "Queue Show 2nd Season", "show_id": ""} with mock.patch.object(APP, "search_anime", return_value=[{"id": "show-42", "title": "Queue Show Season 2"}]): prepared = APP.prepare_download_job(dict(job)) self.assertEqual(prepared["show_id"], "show-42") self.assertEqual(prepared["title"], "Queue Show Season 2") def test_prepare_download_job_resolves_unique_match_without_result_index(self): job = {"query": "Queue Show", "mode": "sub", "title": "Queue Show", "show_id": ""} with mock.patch.object( APP, "search_anime", return_value=[ {"id": "show-42", "title": "Queue Show"}, {"id": "show-99", "title": "Different Show"}, ], ): prepared = APP.prepare_download_job(dict(job)) self.assertEqual(prepared["show_id"], "show-42") self.assertEqual(prepared["title"], "Queue Show") def test_prepare_download_job_skips_ambiguous_match_without_result_index(self): job = {"query": "Queue Show", "mode": "sub", "title": "Queue Show", "show_id": ""} with mock.patch.object( APP, "search_anime", return_value=[ {"id": "show-42", "title": "Queue Show"}, {"id": "show-99", "title": "Queue Show (2024)"}, ], ): prepared = APP.prepare_download_job(dict(job)) self.assertEqual(prepared["show_id"], "") def test_command_for_job_omits_result_switch_when_result_index_missing(self): job = APP.app_support.build_job( { "query": "Queue Show", "title": "Queue Show", "mode": "dub", "quality": "best", "episodes": "1-10", "download_dir": "/tmp/example", "season": "1", }, {"mode": "sub", "quality": "best", "download_dir": "/tmp/example"}, ) self.assertNotIn("-S", APP.app_support.command_for_job(job)) def test_command_for_job_ignores_legacy_result_index(self): command = APP.app_support.command_for_job( { "query": "Queue Show", "quality": "best", "episodes": "1-10", "mode": "dub", "result_index": 2, } ) self.assertNotIn("-S", command) def test_download_watchlist_item_queues_full_episode_range_for_selected_mode(self): item = { "show_id": "show-42", "title": "Queue Show", "auto_download_name": "Queue Show Library", "auto_download_series": "4", "auto_download_quality": "720", "media_type": "movie", } job = {"id": "job-1", "show_id": "show-42", "episodes": "1-12", "mode": "dub"} with mock.patch.object(APP, "ensure_runtime", return_value={"watchlist": APP.WATCHLIST, "download_queue": mock.Mock(), "config": {"quality": "best", "download_dir": "/tmp/downloads"}}), mock.patch.object( APP.WATCHLIST, "get", return_value=item ), mock.patch.object(APP, "episode_list", return_value=["1", "2", "12"]): runtime = APP.ensure_runtime() runtime["download_queue"].add.return_value = job result = APP.download_watchlist_item("show-42", "dub") runtime["download_queue"].add.assert_called_once_with( { "show_id": "show-42", "query": "Queue Show", "title": "Queue Show", "anime_name": "Queue Show Library", "media_type": "movie", "season": "4", "episode_offset": None, "mode": "dub", "quality": "720", "episodes": "1-12", "download_dir": "/tmp/downloads", } ) self.assertEqual(result["job"], job) self.assertIn("Queued Queue Show for dub download.", result["message"]) def test_remove_from_watchlist_detaches_existing_queue_jobs(self): runtime = APP.ensure_runtime() APP.WATCHLIST.add({"show_id": "show-queue-remove", "title": "Queue Remove Show"}) created = runtime["download_queue"].add( { "show_id": "show-queue-remove", "query": "Queue Remove Show", "title": "Queue Remove Show", "anime_name": "Queue Remove Show", "mode": "sub", "quality": "best", "episodes": "1-2", "download_dir": "/tmp/example", "season": "1", } ) APP.remove_from_watchlist("show-queue-remove") detached = runtime["download_queue"].get(created["id"]) self.assertEqual(detached["show_id"], "") self.assertTrue(detached["skip_watchlist_sync"]) def test_download_watchlist_item_does_not_require_search_match(self): item = { "show_id": "show-42", "title": "Queue Show", "auto_download_name": "Queue Show", "auto_download_series": "1", "media_type": "tv", } runtime = { "watchlist": APP.WATCHLIST, "download_queue": mock.Mock(), "config": {"quality": "best", "download_dir": "/tmp/downloads"}, } runtime["download_queue"].add.return_value = {"id": "job-1"} with mock.patch.object(APP, "ensure_runtime", return_value=runtime), mock.patch.object( APP.WATCHLIST, "get", return_value=item ), mock.patch.object(APP, "search_anime", side_effect=AssertionError("search should not be required")), mock.patch.object( APP, "episode_list", return_value=["1", "2"] ): result = APP.download_watchlist_item("show-42", "dub") self.assertEqual(result["job"]["id"], "job-1") class JellyfinSyncTests(unittest.TestCase): def setUp(self): with APP.WATCHLIST._connect() as conn: conn.execute("DELETE FROM watchlist") def seed_watchlist_item(self, **overrides): now = APP.now_iso() item = { "show_id": "show-jelly-1", "title": "Jelly Show", "category": "finished", "downloaded": True, "status": "updated", "status_message": "Finished.", "created_at": now, "updated_at": now, "last_checked": now, "sub_count": 12, "dub_count": 12, "expected_count": 12, "airing_status": "Finished", "sub_latest_episode": "12", "dub_latest_episode": "12", "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": None, "anidb_title": None, "media_type": "tv", "auto_download_mode": "dub", "auto_download_name": "Jelly Show", "auto_download_series": "1", "downloaded_sub_episodes_json": None, "downloaded_dub_episodes_json": APP.encode_episode_values([str(index) for index in range(1, 13)]), "thumbnail_path": None, "thumbnail_checked_at": None, } item.update(overrides) with APP.WATCHLIST.lock, APP.WATCHLIST._connect() as conn: APP.WATCHLIST._upsert_conn(conn, item) def test_run_jellyfin_sync_moves_eligible_tv_library(self): self.seed_watchlist_item() with tempfile.TemporaryDirectory() as temp_root: download_root = Path(temp_root) / "downloads" source_file = download_root / "tv" / "Jelly Show" / "Season 01" / "Jelly Show - S01E01.mp4" source_file.parent.mkdir(parents=True, exist_ok=True) source_file.write_bytes(b"episode-data") jellyfin_tv = Path(temp_root) / "jellyfin-tv" result = APP.run_jellyfin_sync( { "download_dir": str(download_root), "jellyfin_sync_enabled": True, "jellyfin_tv_dir": str(jellyfin_tv), "jellyfin_movie_dir": "", } ) self.assertEqual(result["moved"], 1) target_file = jellyfin_tv / "Jelly Show" / "Season 01" / "Jelly Show - S01E01.mp4" self.assertTrue(target_file.exists()) self.assertFalse((download_root / "tv" / "Jelly Show").exists()) def test_trigger_jellyfin_sync_for_item_sends_success_notification(self): self.seed_watchlist_item() item = APP.WATCHLIST.get("show-jelly-1") with tempfile.TemporaryDirectory() as temp_root: download_root = Path(temp_root) / "downloads" source_file = download_root / "tv" / "Jelly Show" / "Season 01" / "Jelly Show - S01E01.mp4" source_file.parent.mkdir(parents=True, exist_ok=True) source_file.write_bytes(b"episode-data") jellyfin_tv = Path(temp_root) / "jellyfin-tv" with mock.patch.object(APP, "send_discord_webhook_event") as send_webhook: result = APP.trigger_jellyfin_sync_for_item( item, automatic=True, config={ "download_dir": str(download_root), "jellyfin_sync_enabled": True, "jellyfin_tv_dir": str(jellyfin_tv), "jellyfin_movie_dir": "", "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["jellyfin_sync_success"], }, ) self.assertTrue(result["moved"]) send_webhook.assert_called_once() self.assertEqual(send_webhook.call_args.args[1], "jellyfin_sync_success") self.assertEqual(send_webhook.call_args.args[2]["title"], "Jelly Show") self.assertEqual(send_webhook.call_args.args[2]["source"], "automatic") def test_trigger_jellyfin_sync_for_item_sends_failure_notification(self): self.seed_watchlist_item() item = APP.WATCHLIST.get("show-jelly-1") with mock.patch.object(APP, "send_discord_webhook_event") as send_webhook: result = APP.trigger_jellyfin_sync_for_item( item, automatic=True, config={ "download_dir": "/tmp/example", "jellyfin_sync_enabled": True, "jellyfin_tv_dir": "", "jellyfin_movie_dir": "", "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["jellyfin_sync_failed"], }, ) self.assertFalse(result["moved"]) self.assertEqual(result["reason"], "target_missing") send_webhook.assert_called_once() self.assertEqual(send_webhook.call_args.args[1], "jellyfin_sync_failed") self.assertEqual(send_webhook.call_args.args[2]["reason"], "target_missing") def test_trigger_jellyfin_sync_for_item_skips_already_moved_notification(self): self.seed_watchlist_item(expected_count=1, dub_count=1, downloaded_dub_episodes_json=APP.encode_episode_values(["1"])) item = APP.WATCHLIST.get("show-jelly-1") with tempfile.TemporaryDirectory() as temp_root: jellyfin_tv = Path(temp_root) / "jellyfin-tv" / "Jelly Show" jellyfin_tv.mkdir(parents=True, exist_ok=True) (jellyfin_tv / "Season 01").mkdir(parents=True, exist_ok=True) (jellyfin_tv / "Season 01" / "Jelly Show - S01E01.mp4").write_bytes(b"episode-data") with mock.patch.object(APP, "send_discord_webhook_event") as send_webhook: result = APP.trigger_jellyfin_sync_for_item( item, automatic=False, config={ "download_dir": str(Path(temp_root) / "downloads"), "jellyfin_sync_enabled": True, "jellyfin_tv_dir": str(Path(temp_root) / "jellyfin-tv"), "jellyfin_movie_dir": "", "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["jellyfin_sync_failed", "jellyfin_sync_success"], }, ) self.assertFalse(result["moved"]) self.assertEqual(result["reason"], "already_moved") send_webhook.assert_not_called() def test_trigger_jellyfin_sync_for_item_flags_incomplete_target_when_source_missing(self): self.seed_watchlist_item(expected_count=12, dub_count=12) item = APP.WATCHLIST.get("show-jelly-1") with tempfile.TemporaryDirectory() as temp_root: jellyfin_tv = Path(temp_root) / "jellyfin-tv" / "Jelly Show" jellyfin_tv.mkdir(parents=True, exist_ok=True) with mock.patch.object(APP, "send_discord_webhook_event") as send_webhook: result = APP.trigger_jellyfin_sync_for_item( item, automatic=False, config={ "download_dir": str(Path(temp_root) / "downloads"), "jellyfin_sync_enabled": True, "jellyfin_tv_dir": str(Path(temp_root) / "jellyfin-tv"), "jellyfin_movie_dir": "", "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["jellyfin_sync_failed"], }, ) self.assertFalse(result["moved"]) self.assertEqual(result["reason"], "target_incomplete") send_webhook.assert_called_once() self.assertEqual(send_webhook.call_args.args[1], "jellyfin_sync_failed") def test_trigger_jellyfin_sync_for_item_skips_incomplete_series(self): self.seed_watchlist_item( expected_count=12, dub_count=8, dub_latest_episode="8", downloaded_dub_episodes_json=APP.encode_episode_values([str(index) for index in range(1, 9)]), ) item = APP.WATCHLIST.get("show-jelly-1") result = APP.trigger_jellyfin_sync_for_item( item, automatic=False, config={"download_dir": "/tmp/example", "jellyfin_tv_dir": "/tmp/jellyfin-tv", "jellyfin_movie_dir": ""}, ) self.assertFalse(result["moved"]) self.assertEqual(result["reason"], "episodes_unavailable") def test_trigger_jellyfin_sync_for_item_skips_when_dub_not_fully_downloaded(self): self.seed_watchlist_item( expected_count=12, dub_count=12, downloaded_dub_episodes_json=APP.encode_episode_values([str(index) for index in range(1, 9)]), ) item = APP.WATCHLIST.get("show-jelly-1") result = APP.trigger_jellyfin_sync_for_item( item, automatic=False, config={"download_dir": "/tmp/example", "jellyfin_tv_dir": "/tmp/jellyfin-tv", "jellyfin_movie_dir": ""}, ) self.assertFalse(result["moved"]) self.assertEqual(result["reason"], "episodes_missing") def test_sync_downloaded_job_to_watchlist_rejects_ambiguous_fallback_match(self): job = {"query": "Queue Show", "mode": "sub", "result_index": 1, "title": "Queue Show"} with mock.patch.object(APP, "search_anime", return_value=[{"id": "show-42", "title": "Different Anime"}]): with self.assertRaises(ValueError) as ctx: APP.sync_downloaded_job_to_watchlist(job) self.assertIn("title mismatch", str(ctx.exception).lower()) def test_sync_downloaded_job_to_watchlist_rejects_cross_season_match(self): job = {"query": "Queue Show Season 2", "mode": "sub", "result_index": 1, "title": "Queue Show Season 2"} with mock.patch.object(APP, "search_anime", return_value=[{"id": "show-42", "title": "Queue Show Season 1"}]): with self.assertRaises(ValueError) as ctx: APP.sync_downloaded_job_to_watchlist(job) self.assertIn("title mismatch", str(ctx.exception).lower()) class WatchlistQueuedRefreshRecoveryTests(unittest.TestCase): def setUp(self): with APP.WATCHLIST._connect() as conn: conn.execute("DELETE FROM watchlist") def seed_watchlist_item(self, **overrides): now = APP.now_iso() item = { "show_id": "show-1", "title": "Example Show", "category": "watching", "downloaded": False, "status": "tracked", "status_message": "Waiting for the first manual refresh.", "created_at": now, "updated_at": now, "last_checked": None, "sub_count": 0, "dub_count": 0, "expected_count": 0, "airing_status": None, "sub_latest_episode": None, "dub_latest_episode": None, "animeschedule_route": None, "animeschedule_title": None, "anidb_aid": None, "anidb_title": None, "thumbnail_path": None, "thumbnail_checked_at": None, } item.update(overrides) with APP.WATCHLIST.lock, APP.WATCHLIST._connect() as conn: APP.WATCHLIST._upsert_conn(conn, item) def test_restart_restores_queued_watchlist_refreshes(self): self.seed_watchlist_item( show_id="show-b", title="Beta Show", status="queued", status_message="Queued refresh", updated_at="2026-05-15T10:00:00+00:00", ) self.seed_watchlist_item( show_id="show-a", title="Alpha Show", status="queued", status_message="Queued refresh", updated_at="2026-05-15T09:00:00+00:00", ) self.seed_watchlist_item( show_id="show-c", title="Gamma Show", status="updated", status_message="Updated", updated_at="2026-05-15T11:00:00+00:00", ) restored = APP.WatchlistStore(start_worker=False) self.assertEqual(restored.refresh_pending, [("show-a", "manual"), ("show-b", "manual")]) self.assertEqual(restored.refresh_pending_set, {"show-a", "show-b"}) class WatchlistRefreshQueueTests(unittest.TestCase): def test_schedule_refresh_skips_show_already_in_flight(self): store = object.__new__(APP.WatchlistStore) store.lock = threading.RLock() store.thumbnail_events = {} store.refresh_pending = [] store.refresh_pending_set = set() store.refresh_inflight_set = {"show-1"} store.refresh_wakeup = threading.Event() store.refresh_stop_event = threading.Event() store.refresh_worker = None store.worker_enabled = False store.ensure_worker = lambda: None store.get = lambda show_id: {"show_id": show_id, "status": "queued"} item = APP.WatchlistStore.schedule_refresh(store, "show-1") self.assertEqual(item["show_id"], "show-1") self.assertEqual(store.refresh_pending, []) self.assertEqual(store.refresh_pending_set, set()) self.assertEqual(store.refresh_inflight_set, {"show-1"}) def test_refresh_loop_clears_inflight_marker_after_refresh(self): store = object.__new__(APP.WatchlistStore) store.lock = threading.RLock() store.thumbnail_events = {} store.refresh_pending = [("show-2", "manual")] store.refresh_pending_set = {"show-2"} store.refresh_inflight_set = set() store.refresh_wakeup = threading.Event() store.refresh_stop_event = threading.Event() store.refresh_worker = None store.worker_enabled = False calls = [] def fake_refresh(show_id, refresh_source="manual"): calls.append((show_id, refresh_source)) store.refresh_stop_event.set() store.refresh = fake_refresh APP.WatchlistStore._refresh_loop(store) self.assertEqual(calls, [("show-2", "manual")]) self.assertEqual(store.refresh_inflight_set, set()) self.assertEqual(store.refresh_pending, []) self.assertEqual(store.refresh_pending_set, set()) def test_remove_drops_pending_and_inflight_refresh_state(self): store = object.__new__(APP.WatchlistStore) store.lock = threading.RLock() thumbnail_event = threading.Event() store.thumbnail_events = {"show-3": thumbnail_event} store.refresh_pending = [("show-3", "manual"), ("show-4", "manual")] store.refresh_pending_set = {"show-3", "show-4"} store.refresh_inflight_set = {"show-3"} store._connect = APP.WATCHLIST._connect store.get = lambda show_id: {"show_id": show_id, "title": "Queued Show", "thumbnail_path": None} class DummyCursor: rowcount = 1 class DummyConn: def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def execute(self, _sql, _params): return DummyCursor() store._connect = lambda: DummyConn() result = APP.WatchlistStore.remove(store, "show-3") self.assertEqual(result["item"]["show_id"], "show-3") self.assertEqual(store.refresh_pending, [("show-4", "manual")]) self.assertEqual(store.refresh_pending_set, {"show-4"}) self.assertEqual(store.refresh_inflight_set, set()) self.assertTrue(thumbnail_event.is_set()) self.assertNotIn("show-3", store.thumbnail_events) class ConfigSnapshotTests(unittest.TestCase): def test_get_config_snapshot_returns_copy(self): original = APP.CONFIG APP.CONFIG = { "mode": "sub", "quality": "best", "download_dir": "/tmp/example", "watchlist_auto_refresh_enabled": False, "watchlist_auto_refresh_minutes": 60, "watchlist_refresh_delay_seconds": 5, "auto_download_enabled": False, "auto_download_mode": "dub", "auto_download_quality": "best", "discord_webhook_url": "", "discord_webhook_events": [], } try: snapshot = APP.get_config_snapshot() snapshot["mode"] = "dub" self.assertEqual(APP.CONFIG["mode"], "sub") finally: APP.CONFIG = original def test_normalize_config_includes_watchlist_schedule_defaults(self): config = APP.normalize_config({}) 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["auto_download_enabled"]) self.assertEqual(config["auto_download_mode"], "dub") self.assertEqual(config["auto_download_quality"], "best") self.assertFalse(config["jellyfin_sync_enabled"]) self.assertEqual(config["jellyfin_tv_dir"], "") self.assertEqual(config["jellyfin_movie_dir"], "") self.assertEqual(config["discord_webhook_url"], "") self.assertEqual(config["discord_webhook_events"], []) def test_normalize_config_parses_watchlist_schedule_fields(self): config = APP.normalize_config( { "watchlist_auto_refresh_enabled": "true", "watchlist_auto_refresh_minutes": "15", "watchlist_refresh_delay_seconds": "7", "auto_download_enabled": "true", "auto_download_mode": "sub", "auto_download_quality": "720p", "jellyfin_sync_enabled": "true", "jellyfin_tv_dir": "~/jellyfin/tv", "jellyfin_movie_dir": "~/jellyfin/movies", } ) self.assertTrue(config["watchlist_auto_refresh_enabled"]) self.assertEqual(config["watchlist_auto_refresh_minutes"], 15) self.assertEqual(config["watchlist_refresh_delay_seconds"], 7) self.assertTrue(config["auto_download_enabled"]) self.assertEqual(config["auto_download_mode"], "sub") self.assertEqual(config["auto_download_quality"], "720") self.assertTrue(config["jellyfin_sync_enabled"]) self.assertIn("jellyfin", config["jellyfin_tv_dir"]) self.assertIn("jellyfin", config["jellyfin_movie_dir"]) def test_normalize_config_filters_discord_webhook_events(self): config = APP.normalize_config( { "discord_webhook_url": " https://discord.example/webhook ", "discord_webhook_events": [ "watchlist_refresh_new_episodes", "runtime_error", "runtime_error", "nope", ], } ) self.assertEqual(config["discord_webhook_url"], "https://discord.example/webhook") self.assertEqual( config["discord_webhook_events"], ["watchlist_refresh_new_episodes", "runtime_error"], ) def test_test_discord_webhook_config_requires_url(self): with self.assertRaises(ValueError): APP.test_discord_webhook_config({"discord_webhook_url": ""}) def test_test_discord_webhook_config_sends_test_event(self): with mock.patch.object(APP, "send_discord_webhook_event") as send_webhook: result = APP.test_discord_webhook_config( { "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": [], } ) self.assertEqual(result["message"], "Discord webhook test notification sent.") send_webhook.assert_called_once() config, event_name = send_webhook.call_args.args[:2] self.assertEqual(config["discord_webhook_url"], "https://discord.example/webhook") self.assertEqual(event_name, "test") self.assertTrue(send_webhook.call_args.kwargs["force"]) class WatchlistAutoRefreshSchedulerTests(unittest.TestCase): def test_tick_waits_then_triggers_refresh(self): config = { "watchlist_auto_refresh_enabled": True, "watchlist_auto_refresh_minutes": 10, } start = mock.Mock(return_value={"created": True}) scheduler = queue_jobs.WatchlistAutoRefreshScheduler(lambda: config, start, start_worker=False) self.assertEqual(scheduler.tick(now_monotonic=100.0), 600) self.assertEqual(scheduler.tick(now_monotonic=699.0), 1) self.assertEqual(scheduler.tick(now_monotonic=700.0), 600) start.assert_called_once_with(source="scheduled") def test_refresh_job_waits_between_series_using_configured_delay(self): class FakeStore: def all_show_ids(self, categories=None): return ["show-a", "show-b", "show-c"] def refresh(self, show_id, include_thumbnail=True): return {"title": show_id, "status": "updated"} service = APP.WatchlistRefreshJobs( FakeStore(), config_getter=lambda: {"watchlist_refresh_delay_seconds": 5}, start_worker=False, ) job = service.start()["job"] job = service._next_pending() with mock.patch.object(service.stop_event, "wait", return_value=False) as wait_mock: service._run_job(job) self.assertEqual(wait_mock.call_count, 2) wait_mock.assert_called_with(5) def test_refresh_job_sends_new_episode_and_failure_notifications(self): class FakeStore: def all_show_ids(self, categories=None): return ["show-a", "show-b"] def refresh(self, show_id, include_thumbnail=True): if show_id == "show-a": return { "show_id": "show-a", "title": "Show A", "status": "updated", "had_new_episodes": True, "sub_count": 12, "dub_count": 4, "sub_delta": 1, "dub_delta": 0, "expected_count": 12, "airing_status": "Finished", "thumbnail_path": None, } return { "show_id": "show-b", "title": "Show B", "status": "error", "status_message": "Could not refresh episodes: boom", "had_new_episodes": False, } service = APP.WatchlistRefreshJobs( FakeStore(), config_getter=lambda: { "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": [ "watchlist_refresh_new_episodes", "watchlist_refresh_failed", ], }, start_worker=False, ) job = service.start()["job"] job = service._next_pending() with mock.patch.object(queue_jobs, "send_discord_webhook_event") as send_webhook: service._run_job(job) self.assertEqual(send_webhook.call_count, 2) self.assertEqual(send_webhook.call_args_list[0].args[1], "watchlist_refresh_new_episodes") self.assertEqual(send_webhook.call_args_list[0].args[2]["items"][0]["title"], "Show A") self.assertEqual(send_webhook.call_args_list[1].args[1], "watchlist_refresh_failed") self.assertEqual(send_webhook.call_args_list[1].args[2]["items"][0]["title"], "Show B") class DiscordWebhookFormattingTests(unittest.TestCase): def test_download_completed_webhook_sends_full_file_list_across_embeds(self): payloads = [] with mock.patch.object(APP.app_support, "_post_discord_webhook", side_effect=lambda url, payload, attachments=None, timeout=10: payloads.append(payload)): result = APP.app_support.send_discord_webhook_event( { "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["download_completed"], }, "download_completed", { "title": "Big Download", "mode": "sub", "episodes": "1-12", "category": "finished", "moved_files": [f"/tmp/downloads/Big Download/File {index:02d}.mp4" for index in range(1, 130)], }, ) self.assertTrue(result["sent"]) self.assertGreaterEqual(len(payloads), 1) all_descriptions = "\n".join( embed.get("description", "") for payload in payloads for embed in payload.get("embeds", []) ) self.assertIn("/tmp/downloads/Big Download/File 01.mp4", all_descriptions) self.assertIn("/tmp/downloads/Big Download/File 129.mp4", all_descriptions) def test_jellyfin_success_webhook_includes_moved_files(self): payloads = [] with mock.patch.object(APP.app_support, "_post_discord_webhook", side_effect=lambda url, payload, attachments=None, timeout=10: payloads.append(payload)): result = APP.app_support.send_discord_webhook_event( { "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["jellyfin_sync_success"], }, "jellyfin_sync_success", { "title": "Jelly Show", "media_type": "tv", "source": "automatic", "target": "/srv/jellyfin/Jelly Show", "moved_files": ["/srv/jellyfin/Jelly Show/Season 01/Jelly Show - S01E01.mp4"], }, ) self.assertTrue(result["sent"]) self.assertEqual(payloads[0]["content"], "ani-cli-web 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): config = { "watchlist_auto_refresh_enabled": True, "watchlist_auto_refresh_minutes": 10, } start = mock.Mock(return_value={"created": True}) scheduler = queue_jobs.WatchlistAutoRefreshScheduler(lambda: config, start, start_worker=False) self.assertEqual(scheduler.tick(now_monotonic=100.0), 600) config["watchlist_auto_refresh_minutes"] = 5 self.assertEqual(scheduler.tick(now_monotonic=150.0), 300) start.assert_not_called() def test_tick_returns_none_when_disabled(self): config = { "watchlist_auto_refresh_enabled": False, "watchlist_auto_refresh_minutes": 10, } start = mock.Mock(return_value={"created": True}) scheduler = queue_jobs.WatchlistAutoRefreshScheduler(lambda: config, start, start_worker=False) self.assertIsNone(scheduler.tick(now_monotonic=100.0)) start.assert_not_called() class AutoDownloadQueueTests(unittest.TestCase): def test_queue_watchlist_auto_download_skips_initial_refresh(self): item = { "show_id": "show-1", "title": "Example Show", "category": "watching", "previous_last_checked": None, "had_new_episodes": True, } with mock.patch.object( APP, "ensure_runtime", return_value={"config": {"auto_download_enabled": True}, "download_queue": mock.Mock()}, ): result = APP.queue_watchlist_auto_download(item, refresh_source="initial") self.assertFalse(result["queued"]) self.assertEqual(result["reason"], "source:initial") def test_queue_watchlist_auto_download_queues_only_missing_new_episodes(self): item = { "show_id": "show-42", "title": "Queue Show", "category": "watching", "previous_last_checked": "2026-05-25T10:00:00+00:00", "had_new_episodes": False, "previous_dub_count": 10, "previous_sub_count": 10, "dub_episode_values": ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12"], "sub_episode_values": ["1", "2", "3"], "auto_download_mode": "dub", "auto_download_quality": "best", "auto_download_name": 'Queue Show: Alt/Name', "auto_download_series": "3", "auto_download_offset": "13", "downloaded_dub_episodes": ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"], "downloaded_sub_episodes": [], } download_queue = mock.Mock() download_queue.covered_episodes.return_value = {"11"} download_queue.add.side_effect = [ {"id": "job-12", "episodes": "12"}, ] runtime = { "config": { "auto_download_enabled": True, "auto_download_mode": "dub", "auto_download_quality": "best", "download_dir": "/tmp/downloads", }, "download_queue": download_queue, } with mock.patch.object(APP, "ensure_runtime", return_value=runtime), mock.patch.object( APP, "search_anime", side_effect=AssertionError("search should not be required"), ): result = APP.queue_watchlist_auto_download(item, refresh_source="scheduled") self.assertTrue(result["queued"]) self.assertEqual(result["episodes"], ["12"]) download_queue.add.assert_called_once_with( { "show_id": "show-42", "query": "Queue Show", "title": "Queue Show", "anime_name": "Queue Show- Alt-Name", "media_type": "tv", "season": "3", "episode_offset": "13", "mode": "dub", "quality": "best", "episodes": "12", "download_dir": "/tmp/downloads", } ) def test_queue_watchlist_auto_download_can_queue_existing_backlog_after_initial_seed_refresh(self): item = { "show_id": "show-99", "title": "Backlog Show", "category": "watching", "previous_last_checked": "2026-05-25T10:00:00+00:00", "had_new_episodes": False, "dub_episode_values": ["1", "2", "3"], "sub_episode_values": [], "auto_download_mode": "dub", "auto_download_quality": "best", "auto_download_name": "Backlog Show", "auto_download_series": "1", "auto_download_offset": None, "downloaded_dub_episodes": [], "downloaded_sub_episodes": [], } download_queue = mock.Mock() download_queue.covered_episodes.return_value = set() download_queue.add.return_value = {"id": "job-backlog", "episodes": "1-3"} runtime = { "config": { "auto_download_enabled": True, "auto_download_mode": "dub", "auto_download_quality": "best", "download_dir": "/tmp/downloads", }, "download_queue": download_queue, } with mock.patch.object(APP, "ensure_runtime", return_value=runtime), mock.patch.object( APP, "search_anime", side_effect=AssertionError("search should not be required"), ): result = APP.queue_watchlist_auto_download(item, refresh_source="manual") self.assertTrue(result["queued"]) self.assertEqual(result["episodes"], ["1-3"]) download_queue.add.assert_called_once_with( { "show_id": "show-99", "query": "Backlog Show", "title": "Backlog Show", "anime_name": "Backlog Show", "media_type": "tv", "season": "1", "episode_offset": None, "mode": "dub", "quality": "best", "episodes": "1-3", "download_dir": "/tmp/downloads", } ) def test_queue_watchlist_auto_download_passes_episode_offset_to_jobs(self): item = { "show_id": "show-77", "title": "Split Season Show", "category": "watching", "previous_last_checked": "2026-05-25T10:00:00+00:00", "had_new_episodes": False, "dub_episode_values": ["1", "2"], "sub_episode_values": [], "auto_download_mode": "dub", "auto_download_quality": "best", "auto_download_name": "Split Season Show", "auto_download_series": "2", "auto_download_offset": "13", "downloaded_dub_episodes": [], "downloaded_sub_episodes": [], } download_queue = mock.Mock() download_queue.covered_episodes.return_value = set() download_queue.add.return_value = {"id": "job-split", "episodes": "1-2"} runtime = { "config": { "auto_download_enabled": True, "auto_download_mode": "dub", "auto_download_quality": "best", "download_dir": "/tmp/downloads", }, "download_queue": download_queue, } with mock.patch.object(APP, "ensure_runtime", return_value=runtime), mock.patch.object( APP, "search_anime", side_effect=AssertionError("search should not be required"), ): result = APP.queue_watchlist_auto_download(item, refresh_source="manual") self.assertTrue(result["queued"]) download_queue.add.assert_called_once_with( { "show_id": "show-77", "query": "Split Season Show", "title": "Split Season Show", "anime_name": "Split Season Show", "media_type": "tv", "season": "2", "episode_offset": "13", "mode": "dub", "quality": "best", "episodes": "1-2", "download_dir": "/tmp/downloads", } ) class StartupBehaviorTests(unittest.TestCase): def test_main_does_not_eagerly_initialize_runtime(self): server = mock.Mock() server.serve_forever.side_effect = KeyboardInterrupt() with mock.patch.object(APP, "initialize_runtime") as initialize_runtime, mock.patch.object( APP, "ThreadingHTTPServer", return_value=server ), mock.patch.object(APP, "shutdown_runtime") as shutdown_runtime: shutdown_runtime.return_value = True APP.main() initialize_runtime.assert_not_called() shutdown_runtime.assert_called_once_with(wait=True, cancel_active_downloads=True) server.serve_forever.assert_called_once() server.server_close.assert_called_once() def test_reset_runtime_wait_defaults_to_cancel_active_downloads(self): APP.initialize_runtime(start_workers=False) try: with mock.patch.object(APP, "shutdown_runtime") as shutdown_runtime: shutdown_runtime.return_value = True APP.reset_runtime(wait=True) shutdown_runtime.assert_called_once_with(wait=True, cancel_active_downloads=None) self.assertIsNone(APP.CONFIG) finally: APP.initialize_runtime(start_workers=False) def test_reset_runtime_refuses_to_clear_globals_when_shutdown_does_not_finish(self): APP.initialize_runtime(start_workers=False) try: current_config = APP.CONFIG current_watchlist = APP.WATCHLIST current_queue = APP.DOWNLOAD_QUEUE current_refresh = APP.WATCHLIST_REFRESH current_auto_refresh = APP.WATCHLIST_AUTO_REFRESH with mock.patch.object(APP, "shutdown_runtime", return_value=False): with self.assertRaises(RuntimeError): APP.reset_runtime(wait=True) self.assertIs(APP.CONFIG, current_config) self.assertIs(APP.WATCHLIST, current_watchlist) self.assertIs(APP.DOWNLOAD_QUEUE, current_queue) self.assertIs(APP.WATCHLIST_REFRESH, current_refresh) self.assertIs(APP.WATCHLIST_AUTO_REFRESH, current_auto_refresh) finally: APP.reset_runtime(wait=True) APP.initialize_runtime(start_workers=False) def test_save_runtime_config_replaces_runtime_config(self): original = APP.CONFIG original_scheduler = APP.WATCHLIST_AUTO_REFRESH try: APP.CONFIG = { "mode": "sub", "quality": "best", "download_dir": "/tmp/old", "watchlist_auto_refresh_enabled": False, "watchlist_auto_refresh_minutes": 60, "watchlist_refresh_delay_seconds": 5, "auto_download_enabled": False, "auto_download_mode": "dub", "auto_download_quality": "best", "discord_webhook_url": "", "discord_webhook_events": [], } APP.WATCHLIST_AUTO_REFRESH = mock.Mock() saved = APP.save_runtime_config( { "mode": "dub", "quality": "720", "download_dir": "/tmp/new", "watchlist_auto_refresh_enabled": True, "watchlist_auto_refresh_minutes": 15, "watchlist_refresh_delay_seconds": 8, "auto_download_enabled": True, "auto_download_mode": "sub", "auto_download_quality": "720", "jellyfin_sync_enabled": True, "jellyfin_tv_dir": "/tmp/jellyfin-tv", "jellyfin_movie_dir": "/tmp/jellyfin-movies", "discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["runtime_error"], } ) self.assertEqual(saved["mode"], "dub") self.assertEqual(APP.CONFIG["mode"], "dub") self.assertEqual(APP.CONFIG["quality"], "720") self.assertTrue(APP.CONFIG["watchlist_auto_refresh_enabled"]) self.assertEqual(APP.CONFIG["watchlist_auto_refresh_minutes"], 15) self.assertEqual(APP.CONFIG["watchlist_refresh_delay_seconds"], 8) self.assertTrue(APP.CONFIG["auto_download_enabled"]) self.assertEqual(APP.CONFIG["auto_download_mode"], "sub") self.assertEqual(APP.CONFIG["auto_download_quality"], "720") self.assertTrue(APP.CONFIG["jellyfin_sync_enabled"]) self.assertEqual(APP.CONFIG["jellyfin_tv_dir"], "/tmp/jellyfin-tv") self.assertEqual(APP.CONFIG["jellyfin_movie_dir"], "/tmp/jellyfin-movies") self.assertEqual(APP.CONFIG["discord_webhook_url"], "https://discord.example/webhook") self.assertEqual(APP.CONFIG["discord_webhook_events"], ["runtime_error"]) APP.WATCHLIST_AUTO_REFRESH.notify_config_changed.assert_called_once_with() finally: APP.CONFIG = original APP.WATCHLIST_AUTO_REFRESH = original_scheduler def test_reset_runtime_clears_runtime_services(self): APP.initialize_runtime(start_workers=False) APP.reset_runtime(wait=True) try: self.assertIsNone(APP.CONFIG) self.assertIsNone(APP.WATCHLIST) self.assertIsNone(APP.DOWNLOAD_QUEUE) self.assertIsNone(APP.WATCHLIST_REFRESH) self.assertIsNone(APP.WATCHLIST_AUTO_REFRESH) finally: APP.initialize_runtime(start_workers=False) class HandlerRouteTests(unittest.TestCase): def test_runtime_backed_route_respects_worker_disable_flag(self): 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): APP.Handler.do_GET(handler) self.assertFalse(APP.WATCHLIST.worker_enabled) self.assertIsNone(APP.WATCHLIST.refresh_worker) self.assertIsNone(APP.DOWNLOAD_QUEUE.worker) self.assertIsNone(APP.WATCHLIST_REFRESH.worker) self.assertIsNone(APP.WATCHLIST_AUTO_REFRESH.worker) finally: APP.reset_runtime(wait=True) 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","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["watchlist_auto_refresh_enabled"]) self.assertEqual(handler.json_payload["watchlist_auto_refresh_minutes"], 25) self.assertEqual(handler.json_payload["watchlist_refresh_delay_seconds"], 9) self.assertTrue(handler.json_payload["auto_download_enabled"]) self.assertEqual(handler.json_payload["auto_download_mode"], "dub") self.assertEqual(handler.json_payload["auto_download_quality"], "720") self.assertTrue(handler.json_payload["jellyfin_sync_enabled"]) self.assertEqual(handler.json_payload["jellyfin_tv_dir"], "/tmp/jellyfin-tv") self.assertEqual(handler.json_payload["jellyfin_movie_dir"], "/tmp/jellyfin-movies") def test_version_route_includes_ani_cli_version(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) def test_filesystem_browse_route_lists_directories(self): with tempfile.TemporaryDirectory() as temp_root: Path(temp_root, "folder-a").mkdir() Path(temp_root, "folder-b").mkdir() Path(temp_root, "file.txt").write_text("example", encoding="utf-8") handler = DummyHandler(f"/api/fs/browse?path={APP.quote(temp_root)}&mode=dir") APP.Handler.do_GET(handler) self.assertEqual(handler.json_status, HTTPStatus.OK) self.assertEqual(handler.json_payload["path"], temp_root) self.assertEqual([entry["name"] for entry in handler.json_payload["entries"]], ["folder-a", "folder-b"]) self.assertTrue(all(entry["is_dir"] for entry in handler.json_payload["entries"])) def test_config_post_accepts_discord_webhook_fields(self): body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","discord_webhook_url":"https://discord.example/webhook","discord_webhook_events":["runtime_error","watchlist_refresh_failed","bad"]}' 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["discord_webhook_url"], "https://discord.example/webhook") self.assertEqual( handler.json_payload["discord_webhook_events"], ["runtime_error", "watchlist_refresh_failed"], ) def test_config_routes_hide_auth_fields_and_preserve_saved_credentials(self): APP.save_runtime_config( { "download_dir": "/tmp/example", "mode": "sub", "quality": "best", "auth_username": "demo", "auth_password": "secret-pass", } ) post_body = b'{"download_dir":"/tmp/updated","mode":"dub","quality":"720"}' post_handler = DummyHandler("/api/config", body=post_body, content_length=len(post_body)) APP.Handler.do_POST(post_handler) self.assertEqual(post_handler.json_status, HTTPStatus.OK) self.assertNotIn("auth_username", post_handler.json_payload) self.assertNotIn("auth_password", post_handler.json_payload) self.assertEqual(APP.get_config_snapshot()["auth_username"], "demo") self.assertEqual(APP.get_config_snapshot()["auth_password"], "secret-pass") get_handler = DummyHandler("/api/config") APP.Handler.do_GET(get_handler) self.assertEqual(get_handler.json_status, HTTPStatus.OK) self.assertNotIn("auth_username", get_handler.json_payload) self.assertNotIn("auth_password", get_handler.json_payload) def test_config_webhook_test_route_uses_unsaved_payload(self): body = b'{"discord_webhook_url":"https://discord.example/webhook","discord_webhook_events":["runtime_error"]}' handler = DummyHandler("/api/config/webhook/test", body=body, content_length=len(body)) test_webhook = mock.Mock(return_value={"message": "ok"}) handler.handler_context = mock.Mock(test_discord_webhook_config=test_webhook) APP.Handler.do_POST(handler) self.assertEqual(handler.json_status, HTTPStatus.OK) self.assertEqual(handler.json_payload["message"], "ok") test_webhook.assert_called_once_with( {"discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["runtime_error"]} ) def test_body_json_rejects_oversized_request(self): handler = DummyHandler("/api/watchlist/upload-thumbnail", body=b"{}", content_length=APP.MAX_JSON_BODY_BYTES + 1) with self.assertRaises(APP.HttpError) as ctx: APP.Handler.body_json(handler) self.assertEqual(ctx.exception.status, HTTPStatus.REQUEST_ENTITY_TOO_LARGE) def test_body_json_rejects_non_utf8_payload(self): handler = DummyHandler("/api/watchlist/upload-thumbnail", body=b"\xff", content_length=1) with self.assertRaises(ValueError) as ctx: APP.Handler.body_json(handler) self.assertIn("utf-8", str(ctx.exception).lower()) def test_config_post_rejects_non_object_json_body(self): body = b"[]" handler = DummyHandler("/api/config", body=body, content_length=len(body)) APP.Handler.do_POST(handler) self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST) self.assertIn("json object", handler.error_message.lower()) def test_remove_post_missing_show_id_returns_bad_request(self): handler = DummyHandler("/api/watchlist/remove", body=b"{}", content_length=2) APP.Handler.do_POST(handler) self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST) self.assertIn("show_id", handler.error_message) def test_upload_thumbnail_post_missing_required_fields_returns_bad_request(self): body = b'{"show_id":"show-1"}' handler = DummyHandler("/api/watchlist/upload-thumbnail", body=body, content_length=len(body)) APP.Handler.do_POST(handler) self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST) self.assertIn("data_url", handler.error_message) def test_watchlist_add_duplicate_returns_ok_instead_of_created(self): body = b'{"show_id":"show-1","title":"Show 1","category":"watching"}' handler = DummyHandler("/api/watchlist/add", body=body, content_length=len(body)) result = {"message": "Already tracking Show 1.", "item": {"show_id": "show-1", "title": "Show 1"}, "created": False} with mock.patch.object(APP.WATCHLIST, "add", return_value=result): APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, result) self.assertEqual(handler.json_status, HTTPStatus.OK) def test_watchlist_add_rejects_non_object_json_body(self): body = b"[]" handler = DummyHandler("/api/watchlist/add", body=body, content_length=len(body)) APP.Handler.do_POST(handler) self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST) self.assertIn("json object", handler.error_message.lower()) def test_ensure_thumbnails_rejects_non_array_show_ids(self): body = b'{"show_ids":"show-1"}' handler = DummyHandler("/api/watchlist/ensure-thumbnails", body=body, content_length=len(body)) APP.Handler.do_POST(handler) self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST) self.assertIn("show_ids", handler.error_message) def test_ensure_thumbnails_parses_string_false_force(self): body = b'{"show_ids":["show-1"],"force":"false"}' handler = DummyHandler("/api/watchlist/ensure-thumbnails", body=body, content_length=len(body)) with mock.patch.object(APP.WATCHLIST, "ensure_thumbnails", return_value={"items": []}) as ensure_thumbnails: APP.Handler.do_POST(handler) ensure_thumbnails.assert_called_once_with(["show-1"], limit=6, force=False) self.assertEqual(handler.json_status, HTTPStatus.OK) def test_ensure_thumbnails_rejects_invalid_force_value(self): body = b'{"show_ids":["show-1"],"force":"maybe"}' handler = DummyHandler("/api/watchlist/ensure-thumbnails", body=body, content_length=len(body)) APP.Handler.do_POST(handler) self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST) self.assertIn("force", handler.error_message) def test_thumbnail_route_serves_cached_file_without_fetching(self): thumb_path = APP.THUMBNAIL_ROOT / "show-1.jpg" thumb_path.parent.mkdir(parents=True, exist_ok=True) thumb_path.write_bytes(b"jpg") handler = DummyHandler("/api/watchlist/thumb/show-1") with mock.patch.object(APP.WATCHLIST, "get", return_value={"thumbnail_path": "show-1.jpg"}) as get_item, mock.patch.object( APP.WATCHLIST, "ensure_thumbnail", side_effect=AssertionError("thumbnail fetch should not run") ): APP.Handler.do_GET(handler) get_item.assert_called_once_with("show-1") self.assertEqual(handler.file_path, thumb_path) self.assertIsNone(handler.error_status) def test_watchlist_refresh_status_route_returns_json(self): handler = DummyHandler("/api/watchlist/refresh-status") payload = {"job": {"id": "refresh-1", "status": "running"}, "running": True} with mock.patch.object(APP.WATCHLIST_REFRESH, "status", return_value=payload): APP.Handler.do_GET(handler) self.assertEqual(handler.json_payload, payload) self.assertEqual(handler.json_status, HTTPStatus.OK) def test_update_all_post_returns_accepted_for_new_job(self): handler = DummyHandler("/api/watchlist/update-all", body=b"{}", content_length=2) payload = {"message": "Watchlist refresh started.", "job": {"id": "refresh-1"}, "created": True} with mock.patch.object(APP.WATCHLIST_REFRESH, "start", return_value=payload): APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, payload) self.assertEqual(handler.json_status, HTTPStatus.ACCEPTED) def test_update_status_post_queues_refresh(self): body = b'{"show_id":"show-1"}' handler = DummyHandler("/api/watchlist/update-status", body=body, content_length=len(body)) queued_item = {"show_id": "show-1", "title": "Show 1", "status": "queued"} with mock.patch.object(APP.WATCHLIST, "queue_refresh", return_value=queued_item): APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, {"message": "Queued refresh for Show 1.", "item": queued_item}) self.assertEqual(handler.json_status, HTTPStatus.OK) def test_watchlist_download_all_post_returns_created_job(self): body = b'{"show_id":"show-1","mode":"sub"}' handler = DummyHandler("/api/watchlist/download-all", body=body, content_length=len(body)) payload = {"message": "Queued Show 1 for sub download.", "job": {"id": "job-1"}, "item": {"show_id": "show-1"}} handler.handler_context = mock.Mock(download_watchlist_item=mock.Mock(return_value=payload)) APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, payload) self.assertEqual(handler.json_status, HTTPStatus.CREATED) def test_watchlist_auto_download_settings_post_returns_updated_item(self): body = b'{"show_id":"show-1","mode":"dub","quality":"best","name":"Example Name","series":"2","episode_offset":"13"}' handler = DummyHandler("/api/watchlist/update-auto-download", body=body, content_length=len(body)) payload = {"message": "Saved auto-download settings for Show 1.", "item": {"show_id": "show-1"}} handler.handler_context = mock.Mock(update_watchlist_auto_download=mock.Mock(return_value=payload)) APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, payload) self.assertEqual(handler.json_status, HTTPStatus.OK) handler.handler_context.update_watchlist_auto_download.assert_called_once_with( "show-1", "dub", "best", "Example Name", "2", "13" ) def test_watchlist_media_type_post_returns_updated_item(self): body = b'{"show_id":"show-1","media_type":"movie"}' handler = DummyHandler("/api/watchlist/update-media-type", body=body, content_length=len(body)) payload = {"message": "Saved Show 1 as Movie.", "item": {"show_id": "show-1", "media_type": "movie"}} handler.handler_context = mock.Mock(update_watchlist_media_type=mock.Mock(return_value=payload)) APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, payload) self.assertEqual(handler.json_status, HTTPStatus.OK) def test_jellyfin_sync_post_returns_summary_payload(self): body = b'{"download_dir":"/tmp/example","jellyfin_sync_enabled":true,"jellyfin_tv_dir":"/tmp/jellyfin-tv","jellyfin_movie_dir":"/tmp/jellyfin-movies"}' handler = DummyHandler("/api/config/jellyfin/sync", body=body, content_length=len(body)) payload = {"message": "Moved 1 watchlist entry to Jellyfin libraries.", "moved": 1, "total": 2, "items": []} handler.handler_context = mock.Mock(run_jellyfin_sync=mock.Mock(return_value=payload)) APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, payload) self.assertEqual(handler.json_status, HTTPStatus.OK) def test_generic_handler_exception_skips_traceback_when_debug_disabled(self): handler = DummyHandler("/api/config") with mock.patch.object(http_handler, "debug_enabled", return_value=False), mock.patch.object( http_handler, "debug_log" ) as debug_log, mock.patch.object(http_handler, "send_discord_webhook_event") as send_webhook, mock.patch.object( http_handler.traceback, "print_exc" ) as print_exc: APP.Handler.exception(handler, RuntimeError("boom")) print_exc.assert_not_called() send_webhook.assert_called_once() debug_log.assert_called_once() self.assertEqual(handler.error_status, HTTPStatus.INTERNAL_SERVER_ERROR) self.assertEqual(handler.error_message, "Internal server error.") class TemplateHelperTests(unittest.TestCase): def test_log_message_skips_queue_poll_requests(self): handler = object.__new__(APP.Handler) handler.path = "/api/queue?page=1&per_page=10" handler.command = "GET" handler.address_string = lambda: "127.0.0.1" with mock.patch("builtins.print") as print_mock: APP.Handler.log_message(handler, '"GET /api/queue?page=1&per_page=10 HTTP/1.1" %s -', 200) print_mock.assert_not_called() def test_log_message_keeps_non_poll_requests(self): handler = object.__new__(APP.Handler) handler.path = "/api/config" handler.command = "GET" handler.address_string = lambda: "127.0.0.1" with mock.patch("builtins.print") as print_mock: APP.Handler.log_message(handler, '"GET /api/config HTTP/1.1" %s -', 200) print_mock.assert_called_once() def test_page_links_marks_active_page(self): markup = APP.render_page_links("config") self.assertIn('class="page-link active" href="/config"', markup) self.assertIn('class="page-link" href="/queue"', markup) self.assertIn('class="page-link" href="/watchlist"', markup) def test_sidebar_brand_can_attach_optional_badge_id(self): markup = APP.render_sidebar_brand("Shared note", "Search", badge_id="serverBadge") self.assertIn('id="serverBadge"', markup) self.assertIn("Shared note", markup) def test_queue_page_queue_loader_guards_against_stale_responses(self): self.assertIn("queueRequestId", APP.QUEUE_HTML) self.assertIn("if (requestId !== state.queueRequestId) return data;", APP.QUEUE_HTML) def test_queue_page_renders_queue_logs_newest_first(self): self.assertIn('[...(job.log || [])].slice(-28).reverse().join("\\n")', APP.QUEUE_HTML) def test_pages_share_browser_api_helper(self): self.assertIn("async function api(path, options = {})", APP.INDEX_HTML) self.assertIn("async function api(path, options = {})", APP.QUEUE_HTML) self.assertIn("async function api(path, options = {})", APP.CONFIG_HTML) self.assertIn("async function api(path, options = {})", APP.WATCHLIST_HTML) def test_config_page_places_dependencies_before_settings_grid(self): self.assertIn('
', APP.CONFIG_HTML) self.assertIn('
', 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="aniCliVersionLine"', APP.CONFIG_HTML) self.assertIn('ani-cli ${data.ani_cli_version || "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) 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_queues_selected_title_without_result_index(self): self.assertIn('query: state.selected.title,', APP.INDEX_HTML) self.assertNotIn('index: state.selected.index,', APP.INDEX_HTML) 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_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_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("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) if __name__ == "__main__": unittest.main()