1798 lines
78 KiB
Python
1798 lines
78 KiB
Python
#!/usr/bin/env python3
|
|
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_token_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("REMOTE_TOKEN", str(ctx.exception))
|
|
|
|
def test_remote_access_accepts_valid_token_for_non_local_clients(self):
|
|
handler = DummyHandler("/api/config")
|
|
handler.client_address = ("192.168.1.25", 8421)
|
|
handler.headers["Authorization"] = "Bearer secret-token"
|
|
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
|
clear=False,
|
|
):
|
|
APP.Handler.ensure_client_access(handler)
|
|
|
|
def test_remote_access_accepts_valid_cookie_token_for_non_local_clients(self):
|
|
handler = DummyHandler("/api/config")
|
|
handler.client_address = ("192.168.1.25", 8421)
|
|
handler.headers["Cookie"] = "ani_cli_web_token=secret-token"
|
|
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
|
clear=False,
|
|
):
|
|
APP.Handler.ensure_client_access(handler)
|
|
|
|
def test_remote_access_accepts_valid_query_token_for_non_local_clients(self):
|
|
handler = DummyHandler("/api/config?token=secret-token")
|
|
handler.client_address = ("192.168.1.25", 8421)
|
|
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
|
clear=False,
|
|
):
|
|
APP.Handler.ensure_client_access(handler)
|
|
|
|
def test_remote_get_bootstraps_cookie_and_redirects_when_query_token_valid(self):
|
|
handler = DummyHandler("/?token=secret-token")
|
|
handler.client_address = ("192.168.1.25", 8421)
|
|
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
|
clear=False,
|
|
):
|
|
APP.Handler.do_GET(handler)
|
|
|
|
headers = dict(handler.response_headers)
|
|
self.assertEqual(handler.response_status, HTTPStatus.SEE_OTHER)
|
|
self.assertEqual(headers.get("Location"), "/")
|
|
self.assertIn("ani_cli_web_token=secret-token", headers.get("Set-Cookie", ""))
|
|
self.assertIn("Max-Age=", headers.get("Set-Cookie", ""))
|
|
|
|
def test_remote_browser_get_without_token_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_REMOTE_TOKEN": "secret-token"},
|
|
clear=False,
|
|
):
|
|
APP.Handler.do_GET(handler)
|
|
|
|
body = handler.wfile.getvalue().decode("utf-8")
|
|
self.assertEqual(handler.response_status, HTTPStatus.UNAUTHORIZED)
|
|
self.assertIn('<form method="post" action="/auth/login">', body)
|
|
self.assertIn("Remote access token", body)
|
|
|
|
def test_remote_login_post_sets_cookie_and_redirects(self):
|
|
body = b"token=secret-token&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_REMOTE_TOKEN": "secret-token"},
|
|
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_token=secret-token", headers.get("Set-Cookie", ""))
|
|
|
|
def test_loopback_proxy_with_forwarded_remote_ip_requires_token(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("REMOTE_TOKEN", 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")
|
|
|
|
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",
|
|
"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["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"])
|
|
|
|
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_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")
|
|
|
|
|
|
class WatchlistRefreshAllTests(unittest.TestCase):
|
|
def test_refresh_all_iterates_every_show_id(self):
|
|
store = object.__new__(APP.WatchlistStore)
|
|
calls = []
|
|
store.all_show_ids = lambda: ["show-a", "show-b", "show-c"]
|
|
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), ("show-c", False)])
|
|
self.assertEqual(result["count"], 3)
|
|
self.assertEqual([item["show_id"] for item in result["items"]], ["show-a", "show-b", "show-c"])
|
|
|
|
|
|
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):
|
|
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):
|
|
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: ["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):
|
|
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):
|
|
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):
|
|
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 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,
|
|
"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_refresh_persists_allmanga_alternative_titles(self):
|
|
self.seed_watchlist_item()
|
|
snapshot = {
|
|
"show_id": "show-1",
|
|
"title": "Example Show",
|
|
"episodes": {"sub": ["1"], "dub": []},
|
|
}
|
|
with mock.patch.object(APP, "fetch_show_episode_snapshot", return_value=snapshot), mock.patch.object(
|
|
APP, "resolve_animeschedule_anime", return_value=None
|
|
), mock.patch.object(
|
|
APP, "fetch_allmanga_alternative_titles", return_value={"english_title": "Example Show English", "alt_titles": ["Example Show English", "Sample Alias"]}
|
|
), mock.patch.object(APP.WatchlistStore, "ensure_thumbnail", return_value=None):
|
|
item = APP.WATCHLIST.refresh("show-1")
|
|
|
|
self.assertEqual(item["english_title"], "Example Show English")
|
|
self.assertEqual(item["display_alt_titles"], ["Example Show English", "Sample Alias"])
|
|
|
|
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: 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")
|
|
self.assertEqual(item["status"], "queued")
|
|
self.assertEqual(item["status_message"], "Queued from test.")
|
|
|
|
def test_mark_downloaded_moves_existing_category_to_finished_and_sets_download_flag(self):
|
|
self.seed_watchlist_item(show_id="show-9", title="Queue Show", category="planned", downloaded=False)
|
|
|
|
with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id: 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-9", title="Queue 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_creates_missing_show_in_finished_category(self):
|
|
with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id: 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_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)
|
|
|
|
with mock.patch.object(APP.WATCHLIST, "schedule_refresh", side_effect=lambda show_id: 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("resolved-show-9", title="Queue Show")
|
|
|
|
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: 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")
|
|
self.assertTrue(result["created"])
|
|
self.assertEqual(result["item"]["status"], "queued")
|
|
self.assertIn("Refresh queued", result["message"])
|
|
|
|
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")
|
|
|
|
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_download_watchlist_item_queues_full_episode_range_for_selected_mode(self):
|
|
item = {"show_id": "show-42", "title": "Queue Show"}
|
|
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, "search_anime", return_value=[{"id": "show-42", "title": "Queue Show", "query": "Queue Show", "index": 3}]), 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",
|
|
"season": "1",
|
|
"index": 3,
|
|
"mode": "dub",
|
|
"quality": "best",
|
|
"episodes": "1-12",
|
|
"download_dir": "/tmp/downloads",
|
|
}
|
|
)
|
|
self.assertEqual(result["job"], job)
|
|
self.assertIn("Queued Queue Show for dub download.", result["message"])
|
|
|
|
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", "show-b"])
|
|
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"]
|
|
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):
|
|
calls.append(show_id)
|
|
store.refresh_stop_event.set()
|
|
|
|
store.refresh = fake_refresh
|
|
|
|
APP.WatchlistStore._refresh_loop(store)
|
|
|
|
self.assertEqual(calls, ["show-2"])
|
|
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", "show-4"]
|
|
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"])
|
|
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,
|
|
"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.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",
|
|
}
|
|
)
|
|
self.assertTrue(config["watchlist_auto_refresh_enabled"])
|
|
self.assertEqual(config["watchlist_auto_refresh_minutes"], 15)
|
|
self.assertEqual(config["watchlist_refresh_delay_seconds"], 7)
|
|
|
|
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):
|
|
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):
|
|
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")
|
|
|
|
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 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,
|
|
"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,
|
|
"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.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"}'
|
|
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)
|
|
|
|
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_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_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.traceback, "print_exc") as print_exc:
|
|
APP.Handler.exception(handler, RuntimeError("boom"))
|
|
|
|
print_exc.assert_not_called()
|
|
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_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(() => loadQueue(), 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("Enter sub or dub", APP.WATCHLIST_HTML)
|
|
|
|
def test_watchlist_page_shows_alternative_titles(self):
|
|
self.assertIn('class="ghost alt-toggle"', APP.WATCHLIST_HTML)
|
|
self.assertIn('class="alt-titles"', APP.WATCHLIST_HTML)
|
|
self.assertIn('Alternative titles', APP.WATCHLIST_HTML)
|
|
self.assertIn('altTitlesNode.classList.toggle("visible")', 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)
|
|
|
|
def test_parse_allmanga_alternative_titles_extracts_english_and_synonyms(self):
|
|
html_text = """
|
|
<div class="show-more-content card-text p-0" style="">
|
|
<span class="mr-1 altnames"> ⚪ Example Show English </span>
|
|
<span class="mr-1 altnames"> ⚪ Sample Alias </span>
|
|
<span class="mr-1 altnames"> ⚪ Example Show </span>
|
|
</div>
|
|
"""
|
|
parsed = APP.parse_allmanga_alternative_titles(html_text, primary_title="Example Show")
|
|
self.assertEqual(parsed["english_title"], "Example Show English")
|
|
self.assertEqual(parsed["alt_titles"], ["Example Show English", "Sample Alias"])
|
|
|
|
|
|
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()
|