Dymas/ani-cli-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add scheduled watchlist refresh support

Browse Source
This commit is contained in:
Dymas
2026-05-17 20:38:16 +02:00
parent 47480efa30
commit 6168acb272
8 changed files with 285 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_app.py
+95 -3
View File
@@ -1048,7 +1048,13 @@ class WatchlistRefreshQueueTests(unittest.TestCase):
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"}
APP.CONFIG = {
"mode": "sub",
"quality": "best",
"download_dir": "/tmp/example",
"watchlist_auto_refresh_enabled": False,
"watchlist_auto_refresh_minutes": 60,
}
try:
snapshot = APP.get_config_snapshot()
snapshot["mode"] = "dub"
@@ -1056,6 +1062,60 @@ class ConfigSnapshotTests(unittest.TestCase):
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)
def test_normalize_config_parses_watchlist_schedule_fields(self):
config = APP.normalize_config(
{
"watchlist_auto_refresh_enabled": "true",
"watchlist_auto_refresh_minutes": "15",
}
)
self.assertTrue(config["watchlist_auto_refresh_enabled"])
self.assertEqual(config["watchlist_auto_refresh_minutes"], 15)
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()
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):
@@ -1092,6 +1152,7 @@ class StartupBehaviorTests(unittest.TestCase):
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)
@@ -1100,20 +1161,41 @@ class StartupBehaviorTests(unittest.TestCase):
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"}
saved = APP.save_runtime_config({"mode": "dub", "quality": "720", "download_dir": "/tmp/new"})
APP.CONFIG = {
"mode": "sub",
"quality": "best",
"download_dir": "/tmp/old",
"watchlist_auto_refresh_enabled": False,
"watchlist_auto_refresh_minutes": 60,
}
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,
}
)
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)
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)
@@ -1123,6 +1205,7 @@ class StartupBehaviorTests(unittest.TestCase):
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)
@@ -1138,10 +1221,19 @@ class HandlerRouteTests(unittest.TestCase):
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"}'
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)
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: