refactor(backend): route config reload through AppContext, fix blocking I/O, dedupe N+1 query

Setup wizard and config reload now funnel through a single
AppContext.reload_settings() entry point instead of mutating settings
directly. Wrap remaining sync file I/O (image cache, settings load/save,
log tail read) in asyncio.to_thread so async request handlers stop
blocking the event loop. Add reset_cache() to each cache-owning parser/
searcher module (tmdb, mikan, title parser, poster cache) and wire them
into reload_settings() so a base-URL/API-key config change invalidates
stale cached results immediately. Fix BangumiDatabase.add_all's N+1 query
by batching semantic-duplicate candidate lookups into one SELECT for the
whole batch instead of one query per candidate.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GTnuEgR78cBk32bLNKar3z
This commit is contained in:
Estrella Pan
2026-07-02 17:39:29 +02:00
parent 42418d5b18
commit d9f41f4b11
20 changed files with 642 additions and 104 deletions

View File

@@ -1,3 +1,4 @@
import asyncio
import logging
from fastapi import APIRouter, Depends
@@ -70,7 +71,8 @@ async def update_config(config: Config, ctx: AppContext = Depends(get_context)):
"""Persist and reload configuration from the supplied payload."""
try:
config_dict = _restore_masked(config.dict(), settings.dict())
settings.save(config_dict=config_dict)
# settings.save() does synchronous file I/O; keep it off the event loop.
await asyncio.to_thread(settings.save, config_dict=config_dict)
# reload_settings reloads from disk, resets the shared HTTP client,
# rebuilds notifications, and re-applies the RSS/rename loops.
await ctx.reload_settings()

View File

@@ -1,3 +1,5 @@
import asyncio
from fastapi import APIRouter, Depends, HTTPException, Response, status
from fastapi.responses import JSONResponse
@@ -11,23 +13,30 @@ router = APIRouter(prefix="/log", tags=["log"])
_TAIL_BYTES = 512 * 1024 # 512 KB
def _read_log_tail() -> bytes:
"""同步读取日志文件的最后 512 KB,供 asyncio.to_thread 在线程池中执行。"""
with open(LOG_PATH, "rb") as f:
f.seek(0, 2)
size = f.tell()
if size > _TAIL_BYTES:
f.seek(-_TAIL_BYTES, 2)
data = f.read()
# Drop first partial line
idx = data.find(b"\n")
if idx != -1:
data = data[idx + 1 :]
else:
f.seek(0)
data = f.read()
return data
@router.get("", response_model=str, dependencies=[Depends(get_current_user)])
async def get_log():
if LOG_PATH.exists():
with open(LOG_PATH, "rb") as f:
f.seek(0, 2)
size = f.tell()
if size > _TAIL_BYTES:
f.seek(-_TAIL_BYTES, 2)
data = f.read()
# Drop first partial line
idx = data.find(b"\n")
if idx != -1:
data = data[idx + 1 :]
else:
f.seek(0)
data = f.read()
return Response(data, media_type="text/plain")
# Up to 512 KB of sync file I/O; keep it off the event loop.
data = await asyncio.to_thread(_read_log_tail)
return Response(data, media_type="text/plain")
else:
return Response("Log file not found", status_code=404)

View File

@@ -1,3 +1,4 @@
import asyncio
import ipaddress
import logging
import socket
@@ -5,16 +6,19 @@ from pathlib import Path
from urllib.parse import urlparse
import httpx
from fastapi import APIRouter, HTTPException, Request
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field
from module.conf import VERSION, settings
from module.core import AppContext
from module.models import Config, ResponseModel
from module.models.config import NotificationProvider as ProviderConfig
from module.network import RequestContent
from module.notification import PROVIDER_REGISTRY
from module.security.jwt import get_password_hash, verify_password
from .deps import get_context
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/setup", tags=["setup"])
@@ -321,7 +325,11 @@ async def test_notification(req: TestNotificationRequest):
@router.post("/complete", response_model=ResponseModel)
async def complete_setup(req: SetupCompleteRequest, request: Request):
async def complete_setup(
req: SetupCompleteRequest,
request: Request,
ctx: AppContext = Depends(get_context),
):
"""Save all wizard configuration and mark setup as complete."""
_require_setup_needed()
await _require_default_admin_or_authenticated(request)
@@ -361,10 +369,12 @@ async def complete_setup(req: SetupCompleteRequest, request: Request):
],
}
settings.save(config_dict)
# Reload settings in-place
config_obj = Config.parse_obj(config_dict)
settings.__dict__.update(config_obj.__dict__)
# settings.save() does synchronous file I/O; keep it off the event loop.
await asyncio.to_thread(settings.save, config_dict)
# Route through the AppContext reload path so the shared httpx client,
# notifier, and scheduler are rebuilt against the new config too — not
# just the in-memory settings dict.
await ctx.reload_settings()
# 3. Add RSS feed if provided
if req.rss_url:

View File

@@ -16,7 +16,11 @@ from module.downloader.download_client import shutdown as downloader_shutdown
from module.models import ResponseModel
from module.network.request_url import reset_shared_client
from module.notification import NotificationManager
from module.parser.analyser.mikan_parser import reset_cache as reset_mikan_cache
from module.parser.analyser.tmdb_parser import reset_cache as reset_tmdb_cache
from module.parser.title_parser import reset_cache as reset_openai_parser
from module.rss import RSSAnalyser
from module.searcher.searcher import reset_cache as reset_poster_cache
from module.update import (
cache_image,
data_migration,
@@ -247,8 +251,12 @@ class AppContext:
it inline used to block both the lifespan and the /start,/restart API
handlers for that long. It now runs as a supervised background task
tracked on ``self._start_task`` so ``stop()`` can cancel it cleanly.
Reloads settings through ``reload_settings()`` (the single reload entry
point) rather than calling ``settings.load()`` directly; the scheduler
is not yet running at this point so its restart branch is a no-op.
"""
settings.load()
await self.reload_settings()
if self._start_task is None or self._start_task.done():
self._start_task = asyncio.create_task(self._run_start_tasks())
self._start_task.add_done_callback(self._supervise_start_task)
@@ -331,9 +339,23 @@ class AppContext:
# ------------------------------------------------------------------ reload
async def reload_settings(self) -> None:
"""Single choreography for a config update: reload + re-apply live state."""
settings.load()
"""Single choreography for a config update: reload + re-apply live state.
This is the ONLY place settings are reloaded from disk — callers must
route through here (not a bare ``settings.load()``) so the HTTP
client, notifier, endpoint-keyed caches, and scheduler all stay in
sync with the new config.
"""
# settings.load() does synchronous file I/O; keep it off the event loop.
await asyncio.to_thread(settings.load)
await reset_shared_client()
# Endpoint-keyed caches (TMDB/Mikan/poster lookups, the OpenAI parser
# singleton) must be dropped too, or a changed tmdb_base_url/bgm_base_url
# keeps serving results cached from the old endpoint.
reset_tmdb_cache()
reset_mikan_cache()
reset_poster_cache()
reset_openai_parser()
self.notifier.rebuild()
if self.scheduler.running:
await self.scheduler.stop_all()

View File

@@ -100,6 +100,32 @@ def build_save_path_index(bangumi_list: list[Bangumi]) -> dict[str, Bangumi]:
return index
def _find_semantic_match(data: Bangumi, candidates: list[Bangumi]) -> Optional[Bangumi]:
"""在已加载的候选列表中查找语义重复项(同一部番剧、命名规则不同)。
被 ``find_semantic_duplicate``(单条、逐条查库)和 ``add_all``(批量、
候选一次性查库后在内存中匹配)共用,匹配规则只维护一处。
"""
for candidate in candidates:
is_exact_duplicate = (
candidate.title_raw == data.title_raw
and candidate.group_name == data.group_name
)
if is_exact_duplicate:
continue
is_semantic_match = (
candidate.dpi == data.dpi
and candidate.subtitle == data.subtitle
and candidate.source == data.source
and _groups_are_similar(candidate.group_name, data.group_name)
)
if is_semantic_match:
return candidate
return None
def match_bangumi_in_list(
torrent_name: str, bangumi_list: list[Bangumi]
) -> Optional[Bangumi]:
@@ -149,31 +175,16 @@ class BangumiDatabase:
result = await self.session.execute(statement)
candidates = result.scalars().all()
for candidate in candidates:
is_exact_duplicate = (
candidate.title_raw == data.title_raw
and candidate.group_name == data.group_name
match = _find_semantic_match(data, candidates)
if match:
logger.debug(
"[Database] Found semantic duplicate: '%s' matches "
"existing '%s' (official: %s)",
data.title_raw,
match.title_raw,
data.official_title,
)
if is_exact_duplicate:
continue
is_semantic_match = (
candidate.dpi == data.dpi
and candidate.subtitle == data.subtitle
and candidate.source == data.source
and _groups_are_similar(candidate.group_name, data.group_name)
)
if is_semantic_match:
logger.debug(
"[Database] Found semantic duplicate: '%s' matches "
"existing '%s' (official: %s)",
data.title_raw,
candidate.title_raw,
data.official_title,
)
return candidate
return None
return match
async def add_title_alias(
self, bangumi_id: int, new_title_raw: str, auto_commit: bool = True
@@ -279,11 +290,33 @@ class BangumiDatabase:
# Filter out exact duplicates
to_add = [d for d in datas if (d.title_raw, d.group_name) not in existing]
# Batch query: load all semantic-duplicate candidates (grouped by
# official_title) in one SELECT, instead of one find_semantic_duplicate()
# query per candidate. Safe because no rows are inserted mid-loop —
# add_title_alias() only mutates existing rows and the new bangumi are
# only added to the session after this loop finishes.
official_titles = {d.official_title for d in to_add}
candidates_by_title: dict[str, list[Bangumi]] = {}
if official_titles:
statement = select(Bangumi).where(
and_(
Bangumi.official_title.in_(official_titles),
Bangumi.deleted == false(),
)
)
result = await self.session.execute(statement)
for candidate in result.scalars().all():
candidates_by_title.setdefault(candidate.official_title, []).append(
candidate
)
# Check for semantic duplicates and add as aliases
semantic_merged = 0
really_to_add = []
for d in to_add:
semantic_match = await self.find_semantic_duplicate(d)
semantic_match = _find_semantic_match(
d, candidates_by_title.get(d.official_title, [])
)
if semantic_match:
# Add as alias instead of creating new entry (defer commit)
await self.add_title_alias(

View File

@@ -33,7 +33,7 @@ class TelegramProvider(NotificationProvider):
"disable_notification": True,
}
photo = load_image(notification.poster_path)
photo = await load_image(notification.poster_path)
if photo:
resp = await self.post_files(self.photo_url, data, files={"photo": photo})
else:

View File

@@ -16,6 +16,12 @@ _MIKAN_CACHE_MAX = 512
_mikan_cache: "OrderedDict[str, tuple[str, str]]" = OrderedDict()
def reset_cache() -> None:
"""清空 Mikan 主页解析缓存。配置重载后必须调用,否则会继续返回旧配置下
缓存的结果。"""
_mikan_cache.clear()
def _cache_result(homepage: str, result: tuple[str, str]) -> tuple[str, str]:
if len(_mikan_cache) >= _MIKAN_CACHE_MAX:
_mikan_cache.popitem(last=False)
@@ -44,12 +50,15 @@ async def mikan_parser(homepage: str):
img = await req.get_content(f"https://{root_path}{poster_path}")
suffix = poster_path.split(".")[-1]
# img can be None if the poster download failed; don't crash on it.
poster_link = save_image(img, suffix) if img else ""
poster_link = await save_image(img, suffix) if img else ""
return _cache_result(homepage, (poster_link, official_title))
return _cache_result(homepage, ("", official_title))
if __name__ == '__main__':
if __name__ == "__main__":
import asyncio
homepage = "https://mikanani.me/Home/Episode/c89b3c6f0c1c0567a618f5288b853823c87a9862"
homepage = (
"https://mikanani.me/Home/Episode/c89b3c6f0c1c0567a618f5288b853823c87a9862"
)
print(asyncio.run(mikan_parser(homepage)))

View File

@@ -23,6 +23,12 @@ _TMDB_CACHE_MAX = 512
_tmdb_cache: OrderedDict[str, "TMDBInfo | None"] = OrderedDict()
def reset_cache() -> None:
"""清空 TMDB 查询缓存。配置重载(如 tmdb_base_url 变更)后必须调用,否则会
继续返回旧接口地址下缓存的结果。"""
_tmdb_cache.clear()
@dataclass
class TMDBInfo:
id: int
@@ -299,7 +305,7 @@ async def tmdb_parser(title, language, test: bool = False) -> TMDBInfo | None:
f"https://image.tmdb.org/t/p/w780{poster_path}"
)
# img is None if the poster download failed; don't crash on it.
poster_link = save_image(img, "jpg") if img else None
poster_link = await save_image(img, "jpg") if img else None
else:
poster_link = "https://image.tmdb.org/t/p/w780" + poster_path
else:

View File

@@ -27,6 +27,14 @@ def _get_openai_parser(kwargs: dict) -> OpenAIParser:
return _openai_parser
def reset_cache() -> None:
"""清空 OpenAI 解析器单例。配置重载后必须调用,否则会继续使用旧配置
(如 api_base/api_key)构建的客户端。"""
global _openai_parser, _openai_parser_kwargs
_openai_parser = None
_openai_parser_kwargs = None
class TitleParser:
def __init__(self):
pass

View File

@@ -1,5 +1,6 @@
import json
import logging
from collections import OrderedDict
from typing import TypeAlias
from module.models import Bangumi, RSSItem, Torrent
@@ -22,8 +23,17 @@ SEARCH_KEY = [
BangumiJSON: TypeAlias = str
# Cache for TMDB poster lookups by official_title
_poster_cache: dict[str, str | None] = {}
# Cache for TMDB poster lookups by official_title. Bounded (LRU-ish,
# oldest-evicted) like _tmdb_cache/_mikan_cache — was previously a plain dict
# and grew unbounded for the life of the process.
_POSTER_CACHE_MAX = 512
_poster_cache: "OrderedDict[str, str | None]" = OrderedDict()
def reset_cache() -> None:
"""清空 TMDB 海报查询缓存。配置重载(如 tmdb_base_url 变更)后必须调用,
否则会继续返回旧接口地址下缓存的结果。"""
_poster_cache.clear()
class SearchTorrent:
@@ -37,18 +47,21 @@ class SearchTorrent:
async def _fetch_tmdb_poster(self, title: str) -> str | None:
"""Fetch poster from TMDB if not in cache."""
if title in _poster_cache:
_poster_cache.move_to_end(title)
return _poster_cache[title]
poster_link = None
try:
tmdb_info = await tmdb_parser(title, "zh", test=True)
if tmdb_info and tmdb_info.poster_link:
_poster_cache[title] = tmdb_info.poster_link
return tmdb_info.poster_link
poster_link = tmdb_info.poster_link
except Exception as e:
logger.debug("[Searcher] Failed to fetch TMDB poster for %s: %s", title, e)
_poster_cache[title] = None
return None
if len(_poster_cache) >= _POSTER_CACHE_MAX:
_poster_cache.popitem(last=False)
_poster_cache[title] = poster_link
return poster_link
async def analyse_keyword(
self, keywords: list[str], site: str = "mikan", limit: int = 100

View File

@@ -63,7 +63,7 @@ async def cache_image():
# Hash local path
img = await req.get_content(bangumi.poster_link)
suffix = bangumi.poster_link.split(".")[-1]
img_path = save_image(img, suffix)
img_path = await save_image(img, suffix)
if img_path:
bangumi.poster_link = img_path
except Exception as e:

View File

@@ -1,21 +1,32 @@
import asyncio
import hashlib
def save_image(img, suffix):
async def save_image(img: bytes | None, suffix: str) -> str | None:
"""保存图片到本地缓存。文件写入是同步阻塞的,放到线程池中执行,避免
在 RSS/通知等异步热路径上阻塞事件循环。"""
if img is None:
# Fetching the poster failed upstream; skip caching instead of
# crashing on hashlib.md5(None).
return None
img_hash = hashlib.md5(img).hexdigest()[0:8]
image_path = f"data/posters/{img_hash}.{suffix}"
with open(image_path, "wb") as f:
f.write(img)
def _write() -> None:
with open(image_path, "wb") as f:
f.write(img)
await asyncio.to_thread(_write)
return f"posters/{img_hash}.{suffix}"
def load_image(img_path):
if img_path:
async def load_image(img_path: str | None) -> bytes | None:
"""读取缓存图片。文件读取是同步阻塞的,放到线程池中执行。"""
if not img_path:
return None
def _read() -> bytes:
with open(f"data/{img_path}", "rb") as f:
return f.read()
else:
return None
return await asyncio.to_thread(_read)

View File

@@ -0,0 +1,88 @@
"""Tests for module.utils.cache_image: async, off-event-loop file I/O.
save_image/load_image do real file I/O and are awaited from the RSS hot path
and telegram sends; they must not block the event loop. Tests chdir into a
tmp_path so they never touch the real backend/data/ directory.
"""
import asyncio
from module.utils.cache_image import load_image, save_image
class TestSaveImage:
async def test_save_image_returns_none_when_img_is_none(self):
"""Fetching the poster failed upstream: skip caching, no filesystem I/O."""
result = await save_image(None, "jpg")
assert result is None
async def test_save_image_writes_file_and_returns_relative_path(
self, tmp_path, monkeypatch
):
monkeypatch.chdir(tmp_path)
(tmp_path / "data" / "posters").mkdir(parents=True)
result = await save_image(b"fake-image-bytes", "jpg")
assert result is not None
assert result.startswith("posters/")
assert result.endswith(".jpg")
assert (tmp_path / "data" / result).read_bytes() == b"fake-image-bytes"
async def test_save_image_offloads_write_to_a_thread(self, tmp_path, monkeypatch):
"""The write must go through asyncio.to_thread, not run inline on the
event loop (regression test for the blocking-I/O fix)."""
monkeypatch.chdir(tmp_path)
(tmp_path / "data" / "posters").mkdir(parents=True)
real_to_thread = asyncio.to_thread
calls = []
async def spy_to_thread(func, /, *args, **kwargs):
calls.append(func)
return await real_to_thread(func, *args, **kwargs)
monkeypatch.setattr("module.utils.cache_image.asyncio.to_thread", spy_to_thread)
await save_image(b"payload", "png")
assert len(calls) == 1
class TestLoadImage:
async def test_load_image_returns_none_for_falsy_path(self):
assert await load_image(None) is None
assert await load_image("") is None
async def test_load_image_reads_file_relative_to_data_dir(
self, tmp_path, monkeypatch
):
monkeypatch.chdir(tmp_path)
posters_dir = tmp_path / "data" / "posters"
posters_dir.mkdir(parents=True)
(posters_dir / "abc123.jpg").write_bytes(b"cached-poster-bytes")
result = await load_image("posters/abc123.jpg")
assert result == b"cached-poster-bytes"
async def test_load_image_offloads_read_to_a_thread(self, tmp_path, monkeypatch):
"""The read must go through asyncio.to_thread, not run inline on the
event loop (regression test for the blocking-I/O fix)."""
monkeypatch.chdir(tmp_path)
posters_dir = tmp_path / "data" / "posters"
posters_dir.mkdir(parents=True)
(posters_dir / "abc123.jpg").write_bytes(b"cached-poster-bytes")
real_to_thread = asyncio.to_thread
calls = []
async def spy_to_thread(func, /, *args, **kwargs):
calls.append(func)
return await real_to_thread(func, *args, **kwargs)
monkeypatch.setattr("module.utils.cache_image.asyncio.to_thread", spy_to_thread)
await load_image("posters/abc123.jpg")
assert len(calls) == 1

View File

@@ -17,6 +17,10 @@ def ctx():
ctx.scheduler = MagicMock()
ctx.scheduler.stop_all = AsyncMock()
ctx.scheduler.start_all = MagicMock()
# A freshly built context has not started its scheduler yet; set this
# explicitly since a bare MagicMock attribute is truthy by default and
# would make reload_settings() think a restart is needed.
ctx.scheduler.running = False
return ctx
@@ -113,6 +117,7 @@ class TestStartTasks:
async def test_start_tasks_waits_then_starts_scheduler(self, ctx):
with (
patch("module.core.context.settings"),
patch("module.core.context.reset_shared_client", new=AsyncMock()),
patch(
"module.core.context.Checker.check_downloader",
new=AsyncMock(return_value=True),
@@ -129,6 +134,19 @@ class TestStartTasks:
assert isinstance(resp, ResponseModel)
assert resp.status is True
async def test_start_tasks_routes_through_reload_settings(self, ctx):
"""start_tasks() must go through reload_settings() (the single reload
entry point) instead of calling settings.load() directly."""
ctx.reload_settings = AsyncMock()
with patch(
"module.core.context.Checker.check_downloader",
new=AsyncMock(return_value=True),
):
await ctx.start_tasks()
await ctx._start_task
ctx.reload_settings.assert_awaited_once()
# ---------------------------------------------------------------------------
# stop()
@@ -200,3 +218,23 @@ class TestReloadSettings:
ctx.notifier.rebuild.assert_called_once()
ctx.scheduler.stop_all.assert_not_awaited()
ctx.scheduler.start_all.assert_not_called()
async def test_reload_resets_endpoint_keyed_caches(self, ctx):
"""A changed tmdb_base_url/bgm_base_url must not keep serving results
cached from the old endpoint — reload_settings() must drop the TMDB,
Mikan, and poster caches plus the OpenAI parser singleton."""
ctx.scheduler.running = False
with (
patch("module.core.context.settings"),
patch("module.core.context.reset_shared_client", new=AsyncMock()),
patch("module.core.context.reset_tmdb_cache") as mock_tmdb,
patch("module.core.context.reset_mikan_cache") as mock_mikan,
patch("module.core.context.reset_poster_cache") as mock_poster,
patch("module.core.context.reset_openai_parser") as mock_openai,
):
await ctx.reload_settings()
mock_tmdb.assert_called_once()
mock_mikan.assert_called_once()
mock_poster.assert_called_once()
mock_openai.assert_called_once()

View File

@@ -9,16 +9,18 @@ from module.models import Bangumi, RSSItem, Torrent
async def _ensure_bangumi(session, bangumi_id: int):
"""确保 bangumi 表中存在指定 id 的记录,满足外键约束。"""
if await session.get(Bangumi, bangumi_id) is None:
session.add(Bangumi(
id=bangumi_id,
official_title=f"Stub Anime {bangumi_id}",
title_raw=f"Stub {bangumi_id}",
group_name="TestGroup",
dpi="1080p",
source="Web",
subtitle="CHT",
rss_link=f"stub_{bangumi_id}",
))
session.add(
Bangumi(
id=bangumi_id,
official_title=f"Stub Anime {bangumi_id}",
title_raw=f"Stub {bangumi_id}",
group_name="TestGroup",
dpi="1080p",
source="Web",
subtitle="CHT",
rss_link=f"stub_{bangumi_id}",
)
)
await session.commit()
@@ -445,6 +447,102 @@ async def test_add_with_semantic_duplicate_creates_alias(db_session):
assert "Frieren Beyond Journey's End" in aliases
async def test_add_all_semantic_duplicates_across_batch_all_merged_as_aliases(
db_session,
):
"""add_all() must catch semantic duplicates for every item in a batch, not
just the first (regression test for the N+1 batching fix)."""
db = BangumiDatabase(db_session)
for i in range(5):
await db.add(
Bangumi(
official_title=f"Anime {i}",
title_raw=f"Raw {i}",
group_name="GroupA",
dpi="1080p",
source="Web",
subtitle="CHT",
rss_link=f"rss{i}",
)
)
batch = [
Bangumi(
official_title=f"Anime {i}",
title_raw=f"Raw {i} New",
group_name="GroupA&GroupB", # naming changed mid-season
dpi="1080p",
source="Web",
subtitle="CHT",
rss_link=f"rss{i}-new",
)
for i in range(5)
]
added = await db.add_all(batch)
assert added == 0 # every item merged as an alias, no new rows
assert len(await db.search_all()) == 5
for i in range(5):
original = await db.search_official_title(f"Anime {i}")
aliases = json.loads(original.title_aliases) if original.title_aliases else []
assert f"Raw {i} New" in aliases
async def test_add_all_semantic_duplicate_lookup_uses_batched_query(
db_session, db_engine
):
"""add_all() loads semantic-duplicate candidates in one batched SELECT
instead of one find_semantic_duplicate() query per candidate."""
from sqlalchemy import event
db = BangumiDatabase(db_session)
for i in range(5):
await db.add(
Bangumi(
official_title=f"Anime {i}",
title_raw=f"Raw {i}",
group_name="GroupA",
dpi="1080p",
source="Web",
subtitle="CHT",
rss_link=f"rss{i}",
)
)
batch = [
Bangumi(
official_title=f"Anime {i}",
title_raw=f"Raw {i} New",
group_name="GroupA&GroupB",
dpi="1080p",
source="Web",
subtitle="CHT",
rss_link=f"rss{i}-new",
)
for i in range(5)
]
select_statements = []
def _record(conn, cursor, statement, parameters, context, executemany):
if statement.strip().upper().startswith("SELECT") and "bangumi" in statement:
select_statements.append(statement)
event.listen(db_engine.sync_engine, "before_cursor_execute", _record)
try:
await db.add_all(batch)
finally:
event.remove(db_engine.sync_engine, "before_cursor_execute", _record)
# Exactly one SELECT for the exact-duplicate check plus one for the
# semantic-duplicate candidates: O(1), not one per batch item (would be
# 1 + 5 = 6 before the fix).
assert len(select_statements) == 2
class TestDeleteByBangumiId:
"""Tests for TorrentDatabase.delete_by_bangumi_id."""
@@ -452,7 +550,11 @@ class TestDeleteByBangumiId:
db = TorrentDatabase(db_session)
await _ensure_bangumi(db_session, 10)
for i in range(3):
await db.add(Torrent(name=f"torrent_{i}", url=f"https://example.com/{i}", bangumi_id=10))
await db.add(
Torrent(
name=f"torrent_{i}", url=f"https://example.com/{i}", bangumi_id=10
)
)
assert len(await db.search_all()) == 3
count = await db.delete_by_bangumi_id(10)
@@ -463,8 +565,12 @@ class TestDeleteByBangumiId:
db = TorrentDatabase(db_session)
await _ensure_bangumi(db_session, 20)
await _ensure_bangumi(db_session, 30)
await db.add(Torrent(name="keep", url="https://example.com/keep", bangumi_id=20))
await db.add(Torrent(name="delete", url="https://example.com/delete", bangumi_id=30))
await db.add(
Torrent(name="keep", url="https://example.com/keep", bangumi_id=20)
)
await db.add(
Torrent(name="delete", url="https://example.com/delete", bangumi_id=30)
)
count = await db.delete_by_bangumi_id(30)
assert count == 1
@@ -475,7 +581,9 @@ class TestDeleteByBangumiId:
async def test_no_match_returns_zero(self, db_session):
db = TorrentDatabase(db_session)
await _ensure_bangumi(db_session, 5)
await db.add(Torrent(name="unrelated", url="https://example.com/1", bangumi_id=5))
await db.add(
Torrent(name="unrelated", url="https://example.com/1", bangumi_id=5)
)
count = await db.delete_by_bangumi_id(999)
assert count == 0
@@ -484,8 +592,12 @@ class TestDeleteByBangumiId:
async def test_skips_null_bangumi_id(self, db_session):
db = TorrentDatabase(db_session)
await _ensure_bangumi(db_session, 7)
await db.add(Torrent(name="orphan", url="https://example.com/orphan", bangumi_id=None))
await db.add(Torrent(name="target", url="https://example.com/target", bangumi_id=7))
await db.add(
Torrent(name="orphan", url="https://example.com/orphan", bangumi_id=None)
)
await db.add(
Torrent(name="target", url="https://example.com/target", bangumi_id=7)
)
count = await db.delete_by_bangumi_id(7)
assert count == 1
@@ -609,10 +721,16 @@ class TestRSSDeleteWithTorrents:
rss = RSSItem(url="https://mikanani.me/RSS/test", name="Test RSS")
await rss_db.add(rss)
await torrent_db.add(Torrent(name="ep01", url="https://example.com/1", rss_id=rss.id))
await torrent_db.add(Torrent(name="ep02", url="https://example.com/2", rss_id=rss.id))
await torrent_db.add(
Torrent(name="ep01", url="https://example.com/1", rss_id=rss.id)
)
await torrent_db.add(
Torrent(name="ep02", url="https://example.com/2", rss_id=rss.id)
)
# 不关联此 RSS 的 torrent
await torrent_db.add(Torrent(name="other", url="https://example.com/3", rss_id=None))
await torrent_db.add(
Torrent(name="other", url="https://example.com/3", rss_id=None)
)
assert len(await torrent_db.search_rss(rss.id)) == 2
@@ -647,7 +765,9 @@ class TestRSSDeleteWithTorrents:
for i in range(5):
await torrent_db.add(
Torrent(name=f"ep{i:02d}", url=f"https://example.com/{i}", rss_id=rss.id)
Torrent(
name=f"ep{i:02d}", url=f"https://example.com/{i}", rss_id=rss.id
)
)
assert len(await torrent_db.search_rss(rss.id)) == 5
@@ -668,10 +788,16 @@ class TestRSSDeleteWithTorrents:
await rss_db.add(rss1)
await rss_db.add(rss2)
await torrent_db.add(Torrent(name="t1", url="https://example.com/1", rss_id=rss1.id))
await torrent_db.add(Torrent(name="t2", url="https://example.com/2", rss_id=rss2.id))
await torrent_db.add(
Torrent(name="t1", url="https://example.com/1", rss_id=rss1.id)
)
await torrent_db.add(
Torrent(name="t2", url="https://example.com/2", rss_id=rss2.id)
)
# rss_id 为 None 的 torrent 应不受影响
await torrent_db.add(Torrent(name="orphan", url="https://example.com/3", rss_id=None))
await torrent_db.add(
Torrent(name="orphan", url="https://example.com/3", rss_id=None)
)
await rss_db.delete_all()

View File

@@ -0,0 +1,23 @@
"""Tests for the Mikan homepage parser's module-level cache."""
import importlib
# `module.parser.analyser.__init__` re-exports the `mikan_parser` function under
# the same name as this submodule, shadowing the submodule on the package
# object — so `import module.parser.analyser.mikan_parser as x` would resolve
# to the function, not the module. Go through importlib to get the module.
mikan_parser_module = importlib.import_module("module.parser.analyser.mikan_parser")
def test_reset_cache_clears_mikan_cache():
"""reset_cache() must drop all cached homepage lookups (called on config
reload so a changed endpoint stops serving stale results)."""
mikan_parser_module._mikan_cache["https://mikanani.me/Home/Episode/stale"] = (
"",
"Stale Title",
)
assert len(mikan_parser_module._mikan_cache) > 0
mikan_parser_module.reset_cache()
assert len(mikan_parser_module._mikan_cache) == 0

View File

@@ -1,6 +1,6 @@
"""Tests for search providers: URL construction, keyword handling."""
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pytest
@@ -92,9 +92,12 @@ class TestSpecialUrl:
subtitle="CHT",
)
with patch("module.searcher.provider.SEARCH_CONFIG", {
"mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
}):
with patch(
"module.searcher.provider.SEARCH_CONFIG",
{
"mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
},
):
result = SearchTorrent.special_url(bangumi, "mikan")
assert isinstance(result, RSSItem)
@@ -116,10 +119,59 @@ class TestSpecialUrl:
subtitle=None,
)
with patch("module.searcher.provider.SEARCH_CONFIG", {
"mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
}):
with patch(
"module.searcher.provider.SEARCH_CONFIG",
{
"mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
},
):
result = SearchTorrent.special_url(bangumi, "mikan")
# Only title_raw should be in the URL
assert "Test" in result.url
# ---------------------------------------------------------------------------
# _poster_cache: bounded LRU + reset_cache()
# ---------------------------------------------------------------------------
class TestPosterCache:
@pytest.fixture(autouse=True)
def _clean_poster_cache(self):
"""Isolate the module-level poster cache between tests."""
from module.searcher import searcher as searcher_module
searcher_module.reset_cache()
yield
searcher_module.reset_cache()
def test_reset_cache_clears_poster_cache(self):
from module.searcher import searcher as searcher_module
searcher_module._poster_cache["Test Anime"] = "http://example.com/p.jpg"
assert len(searcher_module._poster_cache) > 0
searcher_module.reset_cache()
assert len(searcher_module._poster_cache) == 0
async def test_poster_cache_evicts_oldest_when_full(self, monkeypatch):
"""_poster_cache is bounded (LRU-ish) like _tmdb_cache/_mikan_cache,
instead of growing without limit for the life of the process."""
from module.searcher import searcher as searcher_module
from module.searcher.searcher import SearchTorrent
monkeypatch.setattr(searcher_module, "_POSTER_CACHE_MAX", 3)
torrent = SearchTorrent()
with patch(
"module.searcher.searcher.tmdb_parser", new=AsyncMock(return_value=None)
):
for i in range(4):
await torrent._fetch_tmdb_poster(f"Title {i}")
assert len(searcher_module._poster_cache) == 3
# The oldest entry ("Title 0") was evicted; the rest remain.
assert "Title 0" not in searcher_module._poster_cache
assert "Title 3" in searcher_module._poster_cache

View File

@@ -1,20 +1,31 @@
from pathlib import Path
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from module.api.deps import get_context
from module.api.setup import SENTINEL_PATH, router
@pytest.fixture
def client():
def mock_ctx():
"""Mock AppContext whose reload_settings is an awaitable no-op."""
ctx = MagicMock()
ctx.reload_settings = AsyncMock()
return ctx
@pytest.fixture
def client(mock_ctx):
"""Create a test client for the FastAPI app."""
from fastapi import FastAPI
app = FastAPI()
app.include_router(router, prefix="/api/v1")
return TestClient(app)
app.dependency_overrides[get_context] = lambda: mock_ctx
yield TestClient(app)
app.dependency_overrides.clear()
@pytest.fixture
@@ -293,6 +304,56 @@ class TestRequestValidation:
assert response.status_code == 422
class TestSetupComplete:
"""Issue: setup wizard must route config reload through AppContext, not
mutate settings.__dict__ directly (the shared httpx client, notifier, and
scheduler must be rebuilt after first-run setup)."""
@staticmethod
def _payload():
return {
"username": "testuser",
"password": "testpassword123",
"downloader_type": "qbittorrent",
"downloader_host": "localhost:8080",
"downloader_username": "admin",
"downloader_password": "admin",
"downloader_path": "/downloads",
"downloader_ssl": False,
"rss_url": "",
"rss_name": "",
"notification_enable": False,
"notification_type": "telegram",
"notification_token": "",
"notification_chat_id": "",
}
def test_complete_routes_through_ctx_reload_settings(
self, client, mock_first_run, mock_ctx
):
"""A successful /setup/complete call saves the config and awaits
ctx.reload_settings(), instead of poking settings.__dict__ directly."""
from fastapi import HTTPException
with patch("module.database.Database") as mock_db_cls:
db_instance = AsyncMock()
# No admin account yet: _require_default_admin_or_authenticated()
# treats this as a fresh install and lets setup proceed.
db_instance.user.get_user = AsyncMock(
side_effect=HTTPException(status_code=404, detail="not found")
)
db_instance.user.update_user = AsyncMock(return_value=True)
mock_db_cls.return_value.__aenter__ = AsyncMock(return_value=db_instance)
mock_db_cls.return_value.__aexit__ = AsyncMock(return_value=False)
response = client.post("/api/v1/setup/complete", json=self._payload())
assert response.status_code == 200
data = response.json()
assert data["status"] is True
mock_ctx.reload_settings.assert_awaited_once()
class TestSentinelPath:
def test_sentinel_path_is_in_config_dir(self):
assert str(SENTINEL_PATH) == "config/.setup_complete"
@@ -379,9 +440,7 @@ class TestTestDownloaderHardening:
from unittest.mock import MagicMock
get_resp = MagicMock(text="qBittorrent WebUI")
login_resp = MagicMock(
status_code=200, text="<html><body>portal</body></html>"
)
login_resp = MagicMock(status_code=200, text="<html><body>portal</body></html>")
cls_patch = self._mock_client(get_resp=get_resp, login_resp=login_resp)
try:
response = self._post(client)

View File

@@ -1,6 +1,7 @@
import pytest
from module.conf import settings
from module.parser import title_parser as title_parser_module
from module.parser.title_parser import TitleParser
@@ -26,3 +27,20 @@ class TestTitleParser:
assert result.dpi == "1080P"
assert result.season == 1
assert result.subtitle == "GB_JP"
class TestOpenAIParserResetCache:
"""reset_cache() must drop the lazy OpenAI-parser singleton on config
reload, or a changed api_key/api_base keeps using the old client."""
def teardown_method(self):
title_parser_module.reset_cache()
def test_reset_cache_clears_singleton(self):
title_parser_module._openai_parser = object()
title_parser_module._openai_parser_kwargs = {"api_key": "old"}
title_parser_module.reset_cache()
assert title_parser_module._openai_parser is None
assert title_parser_module._openai_parser_kwargs is None

View File

@@ -70,6 +70,17 @@ async def test_tmdb_parser(mocker):
assert tmdb_info.last_season == bangumi_season
def test_reset_cache_clears_tmdb_cache():
"""reset_cache() must drop all cached lookups (called on config reload so
a changed tmdb_base_url stops serving results from the old endpoint)."""
tmdb_parser_module._tmdb_cache["stale-key"] = None
assert len(tmdb_parser_module._tmdb_cache) > 0
tmdb_parser_module.reset_cache()
assert len(tmdb_parser_module._tmdb_cache) == 0
@pytest.mark.live
@pytest.mark.skipif(
not os.environ.get("RUN_LIVE_TMDB_TESTS"),