nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-11-01 21:26:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): replace Program god-object with lifespan composition root

Browse Source 
AppContext is built in create_app and owned by the lifespan: startup() is
awaited before serving (errors abort boot), and start_tasks() runs as a
supervised background task. A generic PeriodicTask/Scheduler replaces the four
Thread mixins, with per-tick bodies moved to plain functions in loops.py. The
MCP app now takes the context instead of importing the api.program global.
Config updates go through reload_settings(), restoring the RSS re-application
that was previously commented out. SessionStore adds session expiry and
constant-time token comparison, and downloader shutdown() is wired into app
shutdown.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014w1Z6Nxy6XTRgkFXqPr9Zh
This commit is contained in:
Estrella Pan
2026-07-02 13:39:24 +02:00
parent 5c73aa0013
commit efdfacfbae
31 changed files with 1322 additions and 608 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
backend/src/main.py
View File

@@ -1,3 +1,4 @@
import asyncio
import logging
import os
from contextlib import asynccontextmanager
@@ -11,8 +12,8 @@ from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from module.api import v1
from module.api.program import program
from module.conf import VERSION, settings, setup_logger
from module.core import AppContext
from module.mcp import create_mcp_app
setup_logger(reset=True)
@@ -32,19 +33,34 @@ uvicorn_logging_config = {
}
def _supervise_start_tasks(task: asyncio.Task) -> None:
"""Log-and-surface any exception from the background start task."""
if task.cancelled():
return
exc = task.exception()
if exc is not None:
logger.error("[Core] Background start task failed", exc_info=exc)
@asynccontextmanager
async def lifespan(app: FastAPI):
import asyncio
# Startup
asyncio.create_task(program.startup())
ctx: AppContext = app.state.ctx
# Run migrations before serving; failure here aborts boot loudly.
await ctx.startup()
# First run just created the DB — do not auto-start loops (matches the old
# Program.startup early return); the user starts them after setup.
if not ctx.first_run_boot:
start_task = asyncio.create_task(ctx.start_tasks())
start_task.add_done_callback(_supervise_start_tasks)
yield
# Shutdown
await program.stop()
await ctx.stop()
def create_app() -> FastAPI:
ctx = AppContext.build(settings)
app = FastAPI(lifespan=lifespan)
app.state.ctx = ctx
app.add_middleware(
CORSMiddleware,
@@ -58,7 +74,7 @@ def create_app() -> FastAPI:
app.include_router(v1, prefix="/api")
# mount MCP server (SSE transport for LLM tool integration)
app.mount("/mcp", create_mcp_app())
app.mount("/mcp", create_mcp_app(ctx))
return app

6
backend/src/module/api/auth.py
View File

@@ -1,4 +1,4 @@
from datetime import datetime, timedelta
from datetime import timedelta
from fastapi import APIRouter, Cookie, Depends, HTTPException, status
from fastapi.responses import JSONResponse, Response
@@ -53,7 +53,7 @@ async def refresh(response: Response, token: str = Cookie(None)):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized"
)
active_user[username] = datetime.now()
active_user.add(username)
return _issue_token(username, response)
@@ -65,7 +65,7 @@ async def logout(response: Response, token: str = Cookie(None)):
payload = decode_token(token)
username = payload.get("sub") if payload else None
if username:
active_user.pop(username, None)
active_user.remove(username)
response.delete_cookie(key="token")
return JSONResponse(
status_code=200,

10
backend/src/module/api/config.py
View File

@@ -4,9 +4,12 @@ from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from module.conf import settings
from module.core import AppContext
from module.models import APIResponse, Config
from module.security.api import UNAUTHORIZED, get_current_user
from .deps import get_context
router = APIRouter(prefix="/config", tags=["config"])
logger = logging.getLogger(__name__)
@@ -63,13 +66,14 @@ async def get_config():
@router.patch(
"/update", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def update_config(config: Config):
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.load()
# update_rss()
# reload_settings reloads from disk, resets the shared HTTP client,
# rebuilds notifications, and re-applies the RSS/rename loops.
await ctx.reload_settings()
logger.info("Config updated")
return JSONResponse(
status_code=200,

10
backend/src/module/api/deps.py Normal file
View File

@@ -0,0 +1,10 @@
"""FastAPI dependencies shared across API routers."""
from fastapi import Request
from module.core import AppContext
def get_context(request: Request) -> AppContext:
"""Return the lifespan-owned :class:`AppContext` stored on ``app.state``."""
return request.app.state.ctx

4
backend/src/module/api/passkey.py
View File

@@ -4,7 +4,7 @@ Passkey 管理 API
"""
import logging
from datetime import datetime, timedelta
from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import JSONResponse, Response
@@ -234,7 +234,7 @@ async def login_with_passkey(
data={"sub": username}, expires_delta=timedelta(days=1)
)
response.set_cookie(key="token", value=token, httponly=True, max_age=86400)
active_user[username] = datetime.now()
active_user.add(username)
return {"access_token": token, "token_type": "bearer"}
raise HTTPException(status_code=resp.status_code, detail=resp.msg_en)

37
backend/src/module/api/program.py
View File

@@ -6,26 +6,27 @@ from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from module.conf import VERSION
from module.core import Program
from module.core import AppContext
from module.models import APIResponse
from module.security.api import UNAUTHORIZED, get_current_user
from module.security.api import get_current_user
from .deps import get_context
from .response import u_response
logger = logging.getLogger(__name__)
program = Program()
router = APIRouter(tags=["program"])
# Note: Lifespan events (startup/shutdown) are now handled in main.py via lifespan context manager
# Note: Lifespan events (startup/shutdown) are handled in main.py via the
# lifespan context manager; the AppContext is injected here per-request.
@router.get(
"/restart", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def restart():
async def restart(ctx: AppContext = Depends(get_context)):
try:
resp = await program.restart()
resp = await ctx.restart()
return u_response(resp)
except Exception as e:
logger.debug(e)
@@ -42,9 +43,9 @@ async def restart():
@router.get(
"/start", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def start():
async def start(ctx: AppContext = Depends(get_context)):
try:
resp = await program.start()
resp = await ctx.start_tasks()
return u_response(resp)
except Exception as e:
logger.debug(e)
@@ -61,32 +62,32 @@ async def start():
@router.get(
"/stop", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def stop():
resp = await program.stop()
async def stop(ctx: AppContext = Depends(get_context)):
resp = await ctx.stop()
return u_response(resp)
@router.get("/status", response_model=dict, dependencies=[Depends(get_current_user)])
async def program_status():
if not program.is_running:
async def program_status(ctx: AppContext = Depends(get_context)):
if not ctx.is_running:
return {
"status": False,
"version": VERSION,
"first_run": program.first_run,
"first_run": ctx.first_run,
}
else:
return {
"status": True,
"version": VERSION,
"first_run": program.first_run,
"first_run": ctx.first_run,
}
@router.get(
"/shutdown", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def shutdown_program():
await program.stop()
async def shutdown_program(ctx: AppContext = Depends(get_context)):
await ctx.stop()
logger.info("Shutting down program...")
os.kill(os.getpid(), signal.SIGINT)
return JSONResponse(
@@ -105,5 +106,5 @@ async def shutdown_program():
response_model=bool,
dependencies=[Depends(get_current_user)],
)
async def check_downloader_status():
return await program.check_downloader()
async def check_downloader_status(ctx: AppContext = Depends(get_context)):
return await ctx.check_downloader()

6
backend/src/module/core/__init__.py
View File

@@ -1 +1,5 @@
from .program import Program
from .context import AppContext
from .offset_scanner import OffsetScanner
from .scheduler import PeriodicTask, Scheduler
__all__ = ["AppContext", "OffsetScanner", "PeriodicTask", "Scheduler"]

298
backend/src/module/core/context.py Normal file
View File

@@ -0,0 +1,298 @@
"""应用组合根（composition root）。
``AppContext`` 取代原先的 ``Program`` 上帝对象：由 ``create_app`` 构造、由
lifespan 拥有其生命周期。它持有 settings、通知管理器、调度器，以及下载器状态的
TTL 缓存，并负责启动迁移、后台任务的启停与配置热重载。
"""
import asyncio
import logging
import time
from module.checker import Checker
from module.conf import LEGACY_DATA_PATH, VERSION, settings
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.rss import RSSAnalyser
from module.update import (
cache_image,
data_migration,
first_run,
from_30_to_31,
from_31_to_32,
run_migrations,
)
from .loops import calendar_tick, offset_scan_tick, rename_tick, rss_tick
from .scheduler import PeriodicTask, Scheduler
logger = logging.getLogger(__name__)
# Downloader-status cache TTL, in seconds (moved from the old ProgramStatus).
DOWNLOADER_STATUS_TTL = 60
# Background loop cadence (seconds).
OFFSET_SCAN_INTERVAL = 6 * 60 * 60
CALENDAR_REFRESH_INTERVAL = 24 * 60 * 60
OFFSET_SCAN_INITIAL_DELAY = 60
CALENDAR_INITIAL_DELAY = 120
# Downloader wait-retry loop on startup.
_DOWNLOADER_MAX_RETRIES = 10
_DOWNLOADER_RETRY_INTERVAL = 30
figlet = r"""
_ ____ _
/\ | | | _ \ (_)
/ \ _ _| |_ ___ | |_) | __ _ _ __ __ _ _ _ _ __ ___ _
/ /\ \| | | | __/ _ \| _ < / _` | '_ \ / _` | | | | '_ ` _ \| |
/ ____ \ |_| | || (_) | |_) | (_| | | | | (_| | |_| | | | | | | |
/_/ \_\__,_|\__\___/|____/ \__,_|_| |_|\__, |\__,_|_| |_| |_|_|
__/ |
|___/
"""
class AppContext:
"""Owns application-wide state and orchestrates startup/shutdown/reload."""
def __init__(
self,
settings_obj,
notifier: NotificationManager,
scheduler: Scheduler,
analyser: RSSAnalyser,
) -> None:
self.settings = settings_obj
self.notifier = notifier
self.scheduler = scheduler
self.analyser = analyser
# Downloader-status TTL cache (was ProgramStatus.check_downloader_status).
self._downloader_status = False
self._downloader_last_check: float = 0.0
self._tasks_started = False
self._startup_done = False
# True when this boot created the database (genuine first run); the
# lifespan then skips auto-starting background tasks, matching the old
# Program.startup() early-return.
self.first_run_boot = False
# ------------------------------------------------------------------ build
@classmethod
def build(cls, settings_obj) -> "AppContext":
"""Pure wiring: construct the object graph, perform no I/O.
The scheduler closures capture ``notifier`` / ``analyser`` directly.
``NotificationManager.rebuild()`` mutates in place, so a config reload is
reflected without re-wiring the tasks; ``interval`` reads settings live.
"""
analyser = RSSAnalyser()
notifier = NotificationManager()
scheduler = Scheduler(
[
PeriodicTask(
name="rss",
run=lambda: rss_tick(analyser),
interval=lambda: settings_obj.program.rss_time,
enabled=Checker.check_analyser,
),
PeriodicTask(
name="rename",
run=lambda: rename_tick(notifier),
interval=lambda: settings_obj.program.rename_time,
enabled=Checker.check_renamer,
),
PeriodicTask(
name="offset_scan",
run=offset_scan_tick,
interval=lambda: OFFSET_SCAN_INTERVAL,
initial_delay=OFFSET_SCAN_INITIAL_DELAY,
),
PeriodicTask(
name="calendar",
run=calendar_tick,
interval=lambda: CALENDAR_REFRESH_INTERVAL,
initial_delay=CALENDAR_INITIAL_DELAY,
),
]
)
return cls(settings_obj, notifier, scheduler, analyser)
# ------------------------------------------------------------------ status
@property
def is_running(self) -> bool:
return self._tasks_started and not Checker.check_first_run()
@property
def first_run(self) -> bool:
return Checker.check_first_run()
async def check_downloader(self) -> bool:
"""Raw downloader reachability check (no cache), used by /check/downloader."""
return await Checker.check_downloader()
async def check_downloader_status(self) -> bool:
"""TTL-cached downloader reachability, used by the startup wait loop."""
now = time.monotonic()
if (
not self._downloader_status
or (now - self._downloader_last_check) >= DOWNLOADER_STATUS_TTL
):
self._downloader_status = await Checker.check_downloader()
self._downloader_last_check = now
return self._downloader_status
# ------------------------------------------------------------------ startup
@staticmethod
def _start_info() -> None:
for line in figlet.splitlines():
logger.info(line.strip("\n"))
logger.info(
f"Version {VERSION} Author: EstrellaXD Twitter: https://twitter.com/Estrella_Pan"
)
logger.info("GitHub: https://github.com/EstrellaXD/Auto_Bangumi/")
logger.info("Starting AutoBangumi...")
async def startup(self) -> None:
"""Run first-run detection and database migrations.
Awaited from the lifespan *before* serving requests, so any failure
aborts boot loudly instead of being swallowed. Does not start tasks.
"""
if self._startup_done:
return
self._start_info()
if not Checker.check_database():
first_run()
logger.info("[Core] No db file exists, create database file.")
self.first_run_boot = True
self._startup_done = True
return
if LEGACY_DATA_PATH.exists():
logger.info(
"[Core] Legacy data detected, starting data migration, please wait patiently."
)
data_migration()
else:
is_same, last_minor = Checker.check_version()
if not is_same:
if last_minor is not None and last_minor == 0:
await from_30_to_31()
logger.info("[Core] Database migrated from 3.0 to 3.1.")
await from_31_to_32()
logger.info("[Core] Database updated.")
else:
# Always check schema version and run pending migrations,
# in case a previous migration was interrupted or failed.
run_migrations()
if not Checker.check_img_cache():
logger.info("[Core] No image cache exists, create image cache.")
await cache_image()
self._startup_done = True
# ------------------------------------------------------------------ tasks
async def _wait_for_downloader(self) -> None:
retry_count = 0
while not await self.check_downloader_status():
retry_count += 1
logger.warning(
f"Downloader is not running. (attempt {retry_count}/{_DOWNLOADER_MAX_RETRIES})"
)
if retry_count >= _DOWNLOADER_MAX_RETRIES:
logger.error(
"Failed to connect to downloader after maximum retries. "
"Please check downloader settings and network/proxy configuration. "
"Program will continue but download functions will not work."
)
break
logger.info("Waiting for downloader to start...")
await asyncio.sleep(_DOWNLOADER_RETRY_INTERVAL)
async def start_tasks(self) -> ResponseModel:
"""Wait for the downloader, then start all enabled background loops."""
settings.load()
await self._wait_for_downloader()
self.scheduler.start_all()
self._tasks_started = True
logger.info("Program running.")
return ResponseModel(
status=True,
status_code=200,
msg_en="Program started.",
msg_zh="程序启动成功。",
)
async def stop(self) -> ResponseModel:
"""Stop background loops and close the cached downloader session."""
if self.is_running:
await self.scheduler.stop_all()
self._tasks_started = False
resp = ResponseModel(
status=True,
status_code=200,
msg_en="Program stopped.",
msg_zh="程序停止成功。",
)
else:
resp = ResponseModel(
status=False,
status_code=406,
msg_en="Program is not running.",
msg_zh="程序未运行。",
)
# Always release the shared qB session; shutdown() is idempotent and the
# session is lazily rebuilt on the next DownloadClient() use.
await downloader_shutdown()
return resp
async def restart(self) -> ResponseModel:
stop_ok = True
try:
await self.stop()
except Exception as e:
logger.warning(f"[Core] Error during stop in restart: {e}")
stop_ok = False
start_ok = True
try:
await self.start_tasks()
except Exception as e:
logger.error(f"[Core] Error during start in restart: {e}")
start_ok = False
if start_ok and stop_ok:
return ResponseModel(
status=True,
status_code=200,
msg_en="Program restarted.",
msg_zh="程序重启成功。",
)
elif start_ok:
return ResponseModel(
status=True,
status_code=200,
msg_en="Program restarted (stop had warnings).",
msg_zh="程序重启成功（停止时有警告）。",
)
else:
return ResponseModel(
status=False,
status_code=500,
msg_en="Program failed to restart.",
msg_zh="程序重启失败。",
)
# ------------------------------------------------------------------ reload
async def reload_settings(self) -> None:
"""Single choreography for a config update: reload + re-apply live state."""
settings.load()
await reset_shared_client()
self.notifier.rebuild()
if self.scheduler.running:
await self.scheduler.stop_all()
self.scheduler.start_all()

70
backend/src/module/core/loops.py Normal file
View File

@@ -0,0 +1,70 @@
"""周期任务的单次 tick 逻辑。
每个函数就是原 ``sub_thread.py`` 里循环体的内容，去掉了 ``self`` 依赖，改为
显式参数注入。异常由 :class:`PeriodicTask` 统一捕获，这里不再包裹顶层 try。
"""
import logging
from module.conf import settings
from module.database import Database
from module.downloader import DownloadClient
from module.manager import Renamer, TorrentManager, eps_complete
from module.notification import NotificationManager
from module.rss import RSSAnalyser, RSSEngine
from .offset_scanner import OffsetScanner
logger = logging.getLogger(__name__)
async def rss_tick(analyser: RSSAnalyser) -> None:
"""Analyse aggregate RSS feeds and refresh the RSS engine once."""
async with DownloadClient() as client:
with Database() as db:
engine = RSSEngine(db)
# Analyse RSS
rss_list = engine.rss.search_aggregate()
for rss in rss_list:
try:
await analyser.rss_to_data(rss, engine)
except Exception:
# RSS 可能在遍历期间被 API 删除，跳过即可
logger.debug(
"[RSSThread] Skipping RSS id=%s, likely deleted",
rss.id if hasattr(rss, "id") else "?",
)
# Run RSS Engine
await engine.refresh_rss(client)
if settings.bangumi_manage.eps_complete:
await eps_complete()
async def rename_tick(notifier: NotificationManager) -> None:
"""Rename completed downloads and notify via the shared notifier."""
async with DownloadClient() as client:
renamer = Renamer(client)
renamed_info = await renamer.rename()
if settings.notification.enable and renamed_info:
for info in renamed_info:
await notifier.send_all(info)
async def offset_scan_tick() -> None:
"""Scan all bangumi for season/episode offset mismatches."""
scanner = OffsetScanner()
flagged = await scanner.scan_all()
logger.info("[OffsetScanThread] Scan complete, flagged %s bangumi", flagged)
async def calendar_tick() -> None:
"""Refresh bangumi calendar metadata."""
with Database() as db:
manager = TorrentManager(db)
resp = await manager.refresh_calendar()
if resp.status:
logger.info("[CalendarRefreshThread] Calendar refresh completed")
else:
logger.warning(
"[CalendarRefreshThread] Calendar refresh failed: %s", resp.msg_en
)

176
backend/src/module/core/program.py
View File

@@ -1,176 +0,0 @@
import asyncio
import logging
from module.conf import VERSION, settings
from module.models import ResponseModel
from module.update import (
cache_image,
data_migration,
first_run,
from_30_to_31,
from_31_to_32,
run_migrations,
start_up,
)
from .sub_thread import CalendarRefreshThread, OffsetScanThread, RenameThread, RSSThread
logger = logging.getLogger(__name__)
figlet = r"""
_ ____ _
/\ | | | _ \ (_)
/ \ _ _| |_ ___ | |_) | __ _ _ __ __ _ _ _ _ __ ___ _
/ /\ \| | | | __/ _ \| _ < / _` | '_ \ / _` | | | | '_ ` _ \| |
/ ____ \ |_| | || (_) | |_) | (_| | | | | (_| | |_| | | | | | | |
/_/ \_\__,_|\__\___/|____/ \__,_|_| |_|\__, |\__,_|_| |_| |_|_|
__/ |
|___/
"""
class Program(RenameThread, RSSThread, OffsetScanThread, CalendarRefreshThread):
def __init__(self):
super().__init__()
self._startup_done = False
@staticmethod
def __start_info():
for line in figlet.splitlines():
logger.info(line.strip("\n"))
logger.info(
f"Version {VERSION} Author: EstrellaXD Twitter: https://twitter.com/Estrella_Pan"
)
logger.info("GitHub: https://github.com/EstrellaXD/Auto_Bangumi/")
logger.info("Starting AutoBangumi...")
async def startup(self):
# Prevent duplicate startup due to nested router lifespan events
if self._startup_done:
return
self.__start_info()
if not self.database:
first_run()
logger.info("[Core] No db file exists, create database file.")
return {"status": "First run detected."}
if self.legacy_data:
logger.info(
"[Core] Legacy data detected, starting data migration, please wait patiently."
)
data_migration()
else:
need_update, last_minor = self.version_update
if need_update:
if last_minor is not None and last_minor == 0:
await from_30_to_31()
logger.info("[Core] Database migrated from 3.0 to 3.1.")
await from_31_to_32()
logger.info("[Core] Database updated.")
else:
# Always check schema version and run pending migrations,
# in case a previous migration was interrupted or failed.
run_migrations()
if not self.img_cache:
logger.info("[Core] No image cache exists, create image cache.")
await cache_image()
await self.start()
self._startup_done = True
async def start(self):
settings.load()
max_retries = 10
retry_count = 0
while not await self.check_downloader_status():
retry_count += 1
logger.warning(
f"Downloader is not running. (attempt {retry_count}/{max_retries})"
)
if retry_count >= max_retries:
logger.error(
"Failed to connect to downloader after maximum retries. "
"Please check downloader settings and network/proxy configuration. "
"Program will continue but download functions will not work."
)
break
logger.info("Waiting for downloader to start...")
await asyncio.sleep(30)
if self.enable_renamer:
self.rename_start()
if self.enable_rss:
self.rss_start()
# Start offset scanner for background mismatch detection
self.scan_start()
# Start calendar refresh (every 24 hours)
self.calendar_start()
self._tasks_started = True
logger.info("Program running.")
return ResponseModel(
status=True,
status_code=200,
msg_en="Program started.",
msg_zh="程序启动成功。",
)
async def stop(self):
if self.is_running:
await self.rename_stop()
await self.rss_stop()
await self.scan_stop()
await self.calendar_stop()
self._tasks_started = False
return ResponseModel(
status=True,
status_code=200,
msg_en="Program stopped.",
msg_zh="程序停止成功。",
)
else:
return ResponseModel(
status=False,
status_code=406,
msg_en="Program is not running.",
msg_zh="程序未运行。",
)
async def restart(self):
stop_ok = True
try:
await self.stop()
except Exception as e:
logger.warning(f"[Core] Error during stop in restart: {e}")
stop_ok = False
start_ok = True
try:
await self.start()
except Exception as e:
logger.error(f"[Core] Error during start in restart: {e}")
start_ok = False
if start_ok and stop_ok:
return ResponseModel(
status=True,
status_code=200,
msg_en="Program restarted.",
msg_zh="程序重启成功。",
)
elif start_ok:
return ResponseModel(
status=True,
status_code=200,
msg_en="Program restarted (stop had warnings).",
msg_zh="程序重启成功（停止时有警告）。",
)
else:
return ResponseModel(
status=False,
status_code=500,
msg_en="Program failed to restart.",
msg_zh="程序重启失败。",
)
def update_database(self):
need_update, _ = self.version_update
if not need_update:
return {"status": "No update found."}
else:
start_up()
return {"status": "Database updated."}

123
backend/src/module/core/scheduler.py Normal file
View File

@@ -0,0 +1,123 @@
"""通用周期任务调度器。
``PeriodicTask`` 把「运行一次 tick，然后等待一个间隔」的循环抽象出来，替代
原先四个手写的 Thread 混入类。``Scheduler`` 负责批量启停这些任务。
"""
import asyncio
import logging
from typing import Awaitable, Callable
logger = logging.getLogger(__name__)
class PeriodicTask:
"""A single asyncio loop that runs ``run`` then waits ``interval`` seconds.
Semantics mirror the original hand-written loops:
* ``initial_delay > 0`` waits before the first tick (offset/calendar loops).
* otherwise the body runs immediately, then waits (rss/rename loops).
* ``interval`` and ``enabled`` are callables read live each time, so a
settings change is picked up without rebuilding the task.
* an exception raised inside a tick is logged and the loop continues.
"""
def __init__(
self,
name: str,
run: Callable[[], Awaitable[None]],
interval: Callable[[], float],
initial_delay: float = 0.0,
enabled: Callable[[], bool] = lambda: True,
) -> None:
self._name = name
self._run = run
self._interval = interval
self._initial_delay = initial_delay
self._enabled = enabled
self._task: asyncio.Task | None = None
self._stop_event = asyncio.Event()
@property
def name(self) -> str:
return self._name
@property
def enabled(self) -> bool:
return self._enabled()
@property
def running(self) -> bool:
return self._task is not None and not self._task.done()
def start(self) -> None:
"""Create the asyncio task. Idempotent while already running."""
if self.running:
return
self._stop_event.clear()
self._task = asyncio.create_task(self._loop(), name=f"periodic:{self._name}")
async def _loop(self) -> None:
try:
if self._initial_delay > 0:
if await self._wait(self._initial_delay):
return
while not self._stop_event.is_set():
try:
await self._run()
except asyncio.CancelledError:
raise
except Exception as e:
logger.error("[%s] Error during tick: %s", self._name, e)
if await self._wait(self._interval()):
return
except asyncio.CancelledError:
pass
async def _wait(self, timeout: float) -> bool:
"""Sleep up to ``timeout`` seconds. Return True if a stop was requested."""
try:
await asyncio.wait_for(self._stop_event.wait(), timeout=timeout)
return True
except asyncio.TimeoutError:
return False
async def stop(self) -> None:
"""Signal, cancel, and await the task. Idempotent."""
if self._task is None:
return
self._stop_event.set()
self._task.cancel()
try:
await self._task
except asyncio.CancelledError:
pass
self._task = None
class Scheduler:
"""Owns a fixed set of :class:`PeriodicTask` and starts/stops them together."""
def __init__(self, tasks: list[PeriodicTask]) -> None:
self._tasks = tasks
self._running = False
@property
def tasks(self) -> list[PeriodicTask]:
return self._tasks
@property
def running(self) -> bool:
return self._running
def start_all(self) -> None:
"""Start every task whose ``enabled()`` currently returns True."""
for task in self._tasks:
if task.enabled:
task.start()
self._running = True
async def stop_all(self) -> None:
await asyncio.gather(*(task.stop() for task in self._tasks))
self._running = False

68
backend/src/module/core/status.py
View File

@@ -1,68 +0,0 @@
import time
from module.checker import Checker
from module.conf import LEGACY_DATA_PATH
DOWNLOADER_STATUS_TTL = 60
class ProgramStatus(Checker):
def __init__(self):
super().__init__()
self._downloader_status = False
self._downloader_last_check: float = 0
self._tasks_started = False
@property
def is_running(self):
if not self._tasks_started or self.check_first_run():
return False
else:
return True
@property
def is_stopped(self):
return not self._tasks_started
@property
def downloader_status(self):
return self._downloader_status
async def check_downloader_status(self) -> bool:
now = time.monotonic()
if (
not self._downloader_status
or (now - self._downloader_last_check) >= DOWNLOADER_STATUS_TTL
):
self._downloader_status = await self.check_downloader()
self._downloader_last_check = now
return self._downloader_status
@property
def enable_rss(self):
return self.check_analyser()
@property
def enable_renamer(self):
return self.check_renamer()
@property
def first_run(self):
return self.check_first_run()
@property
def legacy_data(self):
return LEGACY_DATA_PATH.exists()
@property
def version_update(self) -> tuple[bool, int | None]:
is_same, last_minor = self.check_version()
return not is_same, last_minor
@property
def database(self):
return self.check_database()
@property
def img_cache(self):
return self.check_img_cache()

215
backend/src/module/core/sub_thread.py
View File

@@ -1,215 +0,0 @@
import asyncio
import logging
from module.conf import settings
from module.database import Database
from module.downloader import DownloadClient
from module.manager import Renamer, TorrentManager, eps_complete
from module.notification import NotificationManager
from module.rss import RSSAnalyser, RSSEngine
from .offset_scanner import OffsetScanner
from .status import ProgramStatus
logger = logging.getLogger(__name__)
# Calendar refresh interval in seconds (24 hours)
CALENDAR_REFRESH_INTERVAL = 24 * 60 * 60
class RSSThread(ProgramStatus):
def __init__(self):
super().__init__()
self._rss_task: asyncio.Task | None = None
self._rss_stop_event = asyncio.Event()
self.analyser = RSSAnalyser()
async def rss_loop(self):
while not self._rss_stop_event.is_set():
try:
async with DownloadClient() as client:
with Database() as db:
engine = RSSEngine(db)
# Analyse RSS
rss_list = engine.rss.search_aggregate()
for rss in rss_list:
try:
await self.analyser.rss_to_data(rss, engine)
except Exception:
# RSS 可能在遍历期间被 API 删除，跳过即可
logger.debug(
"[RSSThread] Skipping RSS id=%s, likely deleted",
rss.id if hasattr(rss, "id") else "?",
)
# Run RSS Engine
await engine.refresh_rss(client)
if settings.bangumi_manage.eps_complete:
await eps_complete()
except Exception as e:
logger.error(f"[RSSThread] Error during RSS loop: {e}")
try:
await asyncio.wait_for(
self._rss_stop_event.wait(),
timeout=settings.program.rss_time,
)
except asyncio.TimeoutError:
pass
def rss_start(self):
self._rss_stop_event.clear()
self._rss_task = asyncio.create_task(self.rss_loop())
async def rss_stop(self):
if self._rss_task and not self._rss_task.done():
self._rss_stop_event.set()
self._rss_task.cancel()
try:
await self._rss_task
except asyncio.CancelledError:
pass
self._rss_task = None
class RenameThread(ProgramStatus):
def __init__(self):
super().__init__()
self._rename_task: asyncio.Task | None = None
self._rename_stop_event = asyncio.Event()
async def rename_loop(self):
while not self._rename_stop_event.is_set():
try:
async with DownloadClient() as client:
renamer = Renamer(client)
renamed_info = await renamer.rename()
if settings.notification.enable and renamed_info:
manager = NotificationManager()
for info in renamed_info:
await manager.send_all(info)
except Exception as e:
logger.error(f"[RenameThread] Error during rename loop: {e}")
try:
await asyncio.wait_for(
self._rename_stop_event.wait(),
timeout=settings.program.rename_time,
)
except asyncio.TimeoutError:
pass
def rename_start(self):
self._rename_stop_event.clear()
self._rename_task = asyncio.create_task(self.rename_loop())
async def rename_stop(self):
if self._rename_task and not self._rename_task.done():
self._rename_stop_event.set()
self._rename_task.cancel()
try:
await self._rename_task
except asyncio.CancelledError:
pass
self._rename_task = None
# Offset scan interval in seconds (6 hours)
OFFSET_SCAN_INTERVAL = 6 * 60 * 60
class OffsetScanThread(ProgramStatus):
"""Background thread for scanning bangumi offset mismatches."""
def __init__(self):
super().__init__()
self._scan_task: asyncio.Task | None = None
self._scan_stop_event = asyncio.Event()
self._scanner = OffsetScanner()
async def scan_loop(self):
# Initial delay to let the system stabilize
await asyncio.sleep(60)
while not self._scan_stop_event.is_set():
try:
flagged = await self._scanner.scan_all()
logger.info(
f"[OffsetScanThread] Scan complete, flagged {flagged} bangumi"
)
except Exception as e:
logger.error(f"[OffsetScanThread] Error during scan: {e}")
try:
await asyncio.wait_for(
self._scan_stop_event.wait(),
timeout=OFFSET_SCAN_INTERVAL,
)
except asyncio.TimeoutError:
pass
def scan_start(self):
self._scan_stop_event.clear()
self._scan_task = asyncio.create_task(self.scan_loop())
logger.info("[OffsetScanThread] Started offset scanner")
async def scan_stop(self):
if self._scan_task and not self._scan_task.done():
self._scan_stop_event.set()
self._scan_task.cancel()
try:
await self._scan_task
except asyncio.CancelledError:
pass
self._scan_task = None
logger.info("[OffsetScanThread] Stopped offset scanner")
class CalendarRefreshThread(ProgramStatus):
"""Background thread for refreshing bangumi calendar data every 24 hours."""
def __init__(self):
super().__init__()
self._calendar_task: asyncio.Task | None = None
self._calendar_stop_event = asyncio.Event()
async def calendar_loop(self):
# Initial delay to let the system stabilize
await asyncio.sleep(120)
while not self._calendar_stop_event.is_set():
try:
with Database() as db:
manager = TorrentManager(db)
resp = await manager.refresh_calendar()
if resp.status:
logger.info(
"[CalendarRefreshThread] Calendar refresh completed"
)
else:
logger.warning(
f"[CalendarRefreshThread] Calendar refresh failed: {resp.msg_en}"
)
except Exception as e:
logger.error(f"[CalendarRefreshThread] Error during refresh: {e}")
try:
await asyncio.wait_for(
self._calendar_stop_event.wait(),
timeout=CALENDAR_REFRESH_INTERVAL,
)
except asyncio.TimeoutError:
pass
def calendar_start(self):
self._calendar_stop_event.clear()
self._calendar_task = asyncio.create_task(self.calendar_loop())
logger.info("[CalendarRefreshThread] Started calendar refresh (every 24h)")
async def calendar_stop(self):
if self._calendar_task and not self._calendar_task.done():
self._calendar_stop_event.set()
self._calendar_task.cancel()
try:
await self._calendar_task
except asyncio.CancelledError:
pass
self._calendar_task = None
logger.info("[CalendarRefreshThread] Stopped calendar refresh")

2
backend/src/module/mcp/__init__.py
View File

@@ -7,7 +7,7 @@ Usage::
from module.mcp import create_mcp_app
app = create_mcp_app() # returns a Starlette ASGI app, mount at /mcp
app = create_mcp_app(ctx) # returns a Starlette ASGI app, mount at /mcp
"""
from .server import create_mcp_starlette_app as create_mcp_app

8
backend/src/module/mcp/resources.py
View File

@@ -16,6 +16,7 @@ from module.manager import TorrentManager
from module.models import Bangumi
from module.rss import RSSEngine
from .runtime import get_context
from .tools import _bangumi_to_dict
logger = logging.getLogger(__name__)
@@ -67,13 +68,12 @@ def handle_resource(uri: str) -> str:
return json.dumps([_bangumi_to_dict(b) for b in items], ensure_ascii=False)
elif uri == "autobangumi://status":
from module.api.program import program
ctx = get_context()
return json.dumps(
{
"version": VERSION,
"running": program.is_running,
"first_run": program.first_run,
"running": ctx.is_running if ctx is not None else False,
"first_run": ctx.first_run if ctx is not None else True,
}
)

23
backend/src/module/mcp/runtime.py Normal file
View File

@@ -0,0 +1,23 @@
"""Runtime holder for the MCP server's :class:`AppContext`.
The MCP handlers are invoked through the module-level SSE ``Server`` singleton,
outside FastAPI's dependency injection. ``create_mcp_app(ctx)`` stashes the
context here so status tools/resources can read live program state without
importing the old ``module.api.program`` global.
"""
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from module.core import AppContext
_context: "Optional[AppContext]" = None
def set_context(ctx: "Optional[AppContext]") -> None:
global _context
_context = ctx
def get_context() -> "Optional[AppContext]":
return _context

9
backend/src/module/mcp/server.py
View File

@@ -19,6 +19,7 @@ from starlette.responses import Response
from starlette.routing import Mount, Route
from .resources import RESOURCE_TEMPLATES, RESOURCES, handle_resource
from .runtime import set_context
from .security import McpAccessMiddleware
from .tools import TOOLS, handle_tool
@@ -68,16 +69,18 @@ async def handle_sse(request: Request):
return Response()
def create_mcp_starlette_app() -> Starlette:
def create_mcp_starlette_app(ctx=None) -> Starlette:
"""Build and return the MCP Starlette sub-application.
Routes:
- ``GET /sse`` - SSE stream for MCP clients
- ``POST /messages/`` - client-to-server message posting
``McpAccessMiddleware`` is applied to enforce configurable IP whitelist
and bearer token access control.
``ctx`` is the application :class:`AppContext`; it is stored so status
tools/resources can report live program state. ``McpAccessMiddleware`` is
applied to enforce configurable IP whitelist and bearer token access control.
"""
set_context(ctx)
app = Starlette(
routes=[
Route("/sse", endpoint=handle_sse),

9
backend/src/module/mcp/tools.py
View File

@@ -11,6 +11,8 @@ from module.models import Bangumi, BangumiUpdate, RSSItem
from module.rss import RSSAnalyser, RSSEngine
from module.searcher import SearchTorrent
from .runtime import get_context
logger = logging.getLogger(__name__)
TOOLS = [
@@ -327,12 +329,11 @@ def _list_rss_feeds() -> list[dict]:
def _get_program_status() -> dict:
from module.api.program import program
ctx = get_context()
return {
"version": VERSION,
"running": program.is_running,
"first_run": program.first_run,
"running": ctx.is_running if ctx is not None else False,
"first_run": ctx.first_run if ctx is not None else True,
}

9
backend/src/module/notification/manager.py
View File

@@ -22,6 +22,15 @@ class NotificationManager:
self.providers: list["NotificationProvider"] = []
self._load_providers()
def rebuild(self):
"""Reload providers from current settings, mutating in place.
Called on config reload so the shared instance (captured by the rename
loop) picks up provider changes without being re-wired.
"""
self.providers = []
self._load_providers()
def _load_providers(self):
"""Initialize providers from configuration."""
from module.notification.providers import PROVIDER_REGISTRY

52
backend/src/module/security/api.py
View File

@@ -1,4 +1,5 @@
from datetime import datetime
import secrets
from datetime import datetime, timedelta
from fastapi import Cookie, Depends, HTTPException, Request, status
from fastapi.security import OAuth2PasswordBearer
@@ -12,7 +13,48 @@ from .jwt import verify_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
active_user: dict[str, datetime] = {}
# Session lifetime mirrors the JWT expiry used at issuance (see auth.py /
# passkey.py, both timedelta(days=1)); the store is a secondary guard so a
# session cannot outlive its token.
SESSION_LIFETIME = timedelta(days=1)
class SessionStore:
"""In-memory registry of active usernames with lazy expiry.
A username counts as present only while its recorded session is younger than
``lifetime``. Expired entries are evicted lazily on access. Replaces the bare
module-global ``dict`` that never expired entries.
"""
def __init__(self, lifetime: timedelta = SESSION_LIFETIME) -> None:
self._lifetime = lifetime
self._sessions: dict[str, datetime] = {}
def add(self, username: str) -> None:
self._sessions[username] = datetime.now()
def remove(self, username: str) -> None:
self._sessions.pop(username, None)
def clear(self) -> None:
"""Invalidate every session (logout-all)."""
self._sessions.clear()
def __contains__(self, username: str) -> bool:
issued = self._sessions.get(username)
if issued is None:
return False
if datetime.now() - issued >= self._lifetime:
self._sessions.pop(username, None)
return False
return True
def __len__(self) -> int:
return len(self._sessions)
active_user = SessionStore()
try:
from module.__version__ import VERSION
@@ -52,7 +94,9 @@ async def get_current_user(request: Request, token: str = Cookie(None)):
auth_header = request.headers.get("authorization", "")
if auth_header.startswith("Bearer "):
api_token = auth_header[7:]
if api_token and api_token in settings.security.login_tokens:
if api_token and any(
secrets.compare_digest(api_token, t) for t in settings.security.login_tokens
):
return "api_token_user"
if not token:
raise UNAUTHORIZED
@@ -88,7 +132,7 @@ def auth_user(user: User):
with Database() as db:
resp = db.user.auth_user(user)
if resp.status:
active_user[user.username] = datetime.now()
active_user.add(user.username)
return resp

32
backend/src/test/test_api_auth.py
View File

@@ -1,6 +1,5 @@
"""Tests for Auth API endpoints."""
from datetime import datetime
from unittest.mock import MagicMock, patch
import pytest
@@ -9,7 +8,7 @@ from fastapi.testclient import TestClient
from module.api import v1
from module.models import ResponseModel
from module.security.api import active_user, get_current_user
from module.security.api import SessionStore, active_user, get_current_user
from module.security.jwt import create_access_token
# ---------------------------------------------------------------------------
@@ -17,6 +16,14 @@ from module.security.jwt import create_access_token
# ---------------------------------------------------------------------------
def _store(*usernames: str) -> SessionStore:
"""Build a SessionStore pre-populated with the given active usernames."""
store = SessionStore()
for username in usernames:
store.add(username)
return store
@pytest.fixture
def app():
"""Create a FastAPI app with v1 routes for testing."""
@@ -118,7 +125,7 @@ class TestRefreshToken:
"""GET /auth/refresh_token returns new token."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
with patch("module.api.auth.active_user", {"testuser": datetime.now()}):
with patch("module.api.auth.active_user", _store("testuser")):
response = authed_client.get("/api/v1/auth/refresh_token")
assert response.status_code == 200
@@ -137,7 +144,7 @@ class TestLogout:
"""GET /auth/logout clears session and returns success."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
with patch("module.api.auth.active_user", {"testuser": datetime.now()}):
with patch("module.api.auth.active_user", _store("testuser")):
response = authed_client.get("/api/v1/auth/logout")
assert response.status_code == 200
@@ -155,7 +162,7 @@ class TestUpdateCredentials:
"""POST /auth/update with valid data updates credentials."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
with patch("module.api.auth.active_user", {"testuser": datetime.now()}):
with patch("module.api.auth.active_user", _store("testuser")):
with patch("module.api.auth.update_user_info", return_value=True):
response = authed_client.post(
"/api/v1/auth/update",
@@ -171,7 +178,7 @@ class TestUpdateCredentials:
"""POST /auth/update with invalid old password fails."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
with patch("module.api.auth.active_user", {"testuser": datetime.now()}):
with patch("module.api.auth.active_user", _store("testuser")):
with patch("module.api.auth.update_user_info", return_value=False):
# When update_user_info returns False, the endpoint implicitly
# returns None which causes an error
@@ -204,7 +211,7 @@ class TestRefreshTokenCookieBehavior:
"""GET /refresh_token updates the active user timestamp."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
active_users: dict = {}
active_users = SessionStore()
with patch("module.api.auth.active_user", active_users):
response = authed_client.get("/api/v1/auth/refresh_token")
assert response.status_code == 200
@@ -214,7 +221,7 @@ class TestRefreshTokenCookieBehavior:
"""GET /refresh_token issues a valid JWT with bearer type."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
with patch("module.api.auth.active_user", {}):
with patch("module.api.auth.active_user", SessionStore()):
response = authed_client.get("/api/v1/auth/refresh_token")
assert response.status_code == 200
data = response.json()
@@ -233,10 +240,7 @@ class TestLogoutCookieBehavior:
"""GET /logout removes the current user from active_user, not others."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
active_users = {
"testuser": datetime.now(),
"otheruser": datetime.now(),
}
active_users = _store("testuser", "otheruser")
with patch("module.api.auth.active_user", active_users):
response = authed_client.get("/api/v1/auth/logout")
assert response.status_code == 200
@@ -246,7 +250,7 @@ class TestLogoutCookieBehavior:
def test_logout_with_no_cookie_still_succeeds(self, authed_client):
"""GET /logout with no cookie clears nothing but returns success."""
with patch("module.api.auth.decode_token", return_value=None):
with patch("module.api.auth.active_user", {}):
with patch("module.api.auth.active_user", SessionStore()):
response = authed_client.get("/api/v1/auth/logout")
assert response.status_code == 200
@@ -270,7 +274,7 @@ class TestUpdateCookieBehavior:
"""POST /auth/update resolves username from cookie and issues new token."""
token = create_access_token(data={"sub": "testuser"})
authed_client.cookies.set("token", token)
with patch("module.api.auth.active_user", {"testuser": datetime.now()}):
with patch("module.api.auth.active_user", _store("testuser")):
with patch("module.api.auth.update_user_info", return_value=True):
response = authed_client.post(
"/api/v1/auth/update",

22
backend/src/test/test_api_config.py
View File

@@ -1,13 +1,14 @@
"""Tests for Config API endpoints and config sanitization."""
import pytest
from unittest.mock import patch, MagicMock
from unittest.mock import patch, AsyncMock, MagicMock
from fastapi import FastAPI
from fastapi.testclient import TestClient
from module.api import v1
from module.api.config import _sanitize_dict, _restore_masked
from module.api.deps import get_context
from module.models.config import Config
from module.security.api import get_current_user
@@ -26,13 +27,22 @@ def app():
@pytest.fixture
def authed_client(app):
"""TestClient with auth dependency overridden."""
def mock_ctx():
"""Mock AppContext whose reload_settings is an awaitable no-op."""
ctx = MagicMock()
ctx.reload_settings = AsyncMock()
return ctx
@pytest.fixture
def authed_client(app, mock_ctx):
"""TestClient with auth and context dependencies overridden."""
async def mock_user():
return "testuser"
app.dependency_overrides[get_current_user] = mock_user
app.dependency_overrides[get_context] = lambda: mock_ctx
client = TestClient(app)
yield client
app.dependency_overrides.clear()
@@ -128,7 +138,7 @@ class TestGetConfig:
class TestUpdateConfig:
def test_update_config_success(self, authed_client, mock_settings):
def test_update_config_success(self, authed_client, mock_settings, mock_ctx):
"""PATCH /config/update updates configuration successfully."""
update_data = {
"program": {
@@ -188,7 +198,9 @@ class TestUpdateConfig:
data = response.json()
assert data["msg_en"] == "Update config successfully."
mock_settings.save.assert_called_once()
mock_settings.load.assert_called_once()
# settings.load() now runs inside ctx.reload_settings(), which the route
# awaits after saving.
mock_ctx.reload_settings.assert_awaited_once()
def test_update_config_failure(self, authed_client, mock_settings):
"""PATCH /config/update handles save failure."""

4
backend/src/test/test_api_passkey.py
View File

@@ -10,7 +10,7 @@ from fastapi.testclient import TestClient
from module.api import v1
from module.models import ResponseModel
from module.models.passkey import Passkey
from module.security.api import get_current_user
from module.security.api import SessionStore, get_current_user
from test.factories import make_passkey
# ---------------------------------------------------------------------------
@@ -340,7 +340,7 @@ class TestAuthVerify:
with patch(
"module.api.passkey.PasskeyAuthStrategy", return_value=mock_strategy
):
with patch("module.api.passkey.active_user", {}):
with patch("module.api.passkey.active_user", SessionStore()):
response = unauthed_client.post(
"/api/v1/passkey/auth/verify",
json={

127
backend/src/test/test_api_program.py
View File

@@ -7,10 +7,10 @@ from fastapi import FastAPI
from fastapi.testclient import TestClient
from module.api import v1
from module.api.deps import get_context
from module.models import ResponseModel
from module.security.api import get_current_user
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@@ -25,13 +25,39 @@ def app():
@pytest.fixture
def authed_client(app):
"""TestClient with auth dependency overridden."""
def mock_ctx():
"""Mock AppContext instance injected via the get_context dependency."""
ctx = MagicMock()
ctx.is_running = True
ctx.first_run = False
ctx.start_tasks = AsyncMock(
return_value=ResponseModel(
status=True, status_code=200, msg_en="Started.", msg_zh="已启动。"
)
)
ctx.stop = AsyncMock(
return_value=ResponseModel(
status=True, status_code=200, msg_en="Stopped.", msg_zh="已停止。"
)
)
ctx.restart = AsyncMock(
return_value=ResponseModel(
status=True, status_code=200, msg_en="Restarted.", msg_zh="已重启。"
)
)
ctx.check_downloader = AsyncMock(return_value=True)
return ctx
@pytest.fixture
def authed_client(app, mock_ctx):
"""TestClient with auth and context dependencies overridden."""
async def mock_user():
return "testuser"
app.dependency_overrides[get_current_user] = mock_user
app.dependency_overrides[get_context] = lambda: mock_ctx
client = TestClient(app)
yield client
app.dependency_overrides.clear()
@@ -43,31 +69,6 @@ def unauthed_client(app):
return TestClient(app)
@pytest.fixture
def mock_program():
"""Mock Program instance."""
program = MagicMock()
program.is_running = True
program.first_run = False
program.start = AsyncMock(
return_value=ResponseModel(
status=True, status_code=200, msg_en="Started.", msg_zh="已启动。"
)
)
program.stop = AsyncMock(
return_value=ResponseModel(
status=True, status_code=200, msg_en="Stopped.", msg_zh="已停止。"
)
)
program.restart = AsyncMock(
return_value=ResponseModel(
status=True, status_code=200, msg_en="Restarted.", msg_zh="已重启。"
)
)
program.check_downloader = AsyncMock(return_value=True)
return program
# ---------------------------------------------------------------------------
# Auth requirement
# ---------------------------------------------------------------------------
@@ -105,19 +106,15 @@ class TestAuthRequired:
class TestStartProgram:
def test_start_success(self, authed_client, mock_program):
def test_start_success(self, authed_client):
"""GET /start returns success response."""
with patch("module.api.program.program", mock_program):
response = authed_client.get("/api/v1/start")
response = authed_client.get("/api/v1/start")
assert response.status_code == 200
def test_start_failure(self, authed_client, mock_program):
def test_start_failure(self, authed_client, mock_ctx):
"""GET /start handles exceptions."""
mock_program.start = AsyncMock(side_effect=Exception("Start failed"))
with patch("module.api.program.program", mock_program):
response = authed_client.get("/api/v1/start")
mock_ctx.start_tasks = AsyncMock(side_effect=Exception("Start failed"))
response = authed_client.get("/api/v1/start")
assert response.status_code == 500
@@ -127,11 +124,9 @@ class TestStartProgram:
class TestStopProgram:
def test_stop_success(self, authed_client, mock_program):
def test_stop_success(self, authed_client):
"""GET /stop returns success response."""
with patch("module.api.program.program", mock_program):
response = authed_client.get("/api/v1/stop")
response = authed_client.get("/api/v1/stop")
assert response.status_code == 200
@@ -141,19 +136,15 @@ class TestStopProgram:
class TestRestartProgram:
def test_restart_success(self, authed_client, mock_program):
def test_restart_success(self, authed_client):
"""GET /restart returns success response."""
with patch("module.api.program.program", mock_program):
response = authed_client.get("/api/v1/restart")
response = authed_client.get("/api/v1/restart")
assert response.status_code == 200
def test_restart_failure(self, authed_client, mock_program):
def test_restart_failure(self, authed_client, mock_ctx):
"""GET /restart handles exceptions."""
mock_program.restart = AsyncMock(side_effect=Exception("Restart failed"))
with patch("module.api.program.program", mock_program):
response = authed_client.get("/api/v1/restart")
mock_ctx.restart = AsyncMock(side_effect=Exception("Restart failed"))
response = authed_client.get("/api/v1/restart")
assert response.status_code == 500
@@ -163,13 +154,12 @@ class TestRestartProgram:
class TestProgramStatus:
def test_status_running(self, authed_client, mock_program):
def test_status_running(self, authed_client, mock_ctx):
"""GET /status returns running status."""
mock_program.is_running = True
mock_program.first_run = False
with patch("module.api.program.program", mock_program):
with patch("module.api.program.VERSION", "3.2.0"):
response = authed_client.get("/api/v1/status")
mock_ctx.is_running = True
mock_ctx.first_run = False
with patch("module.api.program.VERSION", "3.2.0"):
response = authed_client.get("/api/v1/status")
assert response.status_code == 200
data = response.json()
@@ -177,13 +167,12 @@ class TestProgramStatus:
assert data["version"] == "3.2.0"
assert data["first_run"] is False
def test_status_stopped(self, authed_client, mock_program):
def test_status_stopped(self, authed_client, mock_ctx):
"""GET /status returns stopped status."""
mock_program.is_running = False
mock_program.first_run = True
with patch("module.api.program.program", mock_program):
with patch("module.api.program.VERSION", "3.2.0"):
response = authed_client.get("/api/v1/status")
mock_ctx.is_running = False
mock_ctx.first_run = True
with patch("module.api.program.VERSION", "3.2.0"):
response = authed_client.get("/api/v1/status")
assert response.status_code == 200
data = response.json()
@@ -197,20 +186,18 @@ class TestProgramStatus:
class TestCheckDownloader:
def test_check_downloader_connected(self, authed_client, mock_program):
def test_check_downloader_connected(self, authed_client, mock_ctx):
"""GET /check/downloader returns True when connected."""
mock_program.check_downloader = AsyncMock(return_value=True)
with patch("module.api.program.program", mock_program):
response = authed_client.get("/api/v1/check/downloader")
mock_ctx.check_downloader = AsyncMock(return_value=True)
response = authed_client.get("/api/v1/check/downloader")
assert response.status_code == 200
assert response.json() is True
def test_check_downloader_disconnected(self, authed_client, mock_program):
def test_check_downloader_disconnected(self, authed_client, mock_ctx):
"""GET /check/downloader returns False when disconnected."""
mock_program.check_downloader = AsyncMock(return_value=False)
with patch("module.api.program.program", mock_program):
response = authed_client.get("/api/v1/check/downloader")
mock_ctx.check_downloader = AsyncMock(return_value=False)
response = authed_client.get("/api/v1/check/downloader")
assert response.status_code == 200
assert response.json() is False

4
backend/src/test/test_auth.py
View File

@@ -201,15 +201,13 @@ class TestGetCurrentUser:
@patch("module.security.api.DEV_AUTH_BYPASS", False)
async def test_valid_token_active_user_succeeds(self):
"""get_current_user returns username for valid token + active user."""
from datetime import datetime
from module.security.api import active_user, get_current_user
token = create_access_token(
data={"sub": "active_user"}, expires_delta=timedelta(hours=1)
)
active_user.clear()
active_user["active_user"] = datetime.now()
active_user.add("active_user")
result = await get_current_user(request=self._mock_request(), token=token)
assert result == "active_user"

198
backend/src/test/test_context.py Normal file
View File

@@ -0,0 +1,198 @@
"""Tests for module.core.context.AppContext."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from module.conf import settings as real_settings
from module.core.context import AppContext
from module.models import ResponseModel
@pytest.fixture
def ctx():
"""A built AppContext whose notifier/scheduler are replaced with mocks."""
ctx = AppContext.build(real_settings)
ctx.notifier = MagicMock()
ctx.scheduler = MagicMock()
ctx.scheduler.stop_all = AsyncMock()
ctx.scheduler.start_all = MagicMock()
return ctx
# ---------------------------------------------------------------------------
# build()
# ---------------------------------------------------------------------------
class TestBuild:
def test_build_registers_four_named_tasks(self):
built = AppContext.build(real_settings)
names = [t.name for t in built.scheduler.tasks]
assert names == ["rss", "rename", "offset_scan", "calendar"]
def test_build_does_no_io(self):
"""build() is pure wiring: not running, startup not done."""
built = AppContext.build(real_settings)
assert built.is_running is False
assert built._startup_done is False
assert built.first_run_boot is False
# ---------------------------------------------------------------------------
# startup()
# ---------------------------------------------------------------------------
class TestStartup:
async def test_first_run_creates_db_and_flags_boot(self, ctx):
"""No database => first_run() runs, boot is flagged, tasks not started."""
with (
patch("module.core.context.Checker.check_database", return_value=False),
patch("module.core.context.first_run") as mock_first_run,
):
await ctx.startup()
mock_first_run.assert_called_once()
assert ctx.first_run_boot is True
assert ctx._startup_done is True
assert ctx._tasks_started is False
async def test_existing_db_runs_pending_migrations(self, ctx):
"""Existing DB, same version => run_migrations() is invoked."""
with (
patch("module.core.context.Checker.check_database", return_value=True),
patch("module.core.context.LEGACY_DATA_PATH") as legacy,
patch(
"module.core.context.Checker.check_version",
return_value=(True, None),
),
patch("module.core.context.Checker.check_img_cache", return_value=True),
patch("module.core.context.run_migrations") as mock_run,
):
legacy.exists.return_value = False
await ctx.startup()
mock_run.assert_called_once()
assert ctx.first_run_boot is False
assert ctx._startup_done is True
async def test_version_bump_runs_cross_version_migrations(self, ctx):
"""A version change from minor 0 triggers 3.0->3.1 then 3.1->3.2."""
with (
patch("module.core.context.Checker.check_database", return_value=True),
patch("module.core.context.LEGACY_DATA_PATH") as legacy,
patch(
"module.core.context.Checker.check_version",
return_value=(False, 0),
),
patch("module.core.context.Checker.check_img_cache", return_value=True),
patch("module.core.context.from_30_to_31", new=AsyncMock()) as m_30_31,
patch("module.core.context.from_31_to_32", new=AsyncMock()) as m_31_32,
):
legacy.exists.return_value = False
await ctx.startup()
m_30_31.assert_awaited_once()
m_31_32.assert_awaited_once()
async def test_startup_is_idempotent(self, ctx):
"""A second startup() call returns immediately without re-running."""
ctx._startup_done = True
with patch("module.core.context.Checker.check_database") as check_db:
await ctx.startup()
check_db.assert_not_called()
# ---------------------------------------------------------------------------
# start_tasks()
# ---------------------------------------------------------------------------
class TestStartTasks:
async def test_start_tasks_waits_then_starts_scheduler(self, ctx):
with (
patch("module.core.context.settings"),
patch(
"module.core.context.Checker.check_downloader",
new=AsyncMock(return_value=True),
),
):
resp = await ctx.start_tasks()
ctx.scheduler.start_all.assert_called_once()
assert ctx._tasks_started is True
assert isinstance(resp, ResponseModel)
assert resp.status is True
# ---------------------------------------------------------------------------
# stop()
# ---------------------------------------------------------------------------
class TestStop:
async def test_stop_running_stops_scheduler_and_downloader(self, ctx):
ctx._tasks_started = True
with (
patch("module.core.context.Checker.check_first_run", return_value=False),
patch(
"module.core.context.downloader_shutdown", new=AsyncMock()
) as mock_shutdown,
):
resp = await ctx.stop()
ctx.scheduler.stop_all.assert_awaited_once()
mock_shutdown.assert_awaited_once()
assert ctx._tasks_started is False
assert resp.status_code == 200
async def test_stop_not_running_still_closes_downloader(self, ctx):
ctx._tasks_started = False
with (
patch("module.core.context.Checker.check_first_run", return_value=False),
patch(
"module.core.context.downloader_shutdown", new=AsyncMock()
) as mock_shutdown,
):
resp = await ctx.stop()
ctx.scheduler.stop_all.assert_not_awaited()
mock_shutdown.assert_awaited_once()
assert resp.status_code == 406
# ---------------------------------------------------------------------------
# reload_settings()
# ---------------------------------------------------------------------------
class TestReloadSettings:
async def test_reload_runs_full_choreography_when_running(self, ctx):
ctx.scheduler.running = True
with (
patch("module.core.context.settings") as mock_settings,
patch(
"module.core.context.reset_shared_client", new=AsyncMock()
) as mock_reset,
):
await ctx.reload_settings()
mock_settings.load.assert_called_once()
mock_reset.assert_awaited_once()
ctx.notifier.rebuild.assert_called_once()
ctx.scheduler.stop_all.assert_awaited_once()
ctx.scheduler.start_all.assert_called_once()
async def test_reload_skips_scheduler_restart_when_idle(self, ctx):
ctx.scheduler.running = False
with (
patch("module.core.context.settings") as mock_settings,
patch("module.core.context.reset_shared_client", new=AsyncMock()),
):
await ctx.reload_settings()
mock_settings.load.assert_called_once()
ctx.notifier.rebuild.assert_called_once()
ctx.scheduler.stop_all.assert_not_awaited()
ctx.scheduler.start_all.assert_not_called()

60
backend/src/test/test_main_lifespan.py Normal file
View File

@@ -0,0 +1,60 @@
"""Tests for the application lifespan wiring in main.create_app."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi.testclient import TestClient
import main
from module.core import AppContext
def _lifespan_ctx(first_run_boot: bool = False) -> MagicMock:
ctx = MagicMock()
ctx.startup = AsyncMock()
ctx.start_tasks = AsyncMock(return_value=None)
ctx.stop = AsyncMock()
ctx.first_run_boot = first_run_boot
return ctx
class TestCreateApp:
def test_stores_real_context_on_state(self):
app = main.create_app()
assert isinstance(app.state.ctx, AppContext)
names = [t.name for t in app.state.ctx.scheduler.tasks]
assert names == ["rss", "rename", "offset_scan", "calendar"]
class TestLifespan:
def test_startup_runs_and_tasks_scheduled(self):
"""Lifespan awaits startup, schedules start_tasks, stops on shutdown."""
ctx = _lifespan_ctx(first_run_boot=False)
with patch("main.AppContext.build", return_value=ctx):
app = main.create_app()
with TestClient(app):
pass
ctx.startup.assert_awaited_once()
ctx.start_tasks.assert_called_once()
ctx.stop.assert_awaited_once()
def test_first_run_boot_skips_task_start(self):
"""On a first-run boot the background tasks are not auto-started."""
ctx = _lifespan_ctx(first_run_boot=True)
with patch("main.AppContext.build", return_value=ctx):
app = main.create_app()
with TestClient(app):
pass
ctx.startup.assert_awaited_once()
ctx.start_tasks.assert_not_called()
ctx.stop.assert_awaited_once()
def test_startup_failure_aborts_boot(self):
"""A failing startup propagates out of the lifespan (boot aborts)."""
ctx = _lifespan_ctx(first_run_boot=False)
ctx.startup = AsyncMock(side_effect=RuntimeError("startup boom"))
with patch("main.AppContext.build", return_value=ctx):
app = main.create_app()
with pytest.raises(RuntimeError, match="startup boom"):
with TestClient(app):
pass

10
backend/src/test/test_mcp_resources.py
View File

@@ -204,7 +204,7 @@ class TestHandleResourceStatus:
def test_returns_json_string(self, mock_program):
with (
patch("module.mcp.resources.VERSION", "3.2.0-test"),
patch("module.api.program.program", mock_program),
patch("module.mcp.resources.get_context", return_value=mock_program),
):
raw = handle_resource("autobangumi://status")
assert isinstance(raw, str)
@@ -213,7 +213,7 @@ class TestHandleResourceStatus:
def test_version_in_output(self, mock_program):
with (
patch("module.mcp.resources.VERSION", "3.2.0-test"),
patch("module.api.program.program", mock_program),
patch("module.mcp.resources.get_context", return_value=mock_program),
):
result = _parse(handle_resource("autobangumi://status"))
assert result["version"] == "3.2.0-test"
@@ -222,7 +222,7 @@ class TestHandleResourceStatus:
mock_program.is_running = True
with (
patch("module.mcp.resources.VERSION", "3.2.0-test"),
patch("module.api.program.program", mock_program),
patch("module.mcp.resources.get_context", return_value=mock_program),
):
result = _parse(handle_resource("autobangumi://status"))
assert result["running"] is True
@@ -231,7 +231,7 @@ class TestHandleResourceStatus:
mock_program.first_run = False
with (
patch("module.mcp.resources.VERSION", "3.2.0-test"),
patch("module.api.program.program", mock_program),
patch("module.mcp.resources.get_context", return_value=mock_program),
):
result = _parse(handle_resource("autobangumi://status"))
assert result["first_run"] is False
@@ -239,7 +239,7 @@ class TestHandleResourceStatus:
def test_all_keys_present(self, mock_program):
with (
patch("module.mcp.resources.VERSION", "3.2.0-test"),
patch("module.api.program.program", mock_program),
patch("module.mcp.resources.get_context", return_value=mock_program),
):
result = _parse(handle_resource("autobangumi://status"))
assert {"version", "running", "first_run"}.issubset(result.keys())

41
backend/src/test/test_mcp_runtime.py Normal file
View File

@@ -0,0 +1,41 @@
"""Tests for MCP runtime context wiring (module.mcp.runtime)."""
from unittest.mock import MagicMock
from module.mcp import create_mcp_app
from module.mcp.runtime import get_context, set_context
from module.mcp.tools import _get_program_status
class TestRuntimeContext:
def test_create_mcp_app_stores_context(self):
sentinel = object()
try:
create_mcp_app(sentinel)
assert get_context() is sentinel
finally:
set_context(None)
def test_get_context_defaults_to_none(self):
set_context(None)
assert get_context() is None
class TestProgramStatusReadsContext:
def test_status_reflects_context(self):
ctx = MagicMock()
ctx.is_running = True
ctx.first_run = False
set_context(ctx)
try:
result = _get_program_status()
finally:
set_context(None)
assert result["running"] is True
assert result["first_run"] is False
def test_status_without_context_is_safe(self):
set_context(None)
result = _get_program_status()
assert result["running"] is False
assert result["first_run"] is True

167
backend/src/test/test_scheduler.py Normal file
View File

@@ -0,0 +1,167 @@
"""Unit tests for module.core.scheduler (PeriodicTask, Scheduler)."""
import asyncio
from unittest.mock import MagicMock
from module.core.scheduler import PeriodicTask, Scheduler
# ---------------------------------------------------------------------------
# PeriodicTask
# ---------------------------------------------------------------------------
class TestPeriodicTask:
async def test_start_runs_body_then_stop(self):
"""start() ticks the body repeatedly; stop() halts it."""
counter = {"n": 0}
async def body():
counter["n"] += 1
task = PeriodicTask("t", run=body, interval=lambda: 0.01)
task.start()
assert task.running is True
await asyncio.sleep(0.05)
await task.stop()
assert task.running is False
assert counter["n"] >= 2
async def test_stop_is_idempotent(self):
"""Calling stop() twice (or before start) does not raise."""
task = PeriodicTask("t", run=lambda: asyncio.sleep(0), interval=lambda: 0.01)
await task.stop() # never started
task.start()
await task.stop()
await task.stop() # already stopped
assert task.running is False
async def test_exception_in_body_does_not_kill_loop(self):
"""A raising tick is logged and the loop keeps running."""
counter = {"n": 0}
async def body():
counter["n"] += 1
raise RuntimeError("boom")
task = PeriodicTask("t", run=body, interval=lambda: 0.01)
task.start()
await asyncio.sleep(0.05)
assert task.running is True # still alive despite exceptions
await task.stop()
assert counter["n"] >= 2
async def test_interval_read_live_each_tick(self):
"""interval() is invoked on every wait, so settings are read live."""
interval = MagicMock(return_value=0.01)
async def body():
pass
task = PeriodicTask("t", run=body, interval=interval)
task.start()
await asyncio.sleep(0.05)
await task.stop()
assert interval.call_count >= 2
async def test_initial_delay_defers_first_tick(self):
"""With initial_delay set, the body does not run before the delay."""
counter = {"n": 0}
async def body():
counter["n"] += 1
task = PeriodicTask("t", run=body, interval=lambda: 0.01, initial_delay=10.0)
task.start()
await asyncio.sleep(0.03)
# Still inside the initial delay window -> body has not run yet.
assert counter["n"] == 0
await task.stop() # stop cleanly during the initial delay
assert counter["n"] == 0
async def test_initial_delay_then_runs(self):
"""After a short initial delay the body starts ticking."""
counter = {"n": 0}
async def body():
counter["n"] += 1
task = PeriodicTask("t", run=body, interval=lambda: 0.01, initial_delay=0.01)
task.start()
await asyncio.sleep(0.05)
await task.stop()
assert counter["n"] >= 1
async def test_start_is_idempotent_no_double_task(self):
"""start() while already running does not spawn a second task."""
async def body():
await asyncio.sleep(0)
task = PeriodicTask("t", run=body, interval=lambda: 0.01)
task.start()
first = task._task
task.start()
assert task._task is first
await task.stop()
async def test_enabled_reflects_callable(self):
"""enabled property proxies the provided callable."""
flag = {"on": False}
task = PeriodicTask(
"t",
run=lambda: asyncio.sleep(0),
interval=lambda: 0.01,
enabled=lambda: flag["on"],
)
assert task.enabled is False
flag["on"] = True
assert task.enabled is True
# ---------------------------------------------------------------------------
# Scheduler
# ---------------------------------------------------------------------------
class TestScheduler:
async def test_start_all_only_enabled(self):
"""start_all starts enabled tasks and skips disabled ones."""
async def body():
await asyncio.sleep(0)
enabled = PeriodicTask("on", run=body, interval=lambda: 0.01)
disabled = PeriodicTask(
"off", run=body, interval=lambda: 0.01, enabled=lambda: False
)
scheduler = Scheduler([enabled, disabled])
scheduler.start_all()
assert scheduler.running is True
assert enabled.running is True
assert disabled.running is False
await scheduler.stop_all()
async def test_stop_all_halts_and_clears_running(self):
"""stop_all stops every task and flips running to False."""
async def body():
await asyncio.sleep(0)
t1 = PeriodicTask("a", run=body, interval=lambda: 0.01)
t2 = PeriodicTask("b", run=body, interval=lambda: 0.01)
scheduler = Scheduler([t1, t2])
scheduler.start_all()
await asyncio.sleep(0.02)
await scheduler.stop_all()
assert scheduler.running is False
assert t1.running is False
assert t2.running is False
async def test_running_false_before_start(self):
scheduler = Scheduler([])
assert scheduler.running is False

100
backend/src/test/test_session_store.py Normal file
View File

@@ -0,0 +1,100 @@
"""Tests for SessionStore expiry and the constant-time bearer-token path."""
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
import pytest
from fastapi import HTTPException
from module.conf import settings
from module.security.api import SessionStore, get_current_user
# ---------------------------------------------------------------------------
# SessionStore
# ---------------------------------------------------------------------------
class TestSessionStore:
def test_add_then_present(self):
store = SessionStore()
store.add("alice")
assert "alice" in store
def test_absent_user_not_present(self):
store = SessionStore()
assert "nobody" not in store
def test_remove(self):
store = SessionStore()
store.add("alice")
store.remove("alice")
assert "alice" not in store
def test_remove_missing_is_noop(self):
store = SessionStore()
store.remove("ghost") # must not raise
assert len(store) == 0
def test_clear(self):
store = SessionStore()
store.add("a")
store.add("b")
store.clear()
assert len(store) == 0
def test_expired_entry_is_evicted_on_access(self):
"""An entry older than the lifetime is treated as absent and dropped."""
store = SessionStore(lifetime=timedelta(hours=1))
store.add("alice")
# Backdate the recorded session beyond the lifetime.
store._sessions["alice"] = datetime.now() - timedelta(hours=2)
assert "alice" not in store
# Lazily evicted, so the backing dict no longer holds it.
assert len(store) == 0
def test_fresh_entry_within_lifetime_is_present(self):
store = SessionStore(lifetime=timedelta(hours=1))
store.add("alice")
store._sessions["alice"] = datetime.now() - timedelta(minutes=30)
assert "alice" in store
# ---------------------------------------------------------------------------
# Bearer-token path uses constant-time comparison
# ---------------------------------------------------------------------------
def _mock_request(authorization: str = "") -> MagicMock:
request = MagicMock()
request.headers = {"authorization": authorization}
return request
class TestBearerTokenCompareDigest:
@patch("module.security.api.DEV_AUTH_BYPASS", False)
async def test_matching_token_authenticates(self):
"""A configured bearer token authenticates via compare_digest."""
with patch.object(settings.security, "login_tokens", ["secret-token"]):
result = await get_current_user(
request=_mock_request("Bearer secret-token"), token=None
)
assert result == "api_token_user"
@patch("module.security.api.DEV_AUTH_BYPASS", False)
async def test_wrong_token_rejected(self):
"""A non-matching bearer token falls through to 401 (no cookie)."""
with patch.object(settings.security, "login_tokens", ["secret-token"]):
with pytest.raises(HTTPException) as exc_info:
await get_current_user(
request=_mock_request("Bearer wrong-token"), token=None
)
assert exc_info.value.status_code == 401
@patch("module.security.api.DEV_AUTH_BYPASS", False)
async def test_no_tokens_configured_rejected(self):
"""With no configured tokens, a bearer header cannot authenticate."""
with patch.object(settings.security, "login_tokens", []):
with pytest.raises(HTTPException):
await get_current_user(
request=_mock_request("Bearer anything"), token=None
)