diff --git a/backend/src/main.py b/backend/src/main.py index 25d33ed2..c4212a4c 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -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 diff --git a/backend/src/module/api/auth.py b/backend/src/module/api/auth.py index 6b362070..1eb395c4 100644 --- a/backend/src/module/api/auth.py +++ b/backend/src/module/api/auth.py @@ -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, diff --git a/backend/src/module/api/config.py b/backend/src/module/api/config.py index e9740e02..49dd1645 100644 --- a/backend/src/module/api/config.py +++ b/backend/src/module/api/config.py @@ -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, diff --git a/backend/src/module/api/deps.py b/backend/src/module/api/deps.py new file mode 100644 index 00000000..188fd72b --- /dev/null +++ b/backend/src/module/api/deps.py @@ -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 diff --git a/backend/src/module/api/passkey.py b/backend/src/module/api/passkey.py index 25a63d5e..b91fedf2 100644 --- a/backend/src/module/api/passkey.py +++ b/backend/src/module/api/passkey.py @@ -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) diff --git a/backend/src/module/api/program.py b/backend/src/module/api/program.py index e3cb07b9..1d174f33 100644 --- a/backend/src/module/api/program.py +++ b/backend/src/module/api/program.py @@ -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() diff --git a/backend/src/module/core/__init__.py b/backend/src/module/core/__init__.py index 84633ad6..4b96469e 100644 --- a/backend/src/module/core/__init__.py +++ b/backend/src/module/core/__init__.py @@ -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"] diff --git a/backend/src/module/core/context.py b/backend/src/module/core/context.py new file mode 100644 index 00000000..939bba26 --- /dev/null +++ b/backend/src/module/core/context.py @@ -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() diff --git a/backend/src/module/core/loops.py b/backend/src/module/core/loops.py new file mode 100644 index 00000000..4ce99891 --- /dev/null +++ b/backend/src/module/core/loops.py @@ -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 + ) diff --git a/backend/src/module/core/program.py b/backend/src/module/core/program.py deleted file mode 100644 index 64b1acf5..00000000 --- a/backend/src/module/core/program.py +++ /dev/null @@ -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."} diff --git a/backend/src/module/core/scheduler.py b/backend/src/module/core/scheduler.py new file mode 100644 index 00000000..9a3d3295 --- /dev/null +++ b/backend/src/module/core/scheduler.py @@ -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 diff --git a/backend/src/module/core/status.py b/backend/src/module/core/status.py deleted file mode 100644 index 7df2a2d9..00000000 --- a/backend/src/module/core/status.py +++ /dev/null @@ -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() diff --git a/backend/src/module/core/sub_thread.py b/backend/src/module/core/sub_thread.py deleted file mode 100644 index 2972fb65..00000000 --- a/backend/src/module/core/sub_thread.py +++ /dev/null @@ -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") diff --git a/backend/src/module/mcp/__init__.py b/backend/src/module/mcp/__init__.py index b3bacb82..cdd0a305 100644 --- a/backend/src/module/mcp/__init__.py +++ b/backend/src/module/mcp/__init__.py @@ -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 diff --git a/backend/src/module/mcp/resources.py b/backend/src/module/mcp/resources.py index a0cdd4d4..2c28769d 100644 --- a/backend/src/module/mcp/resources.py +++ b/backend/src/module/mcp/resources.py @@ -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, } ) diff --git a/backend/src/module/mcp/runtime.py b/backend/src/module/mcp/runtime.py new file mode 100644 index 00000000..2f429d30 --- /dev/null +++ b/backend/src/module/mcp/runtime.py @@ -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 diff --git a/backend/src/module/mcp/server.py b/backend/src/module/mcp/server.py index 0bdc85d3..137184ef 100644 --- a/backend/src/module/mcp/server.py +++ b/backend/src/module/mcp/server.py @@ -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), diff --git a/backend/src/module/mcp/tools.py b/backend/src/module/mcp/tools.py index ef988ff1..d9618ef6 100644 --- a/backend/src/module/mcp/tools.py +++ b/backend/src/module/mcp/tools.py @@ -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, } diff --git a/backend/src/module/notification/manager.py b/backend/src/module/notification/manager.py index 378c875d..75a063fb 100644 --- a/backend/src/module/notification/manager.py +++ b/backend/src/module/notification/manager.py @@ -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 diff --git a/backend/src/module/security/api.py b/backend/src/module/security/api.py index 23b2b154..127a98a7 100644 --- a/backend/src/module/security/api.py +++ b/backend/src/module/security/api.py @@ -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 diff --git a/backend/src/test/test_api_auth.py b/backend/src/test/test_api_auth.py index d03d01d9..9956707d 100644 --- a/backend/src/test/test_api_auth.py +++ b/backend/src/test/test_api_auth.py @@ -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", diff --git a/backend/src/test/test_api_config.py b/backend/src/test/test_api_config.py index 8cde2c29..0e9400ef 100644 --- a/backend/src/test/test_api_config.py +++ b/backend/src/test/test_api_config.py @@ -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.""" diff --git a/backend/src/test/test_api_passkey.py b/backend/src/test/test_api_passkey.py index 43f56905..ad034ddf 100644 --- a/backend/src/test/test_api_passkey.py +++ b/backend/src/test/test_api_passkey.py @@ -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={ diff --git a/backend/src/test/test_api_program.py b/backend/src/test/test_api_program.py index 338df926..3da7b71d 100644 --- a/backend/src/test/test_api_program.py +++ b/backend/src/test/test_api_program.py @@ -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 diff --git a/backend/src/test/test_auth.py b/backend/src/test/test_auth.py index aab56cca..b944435d 100644 --- a/backend/src/test/test_auth.py +++ b/backend/src/test/test_auth.py @@ -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" diff --git a/backend/src/test/test_context.py b/backend/src/test/test_context.py new file mode 100644 index 00000000..5755e2e9 --- /dev/null +++ b/backend/src/test/test_context.py @@ -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() diff --git a/backend/src/test/test_main_lifespan.py b/backend/src/test/test_main_lifespan.py new file mode 100644 index 00000000..b2688df4 --- /dev/null +++ b/backend/src/test/test_main_lifespan.py @@ -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 diff --git a/backend/src/test/test_mcp_resources.py b/backend/src/test/test_mcp_resources.py index e2fe8529..4748ff78 100644 --- a/backend/src/test/test_mcp_resources.py +++ b/backend/src/test/test_mcp_resources.py @@ -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()) diff --git a/backend/src/test/test_mcp_runtime.py b/backend/src/test/test_mcp_runtime.py new file mode 100644 index 00000000..badeca50 --- /dev/null +++ b/backend/src/test/test_mcp_runtime.py @@ -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 diff --git a/backend/src/test/test_scheduler.py b/backend/src/test/test_scheduler.py new file mode 100644 index 00000000..82f0c1a5 --- /dev/null +++ b/backend/src/test/test_scheduler.py @@ -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 diff --git a/backend/src/test/test_session_store.py b/backend/src/test/test_session_store.py new file mode 100644 index 00000000..3599099f --- /dev/null +++ b/backend/src/test/test_session_store.py @@ -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 + )