diff --git a/backend/src/module/api/__init__.py b/backend/src/module/api/__init__.py index cd68605a..7ef342da 100644 --- a/backend/src/module/api/__init__.py +++ b/backend/src/module/api/__init__.py @@ -4,6 +4,7 @@ from .auth import router as auth_router from .bangumi import router as bangumi_router from .config import router as config_router from .downloader import router as downloader_router +from .events import router as events_router from .log import router as log_router from .notification import router as notification_router from .passkey import router as passkey_router @@ -23,6 +24,7 @@ v1.include_router(program_router) v1.include_router(bangumi_router) v1.include_router(config_router) v1.include_router(downloader_router) +v1.include_router(events_router) v1.include_router(rss_router) v1.include_router(search_router) v1.include_router(setup_router) diff --git a/backend/src/module/api/events.py b/backend/src/module/api/events.py new file mode 100644 index 00000000..3f8963aa --- /dev/null +++ b/backend/src/module/api/events.py @@ -0,0 +1,87 @@ +"""SSE 端点：合并推送 status/downloader/log 更新，取代前端的三个轮询循环 +(hooks/useAppInfo.ts:24、pages/index/downloader.vue:28、store/log.ts:25)。 + +单个连接按 tick 节拍分别推送三种事件，节拍与原轮询间隔保持一致， +前端订阅失败时回退到原有轮询逻辑。 +""" + +import asyncio +import json +import logging + +from fastapi import APIRouter, Depends, Request +from sse_starlette.sse import EventSourceResponse + +from module.api.deps import get_context +from module.api.log import _read_log_tail +from module.conf import LOG_PATH, VERSION +from module.core import AppContext +from module.downloader import DownloadClient +from module.security.api import get_current_user + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/events", tags=["events"]) + +# tick 节拍（秒）及各事件相对节拍的推送倍数，与原轮询间隔一一对应： +# status 3s、downloader 5s、log 10s。 +_TICK_SECONDS = 1 +_STATUS_EVERY = 3 +_DOWNLOADER_EVERY = 5 +_LOG_EVERY = 10 + + +def _status_payload(ctx: AppContext) -> dict: + """构造与 GET /program/status 相同结构的状态负载。""" + return { + "status": ctx.is_running, + "version": VERSION, + "first_run": ctx.first_run, + } + + +async def _downloader_payload() -> list[dict] | None: + """获取种子列表；下载器未配置或不可达时返回 None（跳过本次推送）。""" + try: + async with DownloadClient() as client: + return await client.get_torrent_info(category="Bangumi", status_filter=None) + except Exception: + logger.debug("SSE: downloader status unavailable", exc_info=True) + return None + + +async def _log_payload() -> str | None: + """读取日志尾部；日志文件不存在时返回 None（跳过本次推送）。""" + if not LOG_PATH.exists(): + return None + data = await asyncio.to_thread(_read_log_tail) + return data.decode("utf-8", errors="replace") + + +@router.get("/stream", dependencies=[Depends(get_current_user)]) +async def event_stream(request: Request, ctx: AppContext = Depends(get_context)): + """单一 SSE 连接，按各自节拍推送 status/downloader/log 事件。""" + + async def generator(): + tick = 0 + while True: + if await request.is_disconnected(): + break + + if tick % _STATUS_EVERY == 0: + yield {"event": "status", "data": json.dumps(_status_payload(ctx))} + + if tick % _DOWNLOADER_EVERY == 0: + torrents = await _downloader_payload() + if torrents is not None: + yield {"event": "downloader", "data": json.dumps(torrents)} + + if tick % _LOG_EVERY == 0: + log_text = await _log_payload() + if log_text is not None: + yield {"event": "log", "data": log_text} + + await asyncio.sleep(_TICK_SECONDS) + tick += _TICK_SECONDS + + return EventSourceResponse(generator()) diff --git a/backend/src/test/test_api_events.py b/backend/src/test/test_api_events.py new file mode 100644 index 00000000..9c9cf0ba --- /dev/null +++ b/backend/src/test/test_api_events.py @@ -0,0 +1,129 @@ +"""Tests for the events SSE endpoint and its payload builders.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from module.api import v1 +from module.api.events import _downloader_payload, _log_payload, _status_payload + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def app(): + """Create a FastAPI app with v1 routes for testing.""" + app = FastAPI() + app.include_router(v1, prefix="/api") + return app + + +@pytest.fixture +def unauthed_client(app): + """TestClient without auth (no override).""" + return TestClient(app) + + +# --------------------------------------------------------------------------- +# Auth requirement — safe to hit synchronously: the dependency rejects the +# request with 401 before the (otherwise-infinite) generator ever starts. +# --------------------------------------------------------------------------- + + +class TestAuthRequired: + @patch("module.security.api.DEV_AUTH_BYPASS", False) + def test_stream_unauthorized(self, unauthed_client): + """GET /events/stream without auth returns 401.""" + response = unauthed_client.get("/api/v1/events/stream") + assert response.status_code == 401 + + +# --------------------------------------------------------------------------- +# Payload builders — exercised directly (not through the infinite generator) +# --------------------------------------------------------------------------- + + +class TestStatusPayload: + def test_status_payload_running(self): + """_status_payload mirrors GET /program/status's running shape.""" + ctx = MagicMock() + ctx.is_running = True + ctx.first_run = False + + payload = _status_payload(ctx) + + assert payload["status"] is True + assert payload["first_run"] is False + assert "version" in payload + + def test_status_payload_stopped(self): + """_status_payload reflects a stopped program.""" + ctx = MagicMock() + ctx.is_running = False + ctx.first_run = True + + payload = _status_payload(ctx) + + assert payload["status"] is False + assert payload["first_run"] is True + + +class TestDownloaderPayload: + async def test_downloader_payload_success(self): + """_downloader_payload returns the torrent list on success.""" + torrents = [{"hash": "abc", "name": "Episode 1"}] + mock_client = AsyncMock() + mock_client.get_torrent_info = AsyncMock(return_value=torrents) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("module.api.events.DownloadClient", return_value=mock_client): + result = await _downloader_payload() + + assert result == torrents + mock_client.get_torrent_info.assert_called_once_with( + category="Bangumi", status_filter=None + ) + + async def test_downloader_payload_unavailable_returns_none(self): + """_downloader_payload swallows client errors and returns None so the + stream skips that tick instead of crashing the connection.""" + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(side_effect=RuntimeError("no downloader")) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("module.api.events.DownloadClient", return_value=mock_client): + result = await _downloader_payload() + + assert result is None + + +class TestLogPayload: + async def test_log_payload_missing_file_returns_none(self, tmp_path): + """_log_payload returns None when the log file doesn't exist yet.""" + missing = tmp_path / "does-not-exist.log" + with patch("module.api.events.LOG_PATH", missing): + result = await _log_payload() + + assert result is None + + async def test_log_payload_reads_tail(self, tmp_path): + """_log_payload decodes the tail bytes read via _read_log_tail. + + _read_log_tail (imported from module.api.log) resolves LOG_PATH via + its own module binding, so both names must be patched together. + """ + log_file = tmp_path / "app.log" + log_file.write_text("hello world\n") + + with ( + patch("module.api.events.LOG_PATH", log_file), + patch("module.api.log.LOG_PATH", log_file), + ): + result = await _log_payload() + + assert result == "hello world\n" diff --git a/webui/src/App.vue b/webui/src/App.vue index 91dd422e..5fe5fa42 100644 --- a/webui/src/App.vue +++ b/webui/src/App.vue @@ -1,6 +1,11 @@