From 4e47cce5f2cc8bbf30a98af123a811a2fb56177e Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sun, 16 Aug 2026 19:48:22 +0800 Subject: [PATCH] fix(runtime): prevent implicit scheduler startup (#6337) --- app/scheduler.py | 8 ++++---- app/startup/scheduler_initializer.py | 2 +- pytest.ini | 1 + tests/test_agent_task_runs.py | 13 ++++--------- tests/test_lifecycle_shutdown.py | 1 + tests/test_media_response_models.py | 29 +++++++++++++++++++++------- tests/test_music_subscribe.py | 10 +++++----- tests/test_scheduler_cache_expiry.py | 24 +++++++++++++++++++++++ 8 files changed, 62 insertions(+), 26 deletions(-) diff --git a/app/scheduler.py b/app/scheduler.py index 25984ad6c..21123c7d8 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -300,6 +300,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): } def __init__(self): + """创建调度器状态；后台任务由应用生命周期显式启动。""" # 定时服务 self._scheduler = None # 退出事件 @@ -314,10 +315,6 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): self._auth_count = 0 # 用户认证失败消息发送 self._auth_message = False - # 对账上个进程未收口的 Agent 任务 - self._reconcile_agent_task_interruptions() - # 初始化 - self.init() def on_config_changed(self) -> None: """ @@ -409,6 +406,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): if settings.DEV: return + # 对账上个进程未收口的 Agent 任务；进程内重复初始化不会重复改写状态。 + self._reconcile_agent_task_interruptions() + with lock: # 各服务的运行状态 mediaserver_chain = MediaServerChain() diff --git a/app/startup/scheduler_initializer.py b/app/startup/scheduler_initializer.py index 28d958792..038e45578 100644 --- a/app/startup/scheduler_initializer.py +++ b/app/startup/scheduler_initializer.py @@ -9,7 +9,7 @@ def init_scheduler(): """ 初始化定时器 """ - Scheduler() + Scheduler().init() def stop_scheduler(): diff --git a/pytest.ini b/pytest.ini index 33cfa3ed1..3c60316d6 100644 --- a/pytest.ini +++ b/pytest.ini @@ -11,3 +11,4 @@ filterwarnings = ignore:datetime.datetime.utcfromtimestamp\(\) is deprecated:DeprecationWarning ignore:'crypt' is deprecated:DeprecationWarning ignore:'audioop' is deprecated:DeprecationWarning + ignore:There is no current event loop:DeprecationWarning:lark_oapi\.ws\.client diff --git a/tests/test_agent_task_runs.py b/tests/test_agent_task_runs.py index 52bb63483..5d328a00b 100644 --- a/tests/test_agent_task_runs.py +++ b/tests/test_agent_task_runs.py @@ -313,15 +313,10 @@ async def test_query_task_returns_owner_scoped_ten_recent_runs(monkeypatch) -> N assert other_run assert oper.finish_run(other_run.run_id, success=True, result="其他用户") - class _SchedulerStub: - """为查询工具提供下一次触发时间，避免启动真实 APScheduler。""" - - @staticmethod - def get_agent_task_next_run(_task_id): - """返回测试任务不需要的下一次触发时间。""" - return None - - monkeypatch.setattr("app.scheduler.Scheduler", _SchedulerStub) + monkeypatch.setattr( + "app.application.scheduling.get_agent_task_next_run", + lambda _task_id: None, + ) detail = json.loads(await _build_query_tool(task.user_id).run(task_id=task.id)) assert detail["total"] == 1 assert [run["run_id"] for run in detail["tasks"][0]["recent_runs"]] == expected[:10] diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 9e92576df..0c1cf8568 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -28,6 +28,7 @@ def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict: "init_plugins", "init_scheduler", "init_monitor", + "replay_pending_transfers", "init_command", "init_workflow", ): diff --git a/tests/test_media_response_models.py b/tests/test_media_response_models.py index ff698076a..c00a068a9 100644 --- a/tests/test_media_response_models.py +++ b/tests/test_media_response_models.py @@ -1,6 +1,6 @@ import pytest from fastapi import FastAPI -from fastapi.testclient import TestClient +from httpx import ASGITransport, AsyncClient from app import schemas from app.api.endpoints import mediaserver as mediaserver_endpoint @@ -9,7 +9,8 @@ from app.domain.context import MediaInfo as CoreMediaInfo from app.schemas.types import MediaSource, MediaType -def test_media_search_response_preserves_core_collection_fields() -> None: +@pytest.mark.asyncio +async def test_media_search_response_preserves_core_collection_fields() -> None: """媒体搜索响应模型应保留 Core MediaInfo 合集输出的全部兼容字段。""" media = CoreMediaInfo(tmdb_info={ "id": 42, @@ -36,7 +37,11 @@ def test_media_search_response_preserves_core_collection_fields() -> None: app = FastAPI() app.include_router(router) - response = TestClient(app).get("/media/search") + async with AsyncClient( + transport=ASGITransport(app=app), + base_url="http://testserver", + ) as client: + response = await client.get("/media/search") assert response.status_code == 200 result = response.json()["data"][0] @@ -52,7 +57,8 @@ def test_media_search_response_preserves_core_collection_fields() -> None: assert "anilist_info" in result -def test_media_response_accepts_cross_source_credit_shapes() -> None: +@pytest.mark.asyncio +async def test_media_response_accepts_cross_source_credit_shapes() -> None: """媒体响应应同时保留豆瓣姓名字符串和 TMDB 演职员对象。""" router = ResponseAPIRouter() @@ -71,7 +77,11 @@ def test_media_response_accepts_cross_source_credit_shapes() -> None: app = FastAPI() app.include_router(router) - response = TestClient(app).get("/recommend") + async with AsyncClient( + transport=ASGITransport(app=app), + base_url="http://testserver", + ) as client: + response = await client.get("/recommend") assert response.status_code == 200 media = response.json()["data"][0] @@ -83,7 +93,8 @@ def test_media_response_accepts_cross_source_credit_shapes() -> None: assert media["directors"][1]["name"] == "导演乙" -def test_media_response_accepts_legacy_source_key() -> None: +@pytest.mark.asyncio +async def test_media_response_accepts_legacy_source_key() -> None: """媒体身份重构前缓存的旧格式条目（source + media_id）应被归一化并正常响应。""" router = ResponseAPIRouter() @@ -103,7 +114,11 @@ def test_media_response_accepts_legacy_source_key() -> None: app = FastAPI() app.include_router(router) - response = TestClient(app).get("/recommend") + async with AsyncClient( + transport=ASGITransport(app=app), + base_url="http://testserver", + ) as client: + response = await client.get("/recommend") assert response.status_code == 200 media = response.json()["data"][0] diff --git a/tests/test_music_subscribe.py b/tests/test_music_subscribe.py index 76b46daf3..5f09c8d64 100644 --- a/tests/test_music_subscribe.py +++ b/tests/test_music_subscribe.py @@ -470,7 +470,7 @@ def test_album_subscription_uses_persisted_snapshot_when_remote_detail_is_unavai media_chain = Mock() media_chain.recognize_media.return_value = None - with patch("app.chain.subscribe.MediaChain", return_value=media_chain): + with patch("app.chain._music.MediaChain", return_value=media_chain): restored = SubscribeChain._recognize_music_subscribe(subscribe) assert restored.music_type == MUSIC_ENTITY_ALBUM @@ -485,7 +485,7 @@ def test_legacy_music_identity_failure_does_not_guess_entity_from_title(): media_chain = Mock() media_chain.recognize_media.return_value = None - with patch("app.chain.subscribe.MediaChain", return_value=media_chain): + with patch("app.chain._music.MediaChain", return_value=media_chain): restored = SubscribeChain._recognize_music_subscribe(subscribe) assert restored is None @@ -501,7 +501,7 @@ def test_album_subscription_without_remote_id_uses_persisted_entity_snapshot(): total_tracks=11, ) - with patch("app.chain.subscribe.MediaChain") as media_chain: + with patch("app.chain._music.MediaChain") as media_chain: restored = SubscribeChain._recognize_music_subscribe(subscribe) assert restored.music_type == MUSIC_ENTITY_ALBUM @@ -543,7 +543,7 @@ def test_legacy_music_subscription_rejects_artist_recognition_result(): media_chain = Mock() media_chain.recognize_media.return_value = artist - with patch("app.chain.subscribe.MediaChain", return_value=media_chain): + with patch("app.chain._music.MediaChain", return_value=media_chain): restored = SubscribeChain._recognize_music_subscribe(subscribe) assert restored is None @@ -569,7 +569,7 @@ def test_album_subscription_preserves_track_count_snapshot_when_remote_omits_it( media_chain = Mock() media_chain.recognize_media.return_value = remote - with patch("app.chain.subscribe.MediaChain", return_value=media_chain): + with patch("app.chain._music.MediaChain", return_value=media_chain): restored = SubscribeChain._recognize_music_subscribe(subscribe) assert restored is not remote diff --git a/tests/test_scheduler_cache_expiry.py b/tests/test_scheduler_cache_expiry.py index 2e3c958f2..1fa26d763 100644 --- a/tests/test_scheduler_cache_expiry.py +++ b/tests/test_scheduler_cache_expiry.py @@ -3,6 +3,7 @@ from unittest.mock import Mock from app import scheduler as scheduler_module from app.scheduler import Scheduler +from app.startup import scheduler_initializer class _BackgroundSchedulerStub: @@ -22,6 +23,28 @@ class _BackgroundSchedulerStub: self.started = True +def test_scheduler_constructor_does_not_start_background_jobs(monkeypatch): + """取得调度器单例不应绕过应用生命周期启动后台任务。""" + init = Mock() + monkeypatch.setattr(Scheduler, "init", init) + + scheduler = object.__new__(Scheduler) + scheduler.__init__() + + init.assert_not_called() + assert scheduler._scheduler is None + + +def test_scheduler_initializer_starts_background_jobs(monkeypatch): + """应用启动入口负责显式启动已经构造的调度器。""" + scheduler = Mock() + monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler)) + + scheduler_initializer.init_scheduler() + + scheduler.init.assert_called_once_with() + + def test_meta_cache_expire_does_not_schedule_bulk_cache_clear(monkeypatch): """单条缓存 TTL 不应再被用于注册整批缓存清理任务。""" background_scheduler = _BackgroundSchedulerStub() @@ -68,6 +91,7 @@ def test_meta_cache_expire_does_not_schedule_bulk_cache_clear(monkeypatch): scheduler._event = threading.Event() scheduler._lock = threading.RLock() scheduler._jobs = {} + scheduler._agent_task_interruptions_reconciled = True scheduler._auth_count = 0 scheduler._auth_message = False