From 4346fc05599923dcc01b28281134772166df2245 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 24 Aug 2026 04:43:20 +0800 Subject: [PATCH] refactor: unify subscribe chain data ports --- app/chain/subscribe.py | 38 +++---- .../backend-architecture-next-stage.md | 10 +- docs/rules/05-architecture.md | 2 +- tests/test_architecture_dependencies.py | 21 ++++ tests/test_music_subscribe.py | 18 +-- tests/test_slash_command_interactions.py | 4 +- tests/test_subscribe_chain.py | 104 ++++++++++-------- tests/test_subscribe_files_info.py | 4 +- tests/test_subscribe_oper.py | 2 +- tests/test_subscribe_search_state.py | 8 +- 10 files changed, 125 insertions(+), 86 deletions(-) diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 6012ca28d..8f96938f7 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -35,9 +35,9 @@ from app.domain.meta.metamusic import MetaMusic from app.domain.meta.words import WordsMatcher from app.domain.metainfo import MetaInfo from app.application.chain.data import ( - DownloadHistoryPortProxy as DownloadHistoryOper, - SitePortProxy as SiteOper, - SubscribePortProxy as SubscribeOper, + get_chain_download_history_port, + get_chain_site_port, + get_chain_subscribe_port, ) from app.application.configuration import ( get_chain_runtime_config_snapshot, @@ -1471,7 +1471,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): @staticmethod def _subscription_query() -> SubscriptionQueryService: """构造绑定订阅 Oper 的查询应用服务。""" - return SubscriptionQueryService(SubscribeOper()) + return SubscriptionQueryService(get_chain_subscribe_port()) @classmethod def exists(cls, mediainfo: MediaInfo, meta: MetaBase = None): @@ -1546,7 +1546,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): if not lock_acquired: return - subscribeoper = SubscribeOper() + subscribeoper = get_chain_subscribe_port() if sid: subscribe = subscribeoper.get(sid) subscribes = [subscribe] if subscribe else [] @@ -1804,7 +1804,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): if subscribe.type != MediaType.MOVIE.value: return - SubscribeOper().update(subscribe.id, { + get_chain_subscribe_port().update(subscribe.id, { "current_priority": priority, "last_update": now }) @@ -1955,7 +1955,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): :return: 返回[]代表所有站点命中，返回None代表没有订阅 """ ret_sites = [] - subscribes = SubscribeOper().list() + subscribes = get_chain_subscribe_port().list() if not subscribes: # 没有订阅 return None @@ -2080,7 +2080,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): processed_torrents = self._prepare_match_torrents(torrents) # 所有订阅 - subscribes = SubscribeOper().list(self.get_states_for_search('R')) + subscribes = get_chain_subscribe_port().list(self.get_states_for_search('R')) total_num = len(subscribes) if progress_callback: progress_callback( @@ -2125,7 +2125,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): # 订阅的站点域名列表 domains = [] if subscribe.sites: - domains = SiteOper().get_domains_by_ids(subscribe.sites) + domains = get_chain_site_port().get_domains_by_ids(subscribe.sites) # 识别媒体信息 mediainfo: MediaInfo = MediaChain().recognize_media( meta=meta, @@ -2397,7 +2397,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): ) # 同步外部修改，更新订阅信息 - subscribe = SubscribeOper().get(subscribe.id) + subscribe = get_chain_subscribe_port().get(subscribe.id) # 判断是否要完成订阅 if subscribe: @@ -2426,7 +2426,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): :param progress_callback: 定时服务进度更新回调 """ # 查询所有订阅 - subscribeoper = SubscribeOper() + subscribeoper = get_chain_subscribe_port() subscribes = subscribeoper.list() total_num = len(subscribes) if progress_callback: @@ -2567,7 +2567,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): return logger.info(f'开始刷新follow用户分享订阅 ...') success_count = 0 - subscribeoper = SubscribeOper() + subscribeoper = get_chain_subscribe_port() share_subscribes = MoviePilotServerHelper.get_subscribe_shares() or [] total_num = len(share_subscribes) if progress_callback: @@ -2668,7 +2668,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): :param progress_callback: 定时服务进度更新回调 """ logger.info(f'开始预缓存订阅日历 ...') - subscribes = await SubscribeOper().async_list() + subscribes = await get_chain_subscribe_port().async_list() total_num = len(subscribes) if progress_callback: progress_callback( @@ -2772,7 +2772,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): note = list(set(note).union(set(items))) # 更新订阅 if note: - SubscribeOper().update(subscribe.id, { + get_chain_subscribe_port().update(subscribe.id, { "note": note }) @@ -2836,7 +2836,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): """ if not update_data: return - SubscribeOper().update(subscribe.id, update_data) + get_chain_subscribe_port().update(subscribe.id, update_data) for key, value in update_data.items(): setattr(subscribe, key, value) @@ -3166,7 +3166,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): return SubscribeInteractionHandler( messenger=self, actions=self, - repository=SubscribeOper(), + repository=get_chain_subscribe_port(), delete_subscription=self._delete_subscription, ) @@ -3341,7 +3341,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): site_id = event_data.get("site_id") if not site_id: return - subscribeoper = SubscribeOper() + subscribeoper = get_chain_subscribe_port() if site_id == "*": # 站点被重置 _system_config().set(SystemConfigKey.RssSites, []) @@ -3461,7 +3461,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): episodes[0] = info # 所有下载记录 - downloadhis = DownloadHistoryOper() + downloadhis = get_chain_download_history_port() download_his = downloadhis.get_by_media_identity( media_source=subscribe.media_source, media_id=subscribe.media_id, @@ -3945,7 +3945,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): old_total_episode=old_total_episode, ) update_data["last_update"] = now - SubscribeOper().update(subscribe.id, update_data) + get_chain_subscribe_port().update(subscribe.id, update_data) for key, value in update_data.items(): setattr(subscribe, key, value) subscribe.last_update = now diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index f9fd19b3a..3f5acfb75 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口。 +> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口。 ## 当前复核结论（2026-08-24） @@ -195,6 +195,14 @@ - 兼容边界不变：三个 DB Oper、Proxy 类、下载公开入口、失败指纹、历史 hash 查询、媒体库判重和插件 调用合同均未修改。 +### 长期整改阶段 18：主订阅 Chain 数据端口收口（2026-08-24） + +- `SubscribeChain` 原先把订阅、站点和下载历史三个 `*PortProxy` 别名为数据库 Oper，查询、搜索、匹配、 + 完成检查与订阅文件视图因此和命名数据端口形成双轨；现在统一调用对应 `get_chain_*_port()`。 +- 订阅、音乐与交互测试改为替换命名 getter，架构门禁禁止主订阅 Chain 重新导入这三个 Proxy。 +- 兼容边界不变：三个 DB Oper、Proxy 类、订阅 Chain 公开方法、订阅字段和事件合同、V2/V3 插件调用 + 方式均未修改。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index a3b513c0e..cb77c09d4 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -131,7 +131,7 @@ Session. `app/db/adapters/` is the concrete persistence-adapter layer: it may depend on Application-owned Protocols, UoW/Session and Oper implementations. This deliberate dependency inversion is the only `DB implementation -> Application contract` direction; Application must remain free of DB imports. -Migrated workflow, user, interaction, messaging, music, site, media-server and download +Migrated workflow, user, interaction, messaging, music, site, media-server, download and subscribe Chain consumers use the named `get_chain_*_port()` functions from `app/application/chain/data.py`; they must not alias migration-time `*PortProxy` classes back to database Oper names. Those proxy classes remain compatibility diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 80c444757..d16369c11 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -522,6 +522,27 @@ def test_download_chain_uses_explicit_data_port_getters(): assert violations == [] +def test_subscribe_chain_uses_explicit_data_port_getters(): + """订阅链不得把三个迁移期 PortProxy 伪装成数据库 Oper。""" + path = APP_ROOT / "chain" / "subscribe.py" + tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + forbidden = { + "DownloadHistoryPortProxy", + "SitePortProxy", + "SubscribePortProxy", + } + violations = [ + f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}:{alias.name}" + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) + and node.module == "app.application.chain.data" + for alias in node.names + if alias.name in forbidden + ] + + assert violations == [] + + def test_plugin_components_do_not_reexport_legacy_abi_names(): """新插件组件只提供 canonical 能力，不得复制旧 Helper、Manager 或 Oper 导出。""" violations: list[str] = [] diff --git a/tests/test_music_subscribe.py b/tests/test_music_subscribe.py index c13dc947f..ac9379bf2 100644 --- a/tests/test_music_subscribe.py +++ b/tests/test_music_subscribe.py @@ -343,7 +343,7 @@ def test_album_best_version_requires_confirmed_full_coverage(): chain = SubscribeChain() with patch("app.chain.subscribe.DownloadChain", return_value=download_chain), \ - patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch.object(chain, "_SubscribeChain__finish_subscribe") as finish: chain._download_music_subscribe(subscribe, album, [downloaded]) @@ -440,7 +440,7 @@ def test_music_rss_match_reuses_cached_context_without_second_site_search(): torrent_helper = Mock() torrent_helper.filter_torrent.return_value = True with patch.object(SubscribeChain, "_recognize_music_subscribe", return_value=target), \ - patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain._music.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain._music.TorrentHelper", return_value=torrent_helper), \ patch("app.chain._music.DownloadChain", return_value=download_chain), \ @@ -662,7 +662,7 @@ def test_album_target_sync_does_not_clear_stable_track_count(): ) subscribe_oper = Mock() - with patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper): + with patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper): SubscribeChain._sync_music_subscribe_target(subscribe, album) subscribe_oper.update.assert_not_called() @@ -707,7 +707,7 @@ def test_subscribe_add_music_uses_explicit_entity_recognize(): add_subscribe = Mock(return_value=(1, "")) with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \ - patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain.subscribe.add_subscribe", add_subscribe), \ patch("app.chain.subscribe.MoviePilotServerHelper"), \ patch("app.chain.subscribe.eventmanager"): @@ -759,7 +759,7 @@ def test_subscribe_add_music_routes_new_album_sources( add_subscribe = Mock(return_value=(1, "")) with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \ - patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain.subscribe.add_subscribe", add_subscribe), \ patch("app.chain.subscribe.MoviePilotServerHelper"), \ patch("app.chain.subscribe.eventmanager"): @@ -792,7 +792,7 @@ def test_subscribe_add_rejects_music_entity_mismatch_before_database_write(): add_subscribe = Mock() with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \ - patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain.subscribe.add_subscribe", add_subscribe): sid, err_msg = SubscribeChain().add( title="叶惠美", @@ -819,7 +819,7 @@ def test_subscribe_add_music_fails_fast_on_offline_fallback(): add_subscribe = Mock(return_value=(1, "")) with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \ - patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain.subscribe.add_subscribe", add_subscribe): sid, err_msg = SubscribeChain().add( title="未知曲目", @@ -852,7 +852,7 @@ def test_follow_preserves_album_entity_and_track_count(): system_config = Mock() system_config.get.return_value = ["follow-user"] - with patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + with patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain.subscribe.get_configured_system_config", return_value=system_config), \ patch( "app.chain.subscribe.MoviePilotServerHelper.get_subscribe_shares", @@ -878,7 +878,7 @@ def test_refresh_enables_music_entry_fetch_when_music_subscribe_exists(): torrents_chain = Mock() torrents_chain.refresh.return_value = {} - with patch("app.chain.subscribe.SubscribeOper", return_value=subscribe_oper), \ + with patch("app.chain.subscribe.get_chain_subscribe_port", return_value=subscribe_oper), \ patch("app.chain.subscribe.get_configured_system_config") as system_config, \ patch("app.chain.subscribe.TorrentsChain", return_value=torrents_chain): system_config.return_value.get.return_value = [] diff --git a/tests/test_slash_command_interactions.py b/tests/test_slash_command_interactions.py index c04af324b..e316558b6 100644 --- a/tests/test_slash_command_interactions.py +++ b/tests/test_slash_command_interactions.py @@ -236,8 +236,10 @@ class TestSlashCommandInteractions(unittest.TestCase): ) ] + subscribe_port = SimpleNamespace(list=lambda: fake_subscribes) with patch( - "app.chain.subscribe.SubscribeOper.list", return_value=fake_subscribes + "app.chain.subscribe.get_chain_subscribe_port", + return_value=subscribe_port, ), patch.object(chain, "post_message") as post_message: chain.remote_list(channel=NotificationChannel.Web, userid="u1", source="web") diff --git a/tests/test_subscribe_chain.py b/tests/test_subscribe_chain.py index 31bcf27bf..6a357d0c5 100644 --- a/tests/test_subscribe_chain.py +++ b/tests/test_subscribe_chain.py @@ -257,6 +257,13 @@ def _load_subscribe_chain_class(): def __init__(self, **kwargs): self.__dict__.update(kwargs) + class _SubscribeDeletedEventData: + """提供订阅删除应用服务在隔离加载时需要的事件快照。""" + + def __init__(self, **kwargs): + """保存事件字段，行为与测试所需的 schema 投影一致。""" + self.__dict__.update(kwargs) + mediaserver_schema_module = ensure_module( "app.schemas.mediaserver", types.ModuleType("app.schemas.mediaserver"), @@ -293,6 +300,7 @@ def _load_subscribe_chain_class(): event_schema_module.SubscribeCompletionCheckEventData = ( _SubscribeCompletionCheckEventData ) + event_schema_module.SubscribeDeletedEventData = _SubscribeDeletedEventData logger_module = ensure_module("app.runtime.log", types.ModuleType("app.runtime.log")) @@ -586,7 +594,7 @@ class SubscribeChainTest(TestCase): chain = SubscribeChain() chain.check_and_handle_existing_media = lambda **kwargs: (False, {}) - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper", _SubscribeOper), patch.object( + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port", _SubscribeOper), patch.object( SUBSCRIBE_CHAIN_MODULE, "TorrentHelper", _PlainTorrentHelper, @@ -669,7 +677,7 @@ class SubscribeChainTest(TestCase): chain.filter_torrents = lambda **_kwargs: [context.torrent_info] chain.finish_subscribe_or_not = lambda **_kwargs: None - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper", _SubscribeOper), patch.object( + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port", _SubscribeOper), patch.object( SUBSCRIBE_CHAIN_MODULE, "TorrentHelper", _TorrentHelper, @@ -834,7 +842,7 @@ class SubscribeChainTest(TestCase): with patch.object(SUBSCRIBE_CHAIN_MODULE, "DownloadChain", _DownloadChain), patch.object( SUBSCRIBE_CHAIN_MODULE, - "SubscribeOper", + "get_chain_subscribe_port", _SubscribeOper, ): satisfied, no_exists = chain.resolve_subscribe_missing( @@ -1054,7 +1062,7 @@ class SubscribeChainTest(TestCase): def _metainfo(title): return SimpleNamespace(name=title, begin_season=None, episode_list=[]) - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper", _SubscribeOper), patch.object( + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port", _SubscribeOper), patch.object( SUBSCRIBE_CHAIN_MODULE, "get_configured_system_config", _SystemConfigOper, @@ -1699,7 +1707,7 @@ class SubscribeChainTest(TestCase): chain = SubscribeChain() mediainfo = SimpleNamespace(title_year="Test Show (2026)") - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls: + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port") as subscribe_oper_cls: subscribe_oper = subscribe_oper_cls.return_value subscribe_oper.update.return_value = None @@ -1734,7 +1742,7 @@ class SubscribeChainTest(TestCase): chain = SubscribeChain() mediainfo = SimpleNamespace(title_year="Test Show (2026)") - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, \ + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port") as subscribe_oper_cls, \ patch.object(SUBSCRIBE_CHAIN_MODULE, "logger") as logger_mock: subscribe_oper = subscribe_oper_cls.return_value subscribe_oper.update.return_value = None @@ -1767,7 +1775,7 @@ class SubscribeChainTest(TestCase): chain = SubscribeChain() mediainfo = SimpleNamespace(title_year="Test Show (2026)") - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls: + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port") as subscribe_oper_cls: subscribe_oper = subscribe_oper_cls.return_value subscribe_oper.update.return_value = None @@ -1794,7 +1802,7 @@ class SubscribeChainTest(TestCase): chain = SubscribeChain() mediainfo = SimpleNamespace(title_year="Test Show (2026)") - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls: + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port") as subscribe_oper_cls: subscribe_oper = subscribe_oper_cls.return_value subscribe_oper.update.return_value = None @@ -1826,7 +1834,7 @@ class SubscribeChainTest(TestCase): meta = SimpleNamespace(type=MediaType.TV) mediainfo = SimpleNamespace(title_year="Test Show (2026)") - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, patch.object( + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port") as subscribe_oper_cls, patch.object( SubscribeChain, "_SubscribeChain__finish_subscribe", ) as finish_mock: @@ -1868,7 +1876,7 @@ class SubscribeChainTest(TestCase): get_backdrop_image=lambda: "backdrop", ) - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, \ + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port") as subscribe_oper_cls, \ _patch_media_recognize(SUBSCRIBE_CHAIN_MODULE, mediainfo): subscribe_oper = subscribe_oper_cls.return_value subscribe_oper.list.return_value = [subscribe] @@ -2084,7 +2092,7 @@ class SubscribeNoteTrackingTest(TestCase): def get(self, *args, **kwargs): return subscribe - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper", _SubscribeOper), patch.object( + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port", _SubscribeOper), patch.object( SubscribeChain, "_SubscribeChain__update_movie_download_priority", ), patch.object( @@ -2124,7 +2132,7 @@ class SubscribeNoteTrackingTest(TestCase): def get(self, *args, **kwargs): return subscribe - with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper", _SubscribeOper), patch.object( + with patch.object(SUBSCRIBE_CHAIN_MODULE, "get_chain_subscribe_port", _SubscribeOper), patch.object( SubscribeChain, "_SubscribeChain__is_best_version_complete", return_value=False, @@ -2266,7 +2274,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append((subscribe_id, payload)) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): summary = self.SubscribeChain().backfill_existing_episodes( subscribe, [1, 2, 3, 9, "bad"], @@ -2293,7 +2301,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): summary = self.SubscribeChain().backfill_existing_episodes( subscribe, [1, 2, 3], @@ -2315,7 +2323,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): invalid = self.SubscribeChain().backfill_existing_episodes( subscribe, [1, 2], @@ -2359,7 +2367,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): summary = self.SubscribeChain().backfill_existing_episodes( subscribe, [1], @@ -2381,7 +2389,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): summary = self.SubscribeChain().backfill_existing_episodes( subscribe, [1], @@ -2404,7 +2412,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): summary = self.SubscribeChain().backfill_existing_episodes( subscribe, [1, 2, 3, 4, 5], @@ -2430,7 +2438,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): self.SubscribeChain().backfill_existing_episodes( subscribe, [3], @@ -2453,7 +2461,7 @@ class SubscribeProgressEntrypointTest(TestCase): updates.append(payload) chain = self.SubscribeChain() - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ patch.object(chain, "refresh_subscribe_progress", return_value={ "scene": "unit", "updated": True, @@ -2489,7 +2497,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): summary = self.SubscribeChain()._SubscribeChain__refresh_subscribe_progress_with_no_exists( subscribe=subscribe, no_exists={}, @@ -2515,7 +2523,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): self.SubscribeChain()._SubscribeChain__refresh_subscribe_progress_with_no_exists( subscribe=subscribe, no_exists={}, @@ -2547,7 +2555,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ _patch_media_recognize(self.module, mediainfo), \ patch.object(self.SubscribeChain, "resolve_subscribe_missing", return_value=(False, no_exists)) as resolve_missing: summary = self.SubscribeChain().refresh_subscribe_progress(subscribe, scene="unit") @@ -2580,7 +2588,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): raise AssertionError("resolve failure must not write progress") - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ _patch_media_recognize(self.module, mediainfo), \ patch.object(self.SubscribeChain, "resolve_subscribe_missing", return_value=(False, {})): summary = self.SubscribeChain().refresh_subscribe_progress(subscribe, scene="unit") @@ -2596,7 +2604,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): raise AssertionError("recognition failure must not write progress") - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ _patch_media_recognize(self.module, None): summary = self.SubscribeChain().refresh_subscribe_progress(subscribe, scene="unit") @@ -2618,7 +2626,7 @@ class SubscribeProgressEntrypointTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ patch.object(self.SubscribeChain, "_SubscribeChain__finish_subscribe"): self.SubscribeChain().finish_subscribe_or_not( subscribe=subscribe, @@ -2738,7 +2746,7 @@ class SubscribeProgressConsolidationTest(TestCase): def update(self, subscribe_id, payload): updates.append((subscribe_id, payload)) - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()): SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( subscribe, mediainfo, @@ -2789,7 +2797,7 @@ class SubscribeProgressConsolidationTest(TestCase): chain.resolve_subscribe_missing = _resolve_missing - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", eventmanager, @@ -2832,7 +2840,7 @@ class SubscribeProgressConsolidationTest(TestCase): def update(self, subscribe_id, payload): updates.append((subscribe_id, payload)) - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", eventmanager, @@ -2968,7 +2976,7 @@ class SubscribeProgressConsolidationTest(TestCase): def update(self, subscribe_id, payload): updates.append((subscribe_id, payload)) - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", eventmanager, @@ -3017,7 +3025,7 @@ class SubscribeProgressConsolidationTest(TestCase): def update(self, subscribe_id, payload): updates.append((subscribe_id, payload)) - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", eventmanager, @@ -3066,7 +3074,7 @@ class SubscribeProgressConsolidationTest(TestCase): def update(self, subscribe_id, payload): updates.append((subscribe_id, payload)) - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", eventmanager, @@ -3109,7 +3117,7 @@ class SubscribeProgressConsolidationTest(TestCase): def update(self, *_args, **_kwargs): raise AssertionError("manual total episode must not be updated") - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", _EventManager(), @@ -3150,7 +3158,7 @@ class SubscribeProgressConsolidationTest(TestCase): def update(self, *_args, **_kwargs): raise AssertionError("non-tv subscribe must not be updated") - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", _EventManager(), @@ -3195,7 +3203,7 @@ class SubscribeProgressConsolidationTest(TestCase): chain = SubscribeChain() - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ _patch_media_recognize(module, lambda **_kwargs: self._mediainfo(total_episode=5)): chain.check() @@ -3241,7 +3249,7 @@ class SubscribeProgressConsolidationTest(TestCase): chain = SubscribeChain() - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", eventmanager, @@ -3300,7 +3308,7 @@ class SubscribeProgressConsolidationTest(TestCase): }, ) - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", eventmanager, @@ -3350,7 +3358,7 @@ class SubscribeProgressConsolidationTest(TestCase): chain = SubscribeChain() - with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + with patch.object(module, "get_chain_subscribe_port", return_value=_SubscribeOper()), patch.object( module, "eventmanager", _EventManager(), @@ -3500,7 +3508,7 @@ class SubscribeDownloadFactsTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts( subscribe, mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"), @@ -3521,7 +3529,7 @@ class SubscribeDownloadFactsTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts( subscribe, mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"), @@ -3548,7 +3556,7 @@ class SubscribeDownloadFactsTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts( subscribe, mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"), @@ -3568,7 +3576,7 @@ class SubscribeDownloadFactsTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts( subscribe, mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"), @@ -3596,7 +3604,7 @@ class SubscribeDownloadFactsTest(TestCase): episode_priority={"1": 60}, ) - with patch.object(self.module, "SubscribeOper") as subscribe_oper_cls: + with patch.object(self.module, "get_chain_subscribe_port") as subscribe_oper_cls: self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts( subscribe, mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"), @@ -3615,7 +3623,7 @@ class SubscribeDownloadFactsTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()): + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()): snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts( subscribe, mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"), @@ -3659,7 +3667,7 @@ class SubscribeDownloadFactsTest(TestCase): def update(self, subscribe_id, payload): updates.append(payload) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ patch.object(self.SubscribeChain, "_SubscribeChain__finish_subscribe"): self.SubscribeChain().finish_subscribe_or_not( subscribe=subscribe, @@ -3695,7 +3703,7 @@ class SubscribeDownloadFactsTest(TestCase): download.meta_info = SimpleNamespace(episode_list=[], season_list=[]) chain = self.SubscribeChain() - with patch.object(self.module, "SubscribeOper") as subscribe_oper_cls, \ + with patch.object(self.module, "get_chain_subscribe_port") as subscribe_oper_cls, \ patch.object(chain, "_SubscribeChain__refresh_subscribe_progress_with_no_exists") as refresh_mock, \ patch.object(chain, "_SubscribeChain__finish_subscribe"): subscribe_oper = subscribe_oper_cls.return_value @@ -3732,7 +3740,7 @@ class SubscribeDownloadFactsTest(TestCase): download.meta_info = SimpleNamespace(episode_list=[], season_list=[]) chain = self.SubscribeChain() - with patch.object(self.module, "SubscribeOper") as subscribe_oper_cls, \ + with patch.object(self.module, "get_chain_subscribe_port") as subscribe_oper_cls, \ patch.object(chain, "_SubscribeChain__refresh_subscribe_progress_with_no_exists") as refresh_mock, \ patch.object(chain, "_SubscribeChain__finish_subscribe"): subscribe_oper = subscribe_oper_cls.return_value @@ -3779,7 +3787,7 @@ class SubscribeDownloadFactsTest(TestCase): def finish_probe(subscribe, **_kwargs): finished.append(subscribe.current_priority) - with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \ + with patch.object(self.module, "get_chain_subscribe_port", return_value=_SubscribeOper()), \ patch.object(chain, "_SubscribeChain__finish_subscribe", side_effect=finish_probe): chain.finish_subscribe_or_not( subscribe=subscribe, diff --git a/tests/test_subscribe_files_info.py b/tests/test_subscribe_files_info.py index 2c1506559..21387e1ce 100644 --- a/tests/test_subscribe_files_info.py +++ b/tests/test_subscribe_files_info.py @@ -93,7 +93,7 @@ def test_subscribe_files_info_merges_multiple_mediaservers(): media_chain.recognize_media.return_value = mediainfo chain = SubscribeChain() - with patch("app.chain.subscribe.DownloadHistoryOper") as download_oper, \ + with patch("app.chain.subscribe.get_chain_download_history_port") as download_oper, \ patch("app.chain.subscribe.MediaChain", return_value=media_chain), \ patch.object(chain, "media_files", return_value=None), \ patch.object(chain, "media_exists", side_effect=_media_exists_side_effect), \ @@ -139,7 +139,7 @@ def test_subscribe_files_info_uses_season_zero_for_tv(): media_chain.recognize_media.return_value = mediainfo chain = SubscribeChain() - with patch("app.chain.subscribe.DownloadHistoryOper") as download_oper, \ + with patch("app.chain.subscribe.get_chain_download_history_port") as download_oper, \ patch("app.chain.subscribe.MediaChain", return_value=media_chain), \ patch.object(chain, "media_files", return_value=None), \ patch.object(chain, "media_exists", side_effect=_media_exists_side_effect), \ diff --git a/tests/test_subscribe_oper.py b/tests/test_subscribe_oper.py index 0f81f4ee9..dadfea89f 100644 --- a/tests/test_subscribe_oper.py +++ b/tests/test_subscribe_oper.py @@ -507,7 +507,7 @@ def test_subscribe_chain_exists_forwards_episode_group(): media = _media("eg-1") meta = SimpleNamespace(begin_season=1) - with patch("app.chain.subscribe.SubscribeOper") as subscribe_oper_cls: + with patch("app.chain.subscribe.get_chain_subscribe_port") as subscribe_oper_cls: subscribe_oper_cls.return_value.exists.return_value = True assert SubscribeChain.exists(media, meta) is True diff --git a/tests/test_subscribe_search_state.py b/tests/test_subscribe_search_state.py index 8a4f2ab94..6346eb661 100644 --- a/tests/test_subscribe_search_state.py +++ b/tests/test_subscribe_search_state.py @@ -75,7 +75,7 @@ def test_new_subscribe_search_keeps_state_when_recently_created(monkeypatch) -> """ _SubscribeOper.subscribe = _new_subscribe(datetime.now()) _SubscribeOper.updates = [] - monkeypatch.setattr(subscribe_module, "SubscribeOper", _SubscribeOper) + monkeypatch.setattr(subscribe_module, "get_chain_subscribe_port", _SubscribeOper) media_chain_class = Mock() with patch.object(subscribe_module, "MediaChain", media_chain_class): @@ -92,7 +92,7 @@ def test_new_subscribe_search_marks_state_after_attempt(monkeypatch) -> None: """ _SubscribeOper.subscribe = _new_subscribe(datetime.now() - timedelta(minutes=2)) _SubscribeOper.updates = [] - monkeypatch.setattr(subscribe_module, "SubscribeOper", _SubscribeOper) + monkeypatch.setattr(subscribe_module, "get_chain_subscribe_port", _SubscribeOper) media_chain = Mock() media_chain.recognize_media.return_value = None @@ -117,7 +117,7 @@ def test_targeted_batch_searches_all_ids_without_state_scan(monkeypatch) -> None subscribe_oper.get.side_effect = subscribes.get monkeypatch.setattr( subscribe_module, - "SubscribeOper", + "get_chain_subscribe_port", lambda: subscribe_oper, ) media_chain = Mock() @@ -136,7 +136,7 @@ def test_subscribe_search_aborts_when_lock_times_out(monkeypatch) -> None: """订阅搜索锁超时后必须中止，不能在无锁状态下继续访问订阅。""" monkeypatch.setattr(SubscribeChain, "_rlock", _TimedOutLock()) subscribe_oper = Mock() - monkeypatch.setattr(subscribe_module, "SubscribeOper", subscribe_oper) + monkeypatch.setattr(subscribe_module, "get_chain_subscribe_port", subscribe_oper) progress = Mock() chain = object.__new__(SubscribeChain)