refactor: unify subscribe chain data ports

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

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 插件调用
+  方式均未修改。
+
 ### 总体判断
 
 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**：

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

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] = []

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 = []

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")
 

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,

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), \

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

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)