From ac7a20132953c3328d12fddde8a93d00ed13b34f Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 28 Aug 2026 05:53:53 +0800 Subject: [PATCH] refactor(chain): remove dead data port indirection --- app/application/chain/context.py | 2 - app/application/chain/data.py | 76 ------------------- app/chain/__init__.py | 2 - app/chain/workflow.py | 4 +- app/startup/initializers/modules.py | 4 +- docs/architecture-optimization-checklist.md | 6 +- docs/architecture-overview.md | 2 +- docs/architecture-refactor-roadmap.md | 5 +- docs/rules/05-architecture.md | 11 +-- tests/conftest.py | 1 - .../architecture/coverage-baseline.json | 4 +- .../architecture/dependency-baseline.json | 9 +-- .../fixtures/architecture/ruff-baseline.json | 3 - tests/test_architecture_dependencies.py | 67 ++++++++++++---- tests/test_chain_runtime_context.py | 44 ++++++++++- tests/test_workflow_execution.py | 18 ++++- 16 files changed, 130 insertions(+), 128 deletions(-) diff --git a/app/application/chain/context.py b/app/application/chain/context.py index 465e0dcc6..20a9e29ad 100644 --- a/app/application/chain/context.py +++ b/app/application/chain/context.py @@ -6,7 +6,6 @@ from collections.abc import Callable from dataclasses import dataclass, field from typing import Any, Optional -from app.application.chain.data import ChainDataPorts from app.application.chain.events import ChainDurableEventWriter from app.application.configuration import ChainRuntimeConfig from app.runtime.stop import StopState, runtime_stop_state @@ -31,7 +30,6 @@ class ChainRuntimeContext: message_queue_factory: MessageQueueFactory module_dispatcher_factory: ModuleDispatcherFactory legacy_transfer_command: Optional[LegacyTransferCommand] = None - data_ports: Optional[ChainDataPorts] = None durable_event_writer: Optional[ChainDurableEventWriter] = None configuration: ChainRuntimeConfig = field( default_factory=lambda: ChainRuntimeConfig(media_extensions=()) diff --git a/app/application/chain/data.py b/app/application/chain/data.py index 5b736c707..2714285f8 100644 --- a/app/application/chain/data.py +++ b/app/application/chain/data.py @@ -12,10 +12,8 @@ from typing import Any, Optional from app.application.transfer.execution import TransferExecutionRepository from app.application.transfer.workflow import TransferAdmissionRepository -from app.application.workflow import WorkflowExecutionPort OperFactory = Callable[[], Any] -WorkflowExecutionPortFactory = Callable[[], WorkflowExecutionPort] TransferAdmissionRepositoryFactory = Callable[[], TransferAdmissionRepository] TransferExecutionRepositoryFactory = Callable[[], TransferExecutionRepository] @@ -26,7 +24,6 @@ class ChainDataPorts: site: OperFactory subscribe: OperFactory - workflow: WorkflowExecutionPortFactory download_history: OperFactory transfer_history: OperFactory transfer_pending: TransferAdmissionRepositoryFactory @@ -36,72 +33,6 @@ class ChainDataPorts: user: OperFactory -class _PortProxyMeta(type): - """让迁移期的 Oper 名称支持按方法打桩，同时仍转发到组合根端口。""" - - def __getattr__(cls, name: str) -> Any: - """把类级方法访问转发到一个新的端口实例。""" - return getattr(cls(), name) - - -class _ChainDataPortProxy(metaclass=_PortProxyMeta): - """将旧的 Oper 调用形态转发到 Chain 数据端口的内部代理。""" - - port_name: str - - def __getattr__(self, name: str) -> Any: - """转发未被测试替换的数据操作。""" - return getattr(getattr(get_chain_data_ports(), self.port_name)(), name) - - -class SitePortProxy(_ChainDataPortProxy): - """站点数据端口代理。""" - - port_name = "site" - - -class SubscribePortProxy(_ChainDataPortProxy): - """订阅数据端口代理。""" - - port_name = "subscribe" - - -class WorkflowPortProxy(_ChainDataPortProxy): - """工作流数据端口代理。""" - - port_name = "workflow" - - -class DownloadHistoryPortProxy(_ChainDataPortProxy): - """下载历史数据端口代理。""" - - port_name = "download_history" - - -class TransferHistoryPortProxy(_ChainDataPortProxy): - """整理历史数据端口代理。""" - - port_name = "transfer_history" - - -class MediaServerPortProxy(_ChainDataPortProxy): - """媒体服务器数据端口代理。""" - - port_name = "media_server" - - -class DownloadFailurePortProxy(_ChainDataPortProxy): - """下载失败数据端口代理。""" - - port_name = "download_failure" - - -class UserPortProxy(_ChainDataPortProxy): - """用户数据端口代理。""" - - port_name = "user" - - _ports: Optional[ChainDataPorts] = None @@ -109,7 +40,6 @@ def configure_chain_data_ports( *, site: OperFactory, subscribe: OperFactory, - workflow: WorkflowExecutionPortFactory, download_history: OperFactory, transfer_history: OperFactory, transfer_pending: TransferAdmissionRepositoryFactory, @@ -123,7 +53,6 @@ def configure_chain_data_ports( _ports = ChainDataPorts( site=site, subscribe=subscribe, - workflow=workflow, download_history=download_history, transfer_history=transfer_history, transfer_pending=transfer_pending, @@ -151,11 +80,6 @@ def get_chain_subscribe_port() -> Any: return get_chain_data_ports().subscribe() -def get_chain_workflow_port() -> WorkflowExecutionPort: - """返回类型化的工作流执行状态事务端口。""" - return get_chain_data_ports().workflow() - - def get_chain_download_history_port() -> Any: """创建下载历史数据端口实例。""" return get_chain_data_ports().download_history() diff --git a/app/chain/__init__.py b/app/chain/__init__.py index b6ccd5315..4e7770c21 100644 --- a/app/chain/__init__.py +++ b/app/chain/__init__.py @@ -8,7 +8,6 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union, cast from app.application.chain.context import ChainRuntimeContext, get_chain_runtime_context -from app.application.chain.data import get_chain_data_ports from app.application.configuration import ( ChainRuntimeConfig, get_chain_runtime_config_snapshot, @@ -59,7 +58,6 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, self.async_filecache = context.async_file_cache self.runtime_config = context.configuration self.stop_state = context.stop_state - self.data_ports = context.data_ports or get_chain_data_ports() self.durable_event_writer = context.durable_event_writer self._module_dispatcher = context.module_dispatcher_factory( module_catalog=self.modulemanager, diff --git a/app/chain/workflow.py b/app/chain/workflow.py index 02198b61c..e3c14606a 100644 --- a/app/chain/workflow.py +++ b/app/chain/workflow.py @@ -13,9 +13,9 @@ from typing import Any, Callable, List, Optional, Tuple from pydantic import BaseModel -from app.application.chain.data import get_chain_workflow_port from app.application.workflow import ( WorkflowSnapshot, + get_configured_workflow_execution, get_configured_workflow_query, get_workflow_manager, ) @@ -1286,7 +1286,7 @@ class WorkflowChain(ChainBase): :param from_begin: 是否从头开始，默认为True :param progress_callback: 定时服务进度更新回调 """ - workflow_execution = get_chain_workflow_port() + workflow_execution = get_configured_workflow_execution() # 重置工作流 if from_begin: diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index 34756f504..372281fac 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -43,7 +43,7 @@ from app.application.chain.context import ( ChainRuntimeContext, configure_chain_runtime_context_provider, ) -from app.application.chain.data import configure_chain_data_ports, get_chain_data_ports +from app.application.chain.data import configure_chain_data_ports from app.application.chain.events import ( restore_download_added, restore_transfer_result, @@ -269,7 +269,6 @@ def _build_chain_runtime_context() -> ChainRuntimeContext: module_dispatcher_factory=ModuleInvocationDispatcher, legacy_transfer_command=_execute_legacy_transfer_command, configuration=build_chain_runtime_config(legacy_settings), - data_ports=get_chain_data_ports(), durable_event_writer=TransactionalChainDurableEventWriter(SessionFactory), stop_state=runtime_stop_state, ) @@ -872,7 +871,6 @@ async def init_modules() -> HostRuntime: async_session=async_session_scope, ), subscribe=lambda: SubscribeOper(), - workflow=lambda: workflow_execution, download_history=lambda: DownloadHistoryOper(), transfer_history=lambda: TransferHistoryOper(), transfer_pending=lambda: TransactionalTransferAdmissionRepository( diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index d0fac3032..c99e0a126 100644 --- a/docs/architecture-optimization-checklist.md +++ b/docs/architecture-optimization-checklist.md @@ -69,7 +69,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 845 / 6,907 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 845 / 6,902 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 2 | 新增 Chain 包根环；另一个是隔离的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -78,7 +78,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | Python 源码量 | 约 271,400 行 | 60 个文件超过 1,000 行，14 个超过 2,000 行 | | 长方法 | 281 个超过 80 行 | 67 个超过 150 行，23 个超过 250 行；大量是私有方法 | | 全量 mypy 历史债务 | 11,809 / 596 文件 | strict frontier 当前覆盖 41 个文件，本批迁移路径的类型债务已清零 | -| Ruff 历史诊断 | 879 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | +| Ruff 历史诊断 | 878 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率低水位 | Application 78.79%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 @@ -289,6 +289,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` Workflow runtime 和中心服务分享均不接收 ORM。 - [x] Workflow 执行写端由 Chain 直连 `WorkflowExecutionPort` 和短 Session/UoW 事务服务；canonical `WorkflowOper` 只保留显式 Session query/stage，旧无 Session 五方法只存在于 SDK Legacy/Compat。 +- [x] 删除 Chain registry 中零消费者 `*PortProxy`/动态转发和 `ChainRuntimeContext.data_ports` + 伪注入；Workflow 执行服务只在 Application owner 配置一次，不再重复注册到 `ChainDataPorts`。 - [ ] `ChainDataPorts`/`AgentDataPorts` 可暂时保留为兼容聚合器，但字段必须显式、可类型检查。 - [ ] 以一个业务纵切面迁移并验证后，再迁移下一组，禁止一次替换所有 Oper。 - [ ] 增加 AST 门禁，禁止向 `ChainDataPorts`、`AgentDataPorts` 和新的 canonical use-case service diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index a344f12f5..e9a53e1ee 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -705,7 +705,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 845 | -| 内部导入边 | 6,907 | +| 内部导入边 | 6,902 | | 非平凡 SCC | 2（`ARCH-107` 临时 Chain 包根环；精确 containment 的 TMDB 移植包环） | | Direct egress | 66（12 条待迁移债务，54 条精确 containment） | | Module Contract V2 spec | 217（其中 215 个进入 `run_module` 观察面） | diff --git a/docs/architecture-refactor-roadmap.md b/docs/architecture-refactor-roadmap.md index 2d3a4c92c..3c24f4faf 100644 --- a/docs/architecture-refactor-roadmap.md +++ b/docs/architecture-refactor-roadmap.md @@ -100,7 +100,8 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。 | S1-L2 Workflow typed query | `DELIVERED` | S0 | `b4f873654`、`a01a35bcb`：Workflow Application Port 不返回 `Any`/ORM，Session 内投影冻结 DTO，正式调用方全部切换；Unit Tests `33098869736`、Pylint `33098869837` 全绿，覆盖率低水位提升至 Application `78.78%` | | S1-L3 Chain/Agent typed data ports | `ACTIVE` | S1-L2 | `ChainDataPorts`/`AgentDataPorts` 的 raw Oper/`Any` factory 全部清零，兼容调用进入 Legacy 层 | | S1-L3.1 Workflow typed execution | `DELIVERED` | S1-L2 | `17d8be2af`、`b33b29876`：Chain 直连类型化事务服务且单次执行只取一个 port；canonical Oper 删除旧 writer/无 Session 写方法，旧 ABI 只在 `_legacy/workflow.py` 与 Compat overlay；Unit Tests `33103913838`、Pylint `33103913935` 全绿，Application 覆盖率低水位提升至 `78.79%` | -| S1-L3.2 Chain registry/DI | `PLANNED` | S1-L3.1 | 显式类型化 factory，删除 PortProxy 与失效的双重注入，构造器注入真实控制调用 | +| S1-L3.2 Chain registry/DI | `ACTIVE` | S1-L3.1 | 显式类型化 factory，删除 PortProxy 与失效的双重注入，构造器注入真实控制调用 | +| S1-L3.2.1 Registry hygiene | `VERIFIED` | S1-L3.1 | 删除零消费者 PortProxy/动态转发和 `ChainRuntimeContext.data_ports` 伪注入；Workflow 退出 Chain registry，只保留 Application owner 单一配置入口 | | S1-L3.3 DownloadFailure/MediaServer | `PLANNED` | S1-L3.2 | 两组窄 DTO/Port/adapter 清零 raw Oper，不跨远端 I/O 持有 Session | | S1-L3.4 User | `PLANNED` | S1-L3.3 | 认证、偏好与渠道绑定投影冻结快照，User Chain/Agent 不接收 ORM | | S1-L3.5 History | `PLANNED` | S1-L3.4 | Download/Transfer history 统一 typed query/mutation，删除下载历史双事务 fail-open | @@ -152,7 +153,7 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。 | S4-L2 Event strict contract | `PLANNED` | S0-L2.6,S1-L6 | 宿主事件输入/输出按风险 strict，诊断例外只属于第三方插件兼容 | | S4-L3 Complexity v2 | `PLANNED` | S3 | 私有方法、class/file、圈复杂度进入门禁；所有超限通过职责拆分归零 | | S4-L4 全量 mypy 清零 | `PLANNED` | S3,S4-L1,S4-L2 | `mypy-baseline.json` 归零并删除债务接受路径，全宿主 strict 类型通过 | -| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 879 条诊断归零，规则集扩展经过独立审查且新增诊断为零 | +| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 878 条诊断归零，规则集扩展经过独立审查且新增诊断为零 | | S4-L6 Coverage/并发/质量证据 | `PLANNED` | S3,S4-L1,S4-L2 | 高风险包纳入 coverage；raw concurrency 分类清零；Module Quality 有真实 evidence test | ### S5：Plugin、Agent、Domain、Startup 与最终收口 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index f1d965d70..ca0762deb 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -137,11 +137,12 @@ 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, download, subscribe and transfer -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 -boundaries while the other established Chain domains migrate independently. +Migrated user, interaction, messaging, music, site, media-server, download, subscribe and transfer +Chain consumers temporarily use the named `get_chain_*_port()` functions from +`app/application/chain/data.py` while each owner establishes typed DTO/Port contracts. +The retired migration-time `*PortProxy` classes and dynamic `__getattr__` forwarding must not +be recreated; they had no host, SDK or plugin consumers. Workflow execution uses its owning +`app.application.workflow` service directly and must not be registered again in `ChainDataPorts`. Agent orchestration, memory and tool implementations follow the same rule via the named `get_agent_*_port()` functions from `app/application/agentdata.py`. The legacy Agent `*Port` proxy classes remain import-compatible boundaries and diff --git a/tests/conftest.py b/tests/conftest.py index a77c3a651..d7078c0cd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -304,7 +304,6 @@ def configure_plugin_system_services(): configure_chain_data_ports( site=site_repository, subscribe=lambda: SubscribeOper(), - workflow=lambda: workflow_execution, download_history=lambda: DownloadHistoryOper(), transfer_history=lambda: TransferHistoryOper(), transfer_pending=lambda: TransactionalTransferAdmissionRepository( diff --git a/tests/fixtures/architecture/coverage-baseline.json b/tests/fixtures/architecture/coverage-baseline.json index 23b52112d..dfacd883c 100644 --- a/tests/fixtures/architecture/coverage-baseline.json +++ b/tests/fixtures/architecture/coverage-baseline.json @@ -1,8 +1,8 @@ { "application": { - "covered_lines": 10001, + "covered_lines": 9978, "percent": 78.79, - "statements": 12694 + "statements": 12664 }, "domain": { "covered_lines": 3392, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index ac1404872..41ebc1d40 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1441,8 +1441,8 @@ "runtime_only": true } }, - "edge_count": 6907, - "edge_sha256": "8b3bb489cfc9573e377d0ee21337633b842b9969fc69fe99b86f547613b86a2a", + "edge_count": 6902, + "edge_sha256": "e2adb079d1df7415b81cbfa8358536e29e06276c2cb5af3a6f6c634e6f58fb20", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3979,7 +3979,6 @@ "app.application.backup -> app.runtime.log", "app.application.chain.context -> app.application", "app.application.chain.context -> app.application.chain", - "app.application.chain.context -> app.application.chain.data", "app.application.chain.context -> app.application.chain.events", "app.application.chain.context -> app.application.configuration", "app.application.chain.context -> app.runtime", @@ -3988,7 +3987,6 @@ "app.application.chain.data -> app.application.transfer", "app.application.chain.data -> app.application.transfer.execution", "app.application.chain.data -> app.application.transfer.workflow", - "app.application.chain.data -> app.application.workflow", "app.application.chain.events -> app.application", "app.application.chain.events -> app.application.history", "app.application.chain.events -> app.application.transfer", @@ -4480,7 +4478,6 @@ "app.chain -> app.application", "app.chain -> app.application.chain", "app.chain -> app.application.chain.context", - "app.chain -> app.application.chain.data", "app.chain -> app.application.configuration", "app.chain -> app.chain._messaging", "app.chain -> app.chain._recognition", @@ -5065,8 +5062,6 @@ "app.chain.webhook -> app.schemas", "app.chain.webhook -> app.schemas.types", "app.chain.workflow -> app.application", - "app.chain.workflow -> app.application.chain", - "app.chain.workflow -> app.application.chain.data", "app.chain.workflow -> app.application.workflow", "app.chain.workflow -> app.chain", "app.chain.workflow -> app.runtime", diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index 4290e7dad..4befecb8f 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -1213,9 +1213,6 @@ "E402": 4, "I001": 1 }, - "tests/test_chain_runtime_context.py": { - "I001": 1 - }, "tests/test_cli_auto_update.py": { "I001": 1 }, diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index aa0026160..51b6ad743 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -347,8 +347,8 @@ def test_workflow_query_adapter_owns_projection_sessions(): assert "_project_workflow(record)" in source -def test_workflow_execution_chain_uses_typed_transaction_port(): - """工作流 Chain 写端不得再经过 raw Oper 或重复获取全局端口。""" +def test_workflow_execution_chain_uses_single_application_owned_port(): + """工作流 Chain 写端必须只使用 Application owner 的唯一配置入口。""" contract_path = APP_ROOT / "application" / "workflow.py" contract_tree = ast.parse( contract_path.read_text(encoding="utf-8"), @@ -384,29 +384,64 @@ def test_workflow_execution_chain_uses_typed_transaction_port(): for node in data_tree.body if isinstance(node, ast.ClassDef) and node.name == "ChainDataPorts" ) - workflow_field = next( - node + data_fields = { + node.target.id for node in data_class.body if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) - and node.target.id == "workflow" - ) - workflow_getter = next( - node + } + data_functions = { + node.name for node in data_tree.body - if isinstance(node, ast.FunctionDef) - and node.name == "get_chain_workflow_port" - ) - assert ast.unparse(workflow_field.annotation) == "WorkflowExecutionPortFactory" - assert ast.unparse(workflow_getter.returns) == "WorkflowExecutionPort" + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + assert "workflow" not in data_fields + assert "get_chain_workflow_port" not in data_functions chain_source = (APP_ROOT / "chain" / "workflow.py").read_text(encoding="utf-8") startup_source = ( APP_ROOT / "startup" / "initializers" / "modules.py" ).read_text(encoding="utf-8") - assert chain_source.count("get_chain_workflow_port()") == 1 - assert "workflow=lambda: workflow_execution" in startup_source - assert "workflow=lambda: WorkflowOper()" not in startup_source + assert chain_source.count("get_configured_workflow_execution()") == 1 + assert "get_chain_workflow_port" not in chain_source + assert "configure_workflow_execution(workflow_execution)" in startup_source + assert "workflow=lambda:" not in startup_source + + +def test_chain_registry_has_no_dynamic_proxies_or_dead_context_injection(): + """Chain registry 不得恢复零消费者动态代理或失效 data_ports 伪注入。""" + data_path = APP_ROOT / "application" / "chain" / "data.py" + data_tree = ast.parse( + data_path.read_text(encoding="utf-8"), + filename=str(data_path), + ) + proxy_classes = { + node.name + for node in data_tree.body + if isinstance(node, ast.ClassDef) + and (node.name.endswith("PortProxy") or node.name == "_PortProxyMeta") + } + dynamic_getters = { + node.name + for node in ast.walk(data_tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == "__getattr__" + } + assert proxy_classes == set() + assert dynamic_getters == set() + + context_source = ( + APP_ROOT / "application" / "chain" / "context.py" + ).read_text(encoding="utf-8") + chain_base_source = (APP_ROOT / "chain" / "__init__.py").read_text( + encoding="utf-8" + ) + startup_source = ( + APP_ROOT / "startup" / "initializers" / "modules.py" + ).read_text(encoding="utf-8") + assert "data_ports" not in context_source + assert "self.data_ports" not in chain_base_source + assert "data_ports=" not in startup_source def test_canonical_workflow_oper_has_no_legacy_writer_or_duplicate_exports(): diff --git a/tests/test_chain_runtime_context.py b/tests/test_chain_runtime_context.py index 833d97d85..6d7cc410c 100644 --- a/tests/test_chain_runtime_context.py +++ b/tests/test_chain_runtime_context.py @@ -2,9 +2,12 @@ from unittest.mock import Mock +import pytest + +from app.application.chain import context as chain_context +from app.application.chain import data as chain_data from app.application.chain.context import ChainRuntimeContext from app.application.configuration import ChainRuntimeConfig -from app.application.chain import context as chain_context from app.chain import ChainBase from app.runtime.extensions.module.dispatcher import ModuleInvocationDispatcher @@ -51,3 +54,42 @@ def test_no_arg_chain_uses_compatibility_context_provider(monkeypatch) -> None: provider.assert_called_once_with() assert chain.modulemanager is context.module_manager assert chain.pluginmanager is context.plugin_manager + + +def test_chain_runtime_context_rejects_unconfigured_provider(monkeypatch) -> None: + """未由组合根配置运行上下文时必须显式拒绝无参 Chain。""" + monkeypatch.setattr( + chain_context, + "_context_provider", + chain_context._unconfigured_chain_runtime_context, + ) + + with pytest.raises(RuntimeError, match="Chain 运行上下文尚未由启动组合根配置"): + chain_context.get_chain_runtime_context() + + +def test_chain_data_registry_rejects_unconfigured_and_returns_factories( + monkeypatch, +) -> None: + """数据 registry 未配置时拒绝访问，配置后按字段返回工厂实例。""" + monkeypatch.setattr(chain_data, "_ports", None) + + with pytest.raises(RuntimeError, match="Chain 数据端口尚未配置"): + chain_data.get_chain_data_ports() + + media_server = Mock() + user = Mock() + chain_data.configure_chain_data_ports( + site=Mock, + subscribe=Mock, + download_history=Mock, + transfer_history=Mock, + transfer_pending=Mock, + transfer_execution=Mock, + media_server=lambda: media_server, + download_failure=Mock, + user=lambda: user, + ) + + assert chain_data.get_chain_media_server_port() is media_server + assert chain_data.get_chain_user_port() is user diff --git a/tests/test_workflow_execution.py b/tests/test_workflow_execution.py index 36fcc1480..0cfd69200 100644 --- a/tests/test_workflow_execution.py +++ b/tests/test_workflow_execution.py @@ -643,7 +643,11 @@ def test_workflow_chain_process_serializes_circular_context(monkeypatch): return fake_oper monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager) - monkeypatch.setattr(workflow_module, "get_chain_workflow_port", get_execution_port) + monkeypatch.setattr( + workflow_module, + "get_configured_workflow_execution", + get_execution_port, + ) monkeypatch.setattr( workflow_module, "get_configured_workflow_query", @@ -952,7 +956,11 @@ def test_workflow_chain_rejects_execution_before_persisting_running_state(monkey workflowoper = _FakeWorkflowOper(workflow) manager = RejectingWorkflowManager([]) monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: manager) - monkeypatch.setattr(workflow_module, "get_chain_workflow_port", lambda: workflowoper) + monkeypatch.setattr( + workflow_module, + "get_configured_workflow_execution", + lambda: workflowoper, + ) monkeypatch.setattr( workflow_module, "get_configured_workflow_query", @@ -995,7 +1003,11 @@ def test_workflow_chain_releases_admitted_owner_when_start_fails(monkeypatch): manager._executions = {} workflowoper = FailingWorkflowOper(_build_workflow()) monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: manager) - monkeypatch.setattr(workflow_module, "get_chain_workflow_port", lambda: workflowoper) + monkeypatch.setattr( + workflow_module, + "get_configured_workflow_execution", + lambda: workflowoper, + ) monkeypatch.setattr( workflow_module, "get_configured_workflow_query",