nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-03 05:44:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(startup): extract network composition owner

Browse Source
This commit is contained in:
jxxghp
2026-08-29 17:10:43 +08:00
parent da67608339
commit 084fa475eb
10 changed files with 363 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
app/startup/composition/network.py Normal file
View File

@@ -0,0 +1,157 @@
"""网络应用端口的宿主组合根。"""
from collections.abc import Mapping
from typing import Any, Callable, Optional, cast
from app.adapters.network.http import AsyncRequestUtils, RequestUtils
from app.adapters.network.ip import IpUtils
from app.application.configuration import get_runtime_settings
from app.application.image import (
ImageResponsePort,
ImageTransport,
InternalAddressPort,
configure_image_ports,
)
from app.application.messaging.ingress import (
MessageIngressPort,
configure_message_ingress_port,
)
from app.application.network import (
NetworkTestResponse,
NetworkTestService,
NetworkTestTransport,
configure_network_test_service,
)
from app.runtime.log import logger
class _NetworkTestTransportAdapter:
"""把通用异步 HTTP Adapter 收窄为网络探测 GET 端口。"""
async def get(
self,
url: str,
*,
proxy: Any = None,
headers: Optional[Mapping[str, str]] = None,
user_agent: Optional[str] = None,
) -> Optional[NetworkTestResponse]:
"""使用固定超时、证书校验和手动重定向策略请求目标。"""
response = await AsyncRequestUtils(
proxies=proxy,
headers=dict(headers) if headers else None,
timeout=10,
ua=user_agent or "",
verify=True,
follow_redirects=False,
).get_res(url, allow_redirects=False)
return cast(Optional[NetworkTestResponse], response)
def _read_network_test_setting(key: str, default: Any = None) -> Any:
"""延迟读取已由组合根发布的部署设置，避免装配阶段提前取值。"""
return get_runtime_settings().get(key, default)
class _ImageTransportAdapter:
"""把通用 HTTP Adapter 收窄为图片应用服务的 GET 端口。"""
def get(
self,
url: str,
*,
options: Mapping[str, Any],
) -> Optional[ImageResponsePort]:
"""同步创建短生命周期请求对象并返回响应。"""
response = RequestUtils(**dict(options)).get_res(url=url)
return cast(Optional[ImageResponsePort], response)
async def async_get(
self,
url: str,
*,
options: Mapping[str, Any],
) -> Optional[ImageResponsePort]:
"""异步创建短生命周期请求对象并返回响应。"""
response = await AsyncRequestUtils(**dict(options)).get_res(url=url)
return cast(Optional[ImageResponsePort], response)
class _InternalAddressAdapter:
"""把通用地址判断收窄为图片代理决策端口。"""
@staticmethod
def is_internal(url: str) -> bool:
"""委托通用 IP 工具判断 URL 是否指向内部地址。"""
probe = cast(Callable[[str], bool], IpUtils.is_internal)
return bool(probe(url))
class _MessageIngressAdapter:
"""通过通用 HTTP Adapter 投递本地消息并负责释放响应。"""
def post(
self,
url: str,
payload: Mapping[str, Any],
*,
timeout: float,
) -> Optional[int]:
"""同步投递消息，关闭响应后返回状态码。"""
response = RequestUtils(timeout=timeout).post_res( # type: ignore[arg-type]
url,
json=dict(payload),
)
if response is None:
return None
try:
return int(response.status_code)
finally:
try:
response.close()
except Exception as error: # noqa: BLE001 - 释放失败不改变投递结果
logger.debug(f"释放本地消息入口响应失败：{error}")
async def async_post(
self,
url: str,
payload: Mapping[str, Any],
*,
timeout: float,
) -> Optional[int]:
"""异步投递消息，关闭响应后返回状态码。"""
response = await AsyncRequestUtils(
timeout=timeout # type: ignore[arg-type]
).post_res(
url,
json=dict(payload),
)
if response is None:
return None
try:
return int(response.status_code)
finally:
try:
await response.aclose()
except Exception as error: # noqa: BLE001 - 释放失败不改变投递结果
logger.debug(f"释放本地消息入口响应失败：{error}")
def configure_application_network_ports() -> None:
"""装配网络探测、图片读取、内部地址判断和消息回环传输端口。"""
network_test_transport: NetworkTestTransport = _NetworkTestTransportAdapter()
image_transport: ImageTransport = _ImageTransportAdapter()
internal_address: InternalAddressPort = _InternalAddressAdapter()
message_ingress: MessageIngressPort = _MessageIngressAdapter()
configure_network_test_service(
NetworkTestService(
transport=network_test_transport,
settings=_read_network_test_setting,
logger=logger,
)
)
configure_image_ports(
transport=image_transport,
internal_address=internal_address,
)
configure_message_ingress_port(message_ingress)

158
app/startup/initializers/modules.py
View File

@@ -1,13 +1,11 @@
import asyncio
import inspect
import sys
from collections.abc import Awaitable, Mapping
from collections.abc import Awaitable
from functools import partial
from typing import Any, Callable, Optional, cast
from typing import Any, Callable, cast
from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper
from app.adapters.network.http import AsyncRequestUtils, RequestUtils
from app.adapters.network.ip import IpUtils
from app.chain.mediaserver import MediaServerChain
from app.chain.tmdb import TmdbChain
@@ -47,16 +45,9 @@ from app.application.configuration import (
TransferRetryConfig,
configure_transfer_retry_config,
get_configured_system_config,
get_runtime_settings,
)
from app.application.history import configure_transfer_history_repository
from app.application.image import (
ImageResponsePort,
ImageTransport,
InternalAddressPort,
configure_image_ports,
configure_wallpaper_providers,
)
from app.application.image import configure_wallpaper_providers
from app.application.messaging.agent import (
dispatch_web_agent_message_event,
shutdown_web_agent_background_tasks,
@@ -69,22 +60,12 @@ from app.application.messaging.chat import (
configure_agent_chat_service,
get_configured_agent_chat_persistence,
)
from app.application.messaging.ingress import (
MessageIngressPort,
configure_message_ingress_port,
)
from app.application.messaging.message import (
MessageHelper,
MessageQueueManager,
stop_message,
)
from app.application.module import configure_module_runtime
from app.application.network import (
NetworkTestResponse,
NetworkTestService,
NetworkTestTransport,
configure_network_test_service,
)
from app.application.outbox import (
ClaimedOutboxMessage,
OutboxDispatcher,
@@ -203,6 +184,7 @@ from app.startup.composition.database import (
start_database_runtime,
stop_database_runtime,
)
from app.startup.composition.network import configure_application_network_ports
from app.startup.composition.security import (
configure_security_access,
configure_security_services,
@@ -470,138 +452,6 @@ def configure_wallpaper_services() -> None:
)
class _NetworkTestTransportAdapter:
"""把通用异步 HTTP Adapter 收窄为网络探测 GET 端口。"""
async def get(
self,
url: str,
*,
proxy: Any = None,
headers: Optional[Mapping[str, str]] = None,
user_agent: Optional[str] = None,
) -> Optional[NetworkTestResponse]:
"""使用固定超时、证书校验和手动重定向策略请求目标。"""
response = await AsyncRequestUtils(
proxies=proxy,
headers=dict(headers) if headers else None,
timeout=10,
ua=user_agent or "",
verify=True,
follow_redirects=False,
).get_res(url, allow_redirects=False)
return cast(Optional[NetworkTestResponse], response)
def _read_network_test_setting(key: str, default: Any = None) -> Any:
"""延迟读取已由组合根发布的部署设置，避免应用服务持有全局对象。"""
return get_runtime_settings().get(key, default)
class _ImageTransportAdapter:
"""把通用 HTTP Adapter 收窄为图片应用服务的 GET 端口。"""
def get(
self,
url: str,
*,
options: Mapping[str, Any],
) -> Optional[ImageResponsePort]:
"""同步创建短生命周期请求对象并返回响应。"""
response = RequestUtils(**dict(options)).get_res(url=url)
return cast(Optional[ImageResponsePort], response)
async def async_get(
self,
url: str,
*,
options: Mapping[str, Any],
) -> Optional[ImageResponsePort]:
"""异步创建短生命周期请求对象并返回响应。"""
response = await AsyncRequestUtils(**dict(options)).get_res(url=url)
return cast(Optional[ImageResponsePort], response)
class _InternalAddressAdapter:
"""把通用地址判断收窄为图片代理决策端口。"""
@staticmethod
def is_internal(url: str) -> bool:
"""返回 URL 是否指向内部地址。"""
probe = cast(Callable[[str], bool], IpUtils.is_internal)
return bool(probe(url))
class _MessageIngressAdapter:
"""通过通用 HTTP Adapter 投递本地消息并负责释放响应。"""
def post(
self,
url: str,
payload: Mapping[str, Any],
*,
timeout: float,
) -> Optional[int]:
"""同步投递消息，关闭响应后返回状态码。"""
response = RequestUtils(timeout=timeout).post_res( # type: ignore[arg-type]
url,
json=dict(payload),
)
if response is None:
return None
try:
return int(response.status_code)
finally:
try:
response.close()
except Exception as error: # noqa: BLE001 - 释放失败不改变投递结果
logger.debug(f"释放本地消息入口响应失败：{error}")
async def async_post(
self,
url: str,
payload: Mapping[str, Any],
*,
timeout: float,
) -> Optional[int]:
"""异步投递消息，关闭响应后返回状态码。"""
response = await AsyncRequestUtils(
timeout=timeout # type: ignore[arg-type]
).post_res(
url,
json=dict(payload),
)
if response is None:
return None
try:
return int(response.status_code)
finally:
try:
await response.aclose()
except Exception as error: # noqa: BLE001 - 释放失败不改变投递结果
logger.debug(f"释放本地消息入口响应失败：{error}")
def configure_application_network_ports() -> None:
"""装配网络探测、图片读取、内部地址判断和消息回环传输端口。"""
network_test_transport: NetworkTestTransport = _NetworkTestTransportAdapter()
image_transport: ImageTransport = _ImageTransportAdapter()
internal_address: InternalAddressPort = _InternalAddressAdapter()
message_ingress: MessageIngressPort = _MessageIngressAdapter()
configure_network_test_service(
NetworkTestService(
transport=network_test_transport,
settings=_read_network_test_setting,
logger=logger,
)
)
configure_image_ports(
transport=image_transport,
internal_address=internal_address,
)
configure_message_ingress_port(message_ingress)
def notify_event_error(title: str, message: str) -> None:
"""将事件总线错误转发到系统消息通道。"""
MessageHelper().put(

2
docs/architecture-optimization-checklist.md
View File

@@ -70,7 +70,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 Python 模块 / 内部依赖边 | 965 / 8,207 | `dependency-baseline.json` 当前快照 |
| 宿主 Python 模块 / 内部依赖边 | 966 / 8,216 | `dependency-baseline.json` 当前快照 |
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

4
docs/architecture-overview.md
View File

@@ -745,8 +745,8 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 965 |
| 内部导入边 | 8,207 |
| Python 模块 | 966 |
| 内部导入边 | 8,216 |
| 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 53（债务已清零，53 条精确 containment） |

6
docs/rules/05-architecture.md
View File

@@ -143,7 +143,11 @@ plugin persistence and legacy API data-port wiring. Initializers call these owne
startup order and must not recreate their concrete construction.
`startup/composition/security.py` owns authentication, user lookup, PassKey and Web
access provider assembly; `initializers/modules.py` only invokes that owner and places
its returned `AuthenticationRuntime` into the host runtime. Database runtime
its returned `AuthenticationRuntime` into the host runtime.
`startup/composition/network.py`
owns concrete network-test, image, internal-address and message-ingress Adapter wiring
while retaining lazy RuntimeSettings reads. Initializers call these owners in
startup order and must not recreate their concrete construction. Database runtime
ownership rejects overlapping lifespans; after a successful worker shutdown, the
configuration/database composition owners must revoke every provider they published
before releasing database connections. Other domain providers remain owned by their

24
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -1089,8 +1089,8 @@
"runtime_only": true
}
},
"edge_count": 8207,
"edge_sha256": "726a95dd9ad2caadc87bbed1395f56e01f0b323217e466380112326fe994e04d",
"edge_count": 8216,
"edge_sha256": "4a7246f19b3fc9ec6e71558d2da17b186c5feb273a44e3160206b3517754e8eb",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -8702,6 +8702,18 @@
"app.startup.composition.database -> app.db.worker",
"app.startup.composition.database -> app.runtime",
"app.startup.composition.database -> app.runtime.settings",
"app.startup.composition.network -> app.adapters",
"app.startup.composition.network -> app.adapters.network",
"app.startup.composition.network -> app.adapters.network.http",
"app.startup.composition.network -> app.adapters.network.ip",
"app.startup.composition.network -> app.application",
"app.startup.composition.network -> app.application.configuration",
"app.startup.composition.network -> app.application.image",
"app.startup.composition.network -> app.application.messaging",
"app.startup.composition.network -> app.application.messaging.ingress",
"app.startup.composition.network -> app.application.network",
"app.startup.composition.network -> app.runtime",
"app.startup.composition.network -> app.runtime.log",
"app.startup.composition.subscription -> app.adapters",
"app.startup.composition.subscription -> app.adapters.external",
"app.startup.composition.subscription -> app.adapters.external.server",
@@ -8824,8 +8836,6 @@
"app.startup.initializers.modules -> app.adapters.network",
"app.startup.initializers.modules -> app.adapters.network.browser",
"app.startup.initializers.modules -> app.adapters.network.doh",
"app.startup.initializers.modules -> app.adapters.network.http",
"app.startup.initializers.modules -> app.adapters.network.ip",
"app.startup.initializers.modules -> app.adapters.system",
"app.startup.initializers.modules -> app.adapters.system.host",
"app.startup.initializers.modules -> app.adapters.system.resource",
@@ -8847,10 +8857,8 @@
"app.startup.initializers.modules -> app.application.messaging",
"app.startup.initializers.modules -> app.application.messaging.agent",
"app.startup.initializers.modules -> app.application.messaging.chat",
"app.startup.initializers.modules -> app.application.messaging.ingress",
"app.startup.initializers.modules -> app.application.messaging.message",
"app.startup.initializers.modules -> app.application.module",
"app.startup.initializers.modules -> app.application.network",
"app.startup.initializers.modules -> app.application.outbox",
"app.startup.initializers.modules -> app.application.security",
"app.startup.initializers.modules -> app.application.security.auth",
@@ -8934,6 +8942,7 @@
"app.startup.initializers.modules -> app.startup.composition.configuration",
"app.startup.initializers.modules -> app.startup.composition.context",
"app.startup.initializers.modules -> app.startup.composition.database",
"app.startup.initializers.modules -> app.startup.composition.network",
"app.startup.initializers.modules -> app.startup.composition.subscription",
"app.startup.initializers.modules -> app.startup.initializers",
"app.startup.initializers.modules -> app.startup.initializers.agent",
@@ -9300,7 +9309,7 @@
"app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions"
],
"module_count": 965,
"module_count": 966,
"modules": [
"app",
"app.adapters",
@@ -10226,6 +10235,7 @@
"app.startup.composition.configuration",
"app.startup.composition.context",
"app.startup.composition.database",
"app.startup.composition.network",
"app.startup.composition.subscription",
"app.startup.initializers",
"app.startup.initializers.agent",

23
tests/test_architecture_dependencies.py
View File

@@ -799,6 +799,29 @@ def test_modules_initializer_delegates_configuration_and_database_composition():
assert "configure_api_data_runtime" in database_source
def test_modules_initializer_delegates_network_composition():
"""模块初始化器只调用网络组合 API，不再内联具体 Adapter 组装。"""
initializer_source = (
APP_ROOT / "startup" / "initializers" / "modules.py"
).read_text(encoding="utf-8")
network_source = (
APP_ROOT / "startup" / "composition" / "network.py"
).read_text(encoding="utf-8")
assert "configure_application_network_ports()" in initializer_source
for detail in (
"_NetworkTestTransportAdapter",
"_ImageTransportAdapter",
"_InternalAddressAdapter",
"_MessageIngressAdapter",
"NetworkTestService(",
"configure_image_ports(",
"configure_message_ingress_port(",
):
assert detail not in initializer_source
assert detail in network_source
def test_canonical_workflow_oper_has_no_legacy_writer_or_duplicate_exports():
"""工作流旧写入口只能存在于 SDK Legacy facade。"""
oper_path = APP_ROOT / "db" / "oper" / "workflow.py"

43
tests/test_configuration_initializer.py
View File

@@ -48,49 +48,6 @@ class _MutableSettings:
return True, ""
@pytest.mark.asyncio
async def test_network_test_transport_enforces_safe_http_options(monkeypatch) -> None:
"""组合根适配器必须固定证书校验、超时和手动重定向策略。"""
captured = {}
response = SimpleNamespace(status_code=200, headers={}, text="ok")
class _RequestUtils:
"""捕获组合根网络探测适配器使用的 HTTP 参数。"""
def __init__(self, **kwargs) -> None:
"""记录通用请求工具的构造参数。"""
captured["options"] = kwargs
async def get_res(self, url, allow_redirects=True):
"""记录请求地址和重定向开关并返回固定响应。"""
captured["url"] = url
captured["allow_redirects"] = allow_redirects
return response
monkeypatch.setattr(modules_initializer, "AsyncRequestUtils", _RequestUtils)
result = await modules_initializer._NetworkTestTransportAdapter().get(
"https://example.com/health",
proxy="http://proxy.example:7890",
headers={"Authorization": "Bearer test"},
user_agent="MoviePilot-Test",
)
assert result is response
assert captured == {
"url": "https://example.com/health",
"allow_redirects": False,
"options": {
"proxies": "http://proxy.example:7890",
"headers": {"Authorization": "Bearer test"},
"timeout": 10,
"ua": "MoviePilot-Test",
"verify": True,
"follow_redirects": False,
},
}
@pytest.mark.asyncio
async def test_runtime_settings_service_uses_legacy_settings_from_startup_root(
monkeypatch,

12
tests/test_message_ingress.py
View File

@@ -125,18 +125,18 @@ async def test_async_forward_message_to_host_uses_same_contract(monkeypatch):
def test_startup_message_ingress_adapter_closes_sync_response(monkeypatch):
"""生产同步注入适配层必须负责关闭 HTTP 响应。"""
from app.startup.initializers import modules as module_initializer
from app.startup.composition import network as network_composition
response = SimpleNamespace(status_code=202, close=MagicMock())
request = MagicMock()
request.post_res.return_value = response
monkeypatch.setattr(
module_initializer,
network_composition,
"RequestUtils",
MagicMock(return_value=request),
)
status_code = module_initializer._MessageIngressAdapter().post(
status_code = network_composition._MessageIngressAdapter().post(
"http://127.0.0.1/message",
{"text": "hello"},
timeout=9,
@@ -149,18 +149,18 @@ def test_startup_message_ingress_adapter_closes_sync_response(monkeypatch):
@pytest.mark.asyncio
async def test_startup_message_ingress_adapter_closes_async_response(monkeypatch):
"""生产异步注入适配层必须负责关闭 HTTP 响应。"""
from app.startup.initializers import modules as module_initializer
from app.startup.composition import network as network_composition
response = SimpleNamespace(status_code=202, aclose=AsyncMock())
request = MagicMock()
request.post_res = AsyncMock(return_value=response)
monkeypatch.setattr(
module_initializer,
network_composition,
"AsyncRequestUtils",
MagicMock(return_value=request),
)
status_code = await module_initializer._MessageIngressAdapter().async_post(
status_code = await network_composition._MessageIngressAdapter().async_post(
"http://127.0.0.1/message",
{"text": "hello"},
timeout=10,

148
tests/test_network_composition.py Normal file
View File

@@ -0,0 +1,148 @@
"""网络应用端口组合根契约测试。"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from app.startup.composition import network as network_composition
@pytest.mark.asyncio
async def test_network_test_transport_enforces_safe_http_options(monkeypatch) -> None:
"""网络探测适配器固定证书校验、超时和手动重定向策略。"""
captured = {}
response = SimpleNamespace(status_code=200, headers={}, text="ok")
class _RequestUtils:
"""捕获网络探测适配器使用的 HTTP 参数。"""
def __init__(self, **kwargs) -> None:
"""记录通用请求工具的构造参数。"""
captured["options"] = kwargs
async def get_res(self, url, allow_redirects=True):
"""记录请求地址和重定向开关并返回固定响应。"""
captured["url"] = url
captured["allow_redirects"] = allow_redirects
return response
monkeypatch.setattr(network_composition, "AsyncRequestUtils", _RequestUtils)
result = await network_composition._NetworkTestTransportAdapter().get(
"https://example.com/health",
proxy="http://proxy.example:7890",
headers={"Authorization": "Bearer test"},
user_agent="MoviePilot-Test",
)
assert result is response
assert captured == {
"url": "https://example.com/health",
"allow_redirects": False,
"options": {
"proxies": "http://proxy.example:7890",
"headers": {"Authorization": "Bearer test"},
"timeout": 10,
"ua": "MoviePilot-Test",
"verify": True,
"follow_redirects": False,
},
}
def test_image_transport_forwards_sync_request_options(monkeypatch) -> None:
"""同步图片适配器完整透传请求参数，并单独传递目标 URL。"""
response = object()
request = MagicMock()
request.get_res.return_value = response
request_factory = MagicMock(return_value=request)
monkeypatch.setattr(network_composition, "RequestUtils", request_factory)
options = {
"timeout": 15,
"proxies": "http://proxy.example:7890",
"headers": {"Accept": "image/*"},
}
result = network_composition._ImageTransportAdapter().get(
"https://images.example/poster.jpg",
options=options,
)
assert result is response
request_factory.assert_called_once_with(**options)
request.get_res.assert_called_once_with(
url="https://images.example/poster.jpg",
)
@pytest.mark.asyncio
async def test_image_transport_forwards_async_request_options(monkeypatch) -> None:
"""异步图片适配器完整透传请求参数，并单独传递目标 URL。"""
response = object()
request = MagicMock()
request.get_res = AsyncMock(return_value=response)
request_factory = MagicMock(return_value=request)
monkeypatch.setattr(network_composition, "AsyncRequestUtils", request_factory)
options = {
"timeout": 20,
"proxies": None,
"verify": True,
}
result = await network_composition._ImageTransportAdapter().async_get(
"https://images.example/backdrop.webp",
options=options,
)
assert result is response
request_factory.assert_called_once_with(**options)
request.get_res.assert_awaited_once_with(
url="https://images.example/backdrop.webp",
)
def test_internal_address_adapter_delegates_to_ip_utility(monkeypatch) -> None:
"""内部地址判断必须原样委托通用 IP 工具。"""
probe = MagicMock(return_value=True)
monkeypatch.setattr(network_composition.IpUtils, "is_internal", probe)
assert network_composition._InternalAddressAdapter().is_internal("http://192.168.1.10/image.jpg") is True
probe.assert_called_once_with("http://192.168.1.10/image.jpg")
def test_network_port_composition_does_not_read_settings_early(monkeypatch) -> None:
"""端口装配只登记延迟 reader，不得在配置发布前读取 RuntimeSettings。"""
get_runtime_settings = MagicMock(side_effect=AssertionError("装配阶段不得读取 RuntimeSettings"))
configured = {}
monkeypatch.setattr(
network_composition,
"get_runtime_settings",
get_runtime_settings,
)
monkeypatch.setattr(
network_composition,
"configure_network_test_service",
lambda service: configured.update(network=service),
)
monkeypatch.setattr(
network_composition,
"configure_image_ports",
lambda **ports: configured.update(image=ports),
)
monkeypatch.setattr(
network_composition,
"configure_message_ingress_port",
lambda port: configured.update(message=port),
)
network_composition.configure_application_network_ports()
get_runtime_settings.assert_not_called()
assert set(configured) == {"network", "image", "message"}
runtime_settings = MagicMock()
runtime_settings.get.return_value = "configured"
get_runtime_settings.side_effect = None
get_runtime_settings.return_value = runtime_settings
assert configured["network"]._settings("PROXY", None) == "configured"
runtime_settings.get.assert_called_once_with("PROXY", None)