Kris
b323922cdc
本次提交清理了多个单元测试文件中的多余空行，调整了部分导入的顺序，同时修复了一处feishu错误翻译的导入顺序问题，移除了application/conftest.py中未使用的fake_report_repo fixture，优化代码格式提升可读性。
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
"""channels 限界上下文单元测试跨层通用 fixture。
|
||
|
||
本文件提供横跨 core / application / adapters / plugins 各层复用的被驱动
|
||
端口桩（LoggerPort / ConfigPort / CachePort / EventPublisherPort /
|
||
TracerPort）。桩使用 ``unittest.mock`` 的 ``MagicMock`` / ``AsyncMock``，
|
||
不实现真实逻辑；端口方法签名以 ``yuxi.channels.contract.ports.driven``
|
||
下的 Protocol 为准。
|
||
|
||
注意：LoggerPort / TracerPort 的方法在端口协议中均为 ``async def``，因此
|
||
此处使用 ``MagicMock(spec=...)`` —— MagicMock 会自动为 async 方法生成
|
||
AsyncMock（可 ``await``），与端口协议保持一致。单文件独有的 helper 应保
|
||
留在对应测试文件内（见 testing-guidelines.md）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
from yuxi.channels.contract.ports.driven.cache_port import CachePort
|
||
from yuxi.channels.contract.ports.driven.config_port import ConfigPort
|
||
from yuxi.channels.contract.ports.driven.event_publisher_port import (
|
||
EventPublisherPort,
|
||
)
|
||
from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
|
||
from yuxi.channels.contract.ports.driven.tracer_port import TracerPort
|
||
from yuxi.channels.core.service.config_manager import ConfigManager
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_logger() -> LoggerPort:
|
||
"""LoggerPort 桩。
|
||
|
||
``MagicMock(spec=LoggerPort)`` 自动为 ``log`` / ``debug`` / ``info`` /
|
||
``warn`` / ``error`` / ``exception`` 生成 AsyncMock（可 ``await``），
|
||
默认返回 None。需要断言调用参数时直接使用对应方法的 ``assert_*``。
|
||
"""
|
||
return MagicMock(spec=LoggerPort)
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_config() -> ConfigPort:
|
||
"""ConfigPort 桩，``get`` 默认返回 None，``update`` 默认返回 None。"""
|
||
mock = AsyncMock(spec=ConfigPort)
|
||
mock.get.return_value = None
|
||
mock.update.return_value = None
|
||
return mock
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_config_manager() -> ConfigManager:
|
||
"""ConfigManager 桩。
|
||
|
||
``AsyncMock(spec=ConfigManager)`` 自动为 ``update`` / ``updateInternal``
|
||
/ ``rollback`` 等 async 方法生成 AsyncMock。``update`` /
|
||
``updateInternal`` 默认返回 None，调用方按需覆盖返回值。
|
||
"""
|
||
mock = AsyncMock(spec=ConfigManager)
|
||
mock.update.return_value = None
|
||
mock.updateInternal.return_value = None
|
||
return mock
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_cache() -> CachePort:
|
||
"""CachePort 桩，预设常用方法返回值。
|
||
|
||
``get`` 返回 None，``set`` / ``delete`` / ``expire`` /
|
||
``acquireAdvisoryLock`` / ``releaseAdvisoryLock`` / ``ping`` 返回 True，
|
||
``incr`` 返回 1。
|
||
"""
|
||
mock = AsyncMock(spec=CachePort)
|
||
mock.get.return_value = None
|
||
mock.set.return_value = True
|
||
mock.delete.return_value = True
|
||
mock.incr.return_value = 1
|
||
mock.expire.return_value = True
|
||
mock.acquireAdvisoryLock.return_value = True
|
||
mock.releaseAdvisoryLock.return_value = True
|
||
mock.ping.return_value = True
|
||
return mock
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_event_publisher() -> EventPublisherPort:
|
||
"""EventPublisherPort 桩，``publish`` 默认返回 None。"""
|
||
mock = AsyncMock(spec=EventPublisherPort)
|
||
mock.publish.return_value = None
|
||
return mock
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_tracer() -> TracerPort:
|
||
"""TracerPort 桩。
|
||
|
||
``MagicMock(spec=TracerPort)`` 自动为 ``startSpan`` / ``endSpan`` /
|
||
``getCurrentTrace`` 生成 AsyncMock（可 ``await``）。``startSpan`` 默认
|
||
返回 Span-like MagicMock，``endSpan`` 返回 None，``getCurrentTrace``
|
||
返回 Option-like MagicMock，调用方可按需覆盖。
|
||
"""
|
||
mock = MagicMock(spec=TracerPort)
|
||
mock.startSpan.return_value = MagicMock()
|
||
mock.endSpan.return_value = None
|
||
mock.getCurrentTrace.return_value = MagicMock()
|
||
return mock
|