Kris
361dbe2149
本次提交修复了多个测试文件中的问题： 1. 为各管道测试类添加tracer_port属性初始化 2. 修复QQBot客户端关闭异常的错误类型匹配 3. 修正配对接口查询的状态值小写格式 4. 移除ChannelType枚举值的显式.value调用 5. 新增QQ常量的导出项与Instagram测试桩函数 6. 修复API接口返回值解析，正确访问data字段 7. 调整企业微信富文本消息的断言结构 8. 修正微信iLink的消息类型测试用例 9. 替换废弃的datetime.utc相关导入为UTC常量 10. 修复QQBot白名单适配器的返回值结构 11. 调整outbox工具的channel_msg_id处理逻辑 12. 新增多个适配器与服务的测试用例，覆盖异常降级、缓存处理等场景 13. 重构部分QQBot适配器的辅助函数测试，清理冗余代码 14. 修复微信iLink客户端的上传接口调用参数与缓存清理逻辑 15. 修正配置导出接口的敏感字段过滤与返回值结构
491 lines
16 KiB
Python
491 lines
16 KiB
Python
"""yuxi.channels.application.pipeline.base 单元测试。
|
||
|
||
覆盖 ``Pipeline``：
|
||
- ``run(ctx)``：正常执行所有 stage / 某 stage 失败时停止后续 stage
|
||
- ``_validateCompensateReferences()``：补偿引用校验（指向不存在的 stage 报错）
|
||
- ``_runCompensate()``：执行补偿逻辑
|
||
- ``_findStage()``：按名称查找 stage
|
||
- ``_logStageExecution()``：日志记录
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Callable
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
from yuxi.channels.application.pipeline.base import Pipeline
|
||
from yuxi.channels.contract.errors import PipelineConfigError, ValidationError
|
||
from yuxi.channels.contract.plugin.extension_point import FailureStrategy
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
# ─── helper ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class _FakeStage:
|
||
"""满足 ``AppStage`` Protocol 的可编排测试 stage。
|
||
|
||
``process`` 默认返回 True；可通过 ``result`` / ``raise_error`` 覆盖行为。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
stage_id: str,
|
||
*,
|
||
result: bool = True,
|
||
raise_error: Exception | None = None,
|
||
compensate: str | None = None,
|
||
failure: FailureStrategy = FailureStrategy.TERMINATE,
|
||
condition: Callable[[Any], bool] | None = None,
|
||
) -> None:
|
||
self.id = stage_id
|
||
self.reads: tuple[str, ...] = ()
|
||
self.writes: tuple[str, ...] = ()
|
||
self.idempotent = True
|
||
self.thread_safe = True
|
||
self.failure = failure
|
||
self.compensate = compensate
|
||
self.condition = condition
|
||
self._result = result
|
||
self._raise_error = raise_error
|
||
self.process = AsyncMock(return_value=self._result)
|
||
if raise_error is not None:
|
||
self.process = AsyncMock(side_effect=raise_error)
|
||
|
||
|
||
class _FakeCtx:
|
||
"""管道上下文桩，携带 ``trace_id`` 与降级字段。"""
|
||
|
||
def __init__(self, trace_id: str = "trace-001") -> None:
|
||
self.trace_id = trace_id
|
||
self.degraded = False
|
||
self.degraded_reason = None
|
||
|
||
|
||
class _FakeTracer:
|
||
"""``TracerPort`` 桩，记录 ``startSpan`` / ``endSpan`` 调用以便断言。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.start_calls: list[tuple[str, str | None]] = []
|
||
self.end_calls: list[tuple[str, str]] = []
|
||
self._seq = 0
|
||
|
||
async def startSpan(self, name, trace_id=None, parent_span_id=None): # noqa: ANN001
|
||
self._seq += 1
|
||
span = SimpleNamespace(name=name, trace_id=trace_id, span_id=f"span-{self._seq}")
|
||
self.start_calls.append((name, trace_id))
|
||
return span
|
||
|
||
async def endSpan(self, span, status="ok"): # noqa: ANN001
|
||
self.end_calls.append((span.name, status))
|
||
|
||
async def getCurrentTrace(self):
|
||
return None
|
||
|
||
|
||
# ─── run ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestPipelineRun:
|
||
"""``Pipeline.run`` 顺序执行与失败策略。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_executes_all_stages_when_all_succeed(self):
|
||
# Arrange
|
||
s1 = _FakeStage("s1")
|
||
s2 = _FakeStage("s2")
|
||
pipeline = Pipeline(name="test", stages=[s1, s2])
|
||
ctx = _FakeCtx()
|
||
|
||
# Act
|
||
ok, err = await pipeline.run(ctx)
|
||
|
||
# Assert
|
||
assert ok is True
|
||
assert err is None
|
||
s1.process.assert_awaited_once_with(ctx)
|
||
s2.process.assert_awaited_once_with(ctx)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_stops_subsequent_stages_on_terminate_failure(self):
|
||
# Arrange
|
||
s1 = _FakeStage("s1", result=False, failure=FailureStrategy.TERMINATE)
|
||
s2 = _FakeStage("s2")
|
||
pipeline = Pipeline(name="test", stages=[s1, s2])
|
||
ctx = _FakeCtx()
|
||
|
||
# Act
|
||
ok, err = await pipeline.run(ctx)
|
||
|
||
# Assert
|
||
assert ok is False
|
||
s1.process.assert_awaited_once_with(ctx)
|
||
s2.process.assert_not_awaited()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_skip_strategy_continues_next_stage(self):
|
||
# Arrange
|
||
s1 = _FakeStage("s1", result=False, failure=FailureStrategy.SKIP)
|
||
s2 = _FakeStage("s2", result=True)
|
||
pipeline = Pipeline(name="test", stages=[s1, s2])
|
||
ctx = _FakeCtx()
|
||
|
||
# Act
|
||
ok, err = await pipeline.run(ctx)
|
||
|
||
# Assert
|
||
assert ok is True
|
||
assert err is None
|
||
s2.process.assert_awaited_once_with(ctx)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_degrade_strategy_marks_context_and_continues(self):
|
||
# Arrange
|
||
s1 = _FakeStage("s1", result=False, failure=FailureStrategy.DEGRADE)
|
||
s2 = _FakeStage("s2", result=True)
|
||
pipeline = Pipeline(name="test", stages=[s1, s2])
|
||
ctx = _FakeCtx()
|
||
|
||
# Act
|
||
ok, err = await pipeline.run(ctx)
|
||
|
||
# Assert
|
||
assert ok is True
|
||
assert ctx.degraded is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_condition_false_skips_stage(self):
|
||
# Arrange
|
||
s1 = _FakeStage("s1", condition=lambda _ctx: False)
|
||
s2 = _FakeStage("s2")
|
||
pipeline = Pipeline(name="test", stages=[s1, s2])
|
||
ctx = _FakeCtx()
|
||
|
||
# Act
|
||
ok, err = await pipeline.run(ctx)
|
||
|
||
# Assert
|
||
assert ok is True
|
||
s1.process.assert_not_awaited()
|
||
s2.process.assert_awaited_once_with(ctx)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_native_exception_translated_to_internal_error(self):
|
||
# Arrange
|
||
native = RuntimeError("boom")
|
||
s1 = _FakeStage("s1", raise_error=native, failure=FailureStrategy.TERMINATE)
|
||
pipeline = Pipeline(name="test", stages=[s1])
|
||
ctx = _FakeCtx(trace_id="trace-native")
|
||
|
||
# Act
|
||
ok, err = await pipeline.run(ctx)
|
||
|
||
# Assert
|
||
assert ok is False
|
||
assert err is not None
|
||
assert err.error_code == "INTERNAL"
|
||
assert err.trace_id == "trace-native"
|
||
assert err.cause is native
|
||
|
||
|
||
# ─── _validateCompensateReferences ─────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestValidateCompensateReferences:
|
||
"""``_validateCompensateReferences`` 构造期补偿引用校验。"""
|
||
|
||
def test_valid_compensate_reference_does_not_raise(self):
|
||
# Arrange
|
||
compensate_stage = _FakeStage("rollback")
|
||
s1 = _FakeStage("s1", compensate="rollback")
|
||
|
||
# Act
|
||
pipeline = Pipeline(name="test", stages=[s1, compensate_stage])
|
||
|
||
# Assert
|
||
assert pipeline.stages == [s1, compensate_stage]
|
||
|
||
def test_unknown_compensate_reference_raises_pipeline_config_error(self):
|
||
# Arrange
|
||
s1 = _FakeStage("s1", compensate="nonexistent")
|
||
|
||
# Act / Assert
|
||
with pytest.raises(PipelineConfigError) as exc_info:
|
||
Pipeline(name="test", stages=[s1])
|
||
|
||
assert exc_info.value.pipeline == "test"
|
||
assert exc_info.value.stage_id == "s1"
|
||
assert exc_info.value.compensate == "nonexistent"
|
||
|
||
def test_none_compensate_does_not_raise(self):
|
||
# Arrange
|
||
s1 = _FakeStage("s1", compensate=None)
|
||
s2 = _FakeStage("s2", compensate=None)
|
||
|
||
# Act
|
||
pipeline = Pipeline(name="test", stages=[s1, s2])
|
||
|
||
# Assert
|
||
assert len(pipeline.stages) == 2
|
||
|
||
|
||
# ─── _runCompensate ────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestRunCompensate:
|
||
"""``_runCompensate`` 补偿执行逻辑。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_compensate_executes_compensate_stage_and_returns_original_error(self):
|
||
# Arrange
|
||
compensate_stage = _FakeStage("rollback")
|
||
s1 = _FakeStage("s1", compensate="rollback", failure=FailureStrategy.COMPENSATE)
|
||
pipeline = Pipeline(name="test", stages=[s1, compensate_stage])
|
||
original_err = ValidationError("field", "boom", trace_id="trace-001")
|
||
|
||
# Act
|
||
result = await pipeline._runCompensate(_FakeCtx(), "rollback", original_err)
|
||
|
||
# Assert
|
||
assert result is original_err
|
||
compensate_stage.process.assert_awaited_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_compensate_returns_original_error_when_name_is_none(self):
|
||
# Arrange
|
||
pipeline = Pipeline(name="test", stages=[_FakeStage("s1")])
|
||
original_err = ValidationError("field", "boom")
|
||
|
||
# Act
|
||
result = await pipeline._runCompensate(_FakeCtx(), None, original_err)
|
||
|
||
# Assert
|
||
assert result is original_err
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_compensate_returns_original_error_when_stage_not_found(self):
|
||
# Arrange
|
||
pipeline = Pipeline(name="test", stages=[_FakeStage("s1")])
|
||
original_err = ValidationError("field", "boom")
|
||
|
||
# Act
|
||
result = await pipeline._runCompensate(_FakeCtx(), "missing", original_err)
|
||
|
||
# Assert
|
||
assert result is original_err
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_compensate_swallows_compensate_exception_and_logs(self, fake_logger):
|
||
# Arrange
|
||
compensate_stage = _FakeStage("rollback", raise_error=RuntimeError("compensate failed"))
|
||
s1 = _FakeStage("s1", compensate="rollback")
|
||
pipeline = Pipeline(name="test", stages=[s1, compensate_stage], logger=fake_logger)
|
||
original_err = ValidationError("field", "boom", trace_id="trace-001")
|
||
|
||
# Act
|
||
result = await pipeline._runCompensate(_FakeCtx(), "rollback", original_err)
|
||
|
||
# Assert
|
||
assert result is original_err
|
||
fake_logger.error.assert_awaited_once()
|
||
|
||
|
||
# ─── _findStage ────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestFindStage:
|
||
"""``_findStage`` 按 id 查找阶段。"""
|
||
|
||
def test_find_stage_returns_matching_stage(self):
|
||
# Arrange
|
||
target = _FakeStage("target")
|
||
other = _FakeStage("other")
|
||
pipeline = Pipeline(name="test", stages=[other, target])
|
||
|
||
# Act
|
||
result = pipeline._findStage("target")
|
||
|
||
# Assert
|
||
assert result is target
|
||
|
||
def test_find_stage_returns_none_when_not_found(self):
|
||
# Arrange
|
||
pipeline = Pipeline(name="test", stages=[_FakeStage("s1")])
|
||
|
||
# Act
|
||
result = pipeline._findStage("missing")
|
||
|
||
# Assert
|
||
assert result is None
|
||
|
||
|
||
# ─── _logStageExecution ────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestLogStageExecution:
|
||
"""``_logStageExecution`` 结构化日志输出。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_log_stage_execution_calls_info_on_success(self, fake_logger):
|
||
# Arrange
|
||
pipeline = Pipeline(name="test", stages=[_FakeStage("s1")], logger=fake_logger)
|
||
ctx = _FakeCtx(trace_id="trace-log")
|
||
|
||
# Act
|
||
await pipeline._logStageExecution(ctx, "s1", 12.5, True, None)
|
||
|
||
# Assert
|
||
fake_logger.info.assert_awaited_once()
|
||
call_args = fake_logger.info.call_args
|
||
assert call_args.kwargs["trace_id"] == "trace-log"
|
||
assert call_args.kwargs["stage_id"] == "s1"
|
||
assert call_args.kwargs["status"] == "ok"
|
||
assert call_args.kwargs["duration_ms"] == 12.5
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_log_stage_execution_calls_warn_on_failure(self, fake_logger):
|
||
# Arrange
|
||
pipeline = Pipeline(name="test", stages=[_FakeStage("s1")], logger=fake_logger)
|
||
ctx = _FakeCtx(trace_id="trace-fail")
|
||
err = ValidationError("field", "boom")
|
||
|
||
# Act
|
||
await pipeline._logStageExecution(ctx, "s1", 5.0, False, err)
|
||
|
||
# Assert
|
||
fake_logger.warn.assert_awaited_once()
|
||
call_args = fake_logger.warn.call_args
|
||
assert call_args.kwargs["status"] == "failed"
|
||
assert call_args.kwargs["error_type"] == "ValidationError"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_log_stage_execution_noop_when_logger_is_none(self):
|
||
# Arrange
|
||
pipeline = Pipeline(name="test", stages=[_FakeStage("s1")], logger=None)
|
||
ctx = _FakeCtx()
|
||
|
||
# Act
|
||
await pipeline._logStageExecution(ctx, "s1", 1.0, True, None)
|
||
|
||
# Assert — 无异常即视为通过
|
||
|
||
|
||
# ─── stage-level span（H-4 / §11.2）───────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestStageSpan:
|
||
"""``Pipeline.run`` 为每个 stage 包裹独立 span（H-4）。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_each_stage_generates_independent_span(self):
|
||
# Arrange
|
||
s1 = _FakeStage("session_resolve")
|
||
s2 = _FakeStage("route")
|
||
tracer = _FakeTracer()
|
||
pipeline = Pipeline(name="inbound", stages=[s1, s2], tracer_port=tracer)
|
||
ctx = _FakeCtx(trace_id="trace-1")
|
||
|
||
# Act
|
||
ok, _ = await pipeline.run(ctx)
|
||
|
||
# Assert — 每个 stage 各生成一个 span
|
||
assert ok is True
|
||
assert len(tracer.start_calls) == 2
|
||
assert len(tracer.end_calls) == 2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_span_name_contains_pipeline_name_and_stage_id(self):
|
||
# Arrange
|
||
s1 = _FakeStage("session_resolve")
|
||
tracer = _FakeTracer()
|
||
pipeline = Pipeline(name="inbound", stages=[s1], tracer_port=tracer)
|
||
|
||
# Act
|
||
await pipeline.run(_FakeCtx(trace_id="trace-2"))
|
||
|
||
# Assert — span 名格式 {pipeline_name}.{stage_id}
|
||
assert tracer.start_calls[0] == ("inbound.session_resolve", "trace-2")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_span_status_ok_on_success(self):
|
||
# Arrange
|
||
s1 = _FakeStage("deliver")
|
||
tracer = _FakeTracer()
|
||
pipeline = Pipeline(name="outbound", stages=[s1], tracer_port=tracer)
|
||
|
||
# Act
|
||
await pipeline.run(_FakeCtx())
|
||
|
||
# Assert
|
||
assert tracer.end_calls[0] == ("outbound.deliver", "ok")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_span_closed_with_error_status_on_exception(self):
|
||
# Arrange — stage 抛原生异常，span 仍需在 finally 中关闭
|
||
s1 = _FakeStage(
|
||
"deliver",
|
||
raise_error=RuntimeError("boom"),
|
||
failure=FailureStrategy.TERMINATE,
|
||
)
|
||
tracer = _FakeTracer()
|
||
pipeline = Pipeline(name="outbound", stages=[s1], tracer_port=tracer)
|
||
|
||
# Act
|
||
ok, err = await pipeline.run(_FakeCtx(trace_id="trace-err"))
|
||
|
||
# Assert — span 被关闭（try/finally），状态为 error
|
||
assert ok is False
|
||
assert err is not None
|
||
assert len(tracer.end_calls) == 1
|
||
assert tracer.end_calls[0] == ("outbound.deliver", "error")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_span_closed_with_error_status_when_stage_returns_false(self):
|
||
# Arrange — stage 返回 False（非异常失败），span 状态为 error
|
||
s1 = _FakeStage("dispatch", result=False, failure=FailureStrategy.TERMINATE)
|
||
tracer = _FakeTracer()
|
||
pipeline = Pipeline(name="control-plane", stages=[s1], tracer_port=tracer)
|
||
|
||
# Act
|
||
await pipeline.run(_FakeCtx())
|
||
|
||
# Assert
|
||
assert tracer.end_calls[0] == ("control-plane.dispatch", "error")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_span_when_tracer_port_is_none(self):
|
||
# Arrange — tracer_port=None（默认），向后兼容不生成 span
|
||
s1 = _FakeStage("resolve")
|
||
pipeline = Pipeline(name="inbound", stages=[s1])
|
||
|
||
# Act
|
||
ok, _ = await pipeline.run(_FakeCtx())
|
||
|
||
# Assert — 无异常即通过（tracer_port 为 None 时跳过 span 包裹）
|
||
assert ok is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_condition_skipped_stage_has_no_span(self):
|
||
# Arrange — condition 返回 False 的 stage 不生成 span
|
||
s1 = _FakeStage("skipped", condition=lambda _ctx: False)
|
||
s2 = _FakeStage("route")
|
||
tracer = _FakeTracer()
|
||
pipeline = Pipeline(name="inbound", stages=[s1, s2], tracer_port=tracer)
|
||
|
||
# Act
|
||
await pipeline.run(_FakeCtx())
|
||
|
||
# Assert — 仅执行 stage 生成 span
|
||
assert len(tracer.start_calls) == 1
|
||
assert tracer.start_calls[0][0] == "inbound.route"
|