Kris
3ae9c0bd21
1. 删除了 test/unit/external_systems/framework/ 下的废弃空测试目录 2. 修复多处测试断言逻辑、参数传递与测试数据构造 3. 新增日志级别、缓存令牌、流事件等DTO单元测试 4. 补充路由绑定、会话仓储、outbox仓储的测试覆盖 5. 更新测试用例中的异常类型、参数校验与业务逻辑断言
155 lines
4.4 KiB
Python
155 lines
4.4 KiB
Python
"""admin.py DTO 单元测试。
|
|
|
|
覆盖 ``AdminSendCmd`` 与 ``AdminSendResult`` 的字段赋值、默认值、不可变
|
|
语义与 ``__post_init__`` 校验逻辑。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
|
|
import pytest
|
|
from yuxi.channels.contract.dtos.admin import (
|
|
AdminSendCmd,
|
|
AdminSendResult,
|
|
)
|
|
from yuxi.channels.contract.dtos.common import (
|
|
FailureDetail,
|
|
MessageContent,
|
|
Operator,
|
|
OperatorRole,
|
|
SkipDetail,
|
|
)
|
|
from yuxi.channels.contract.errors import ValidationError
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
def _make_operator() -> Operator:
|
|
return Operator(user_id="admin-1", role=OperatorRole.ADMIN_USER)
|
|
|
|
|
|
def _make_content() -> MessageContent:
|
|
return MessageContent(text="hello")
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestAdminSendCmd:
|
|
"""管理员发送命令 DTO 测试。"""
|
|
|
|
def test_fields_are_assigned(self):
|
|
# Arrange
|
|
operator = _make_operator()
|
|
content = _make_content()
|
|
# Act
|
|
cmd = AdminSendCmd(
|
|
target="sess-1",
|
|
content=content,
|
|
conversation_policy="reuse",
|
|
idempotency_key="idem-1",
|
|
operator=operator,
|
|
)
|
|
# Assert
|
|
assert cmd.target == "sess-1"
|
|
assert cmd.content is content
|
|
assert cmd.conversation_policy == "reuse"
|
|
assert cmd.idempotency_key == "idem-1"
|
|
assert cmd.operator is operator
|
|
assert cmd.target_type == "session_id"
|
|
|
|
def test_target_type_peer_id(self):
|
|
# Arrange
|
|
operator = _make_operator()
|
|
content = _make_content()
|
|
# Act
|
|
cmd = AdminSendCmd(
|
|
target="peer-1",
|
|
content=content,
|
|
conversation_policy="new",
|
|
idempotency_key="idem-1",
|
|
operator=operator,
|
|
target_type="peer_id",
|
|
)
|
|
# Assert
|
|
assert cmd.target_type == "peer_id"
|
|
|
|
def test_is_frozen(self):
|
|
# Arrange
|
|
operator = _make_operator()
|
|
content = _make_content()
|
|
cmd = AdminSendCmd(
|
|
target="sess-1",
|
|
content=content,
|
|
conversation_policy="reuse",
|
|
idempotency_key="idem-1",
|
|
operator=operator,
|
|
)
|
|
# Act / Assert
|
|
with pytest.raises(dataclasses.FrozenInstanceError):
|
|
cmd.target = "other" # type: ignore[misc]
|
|
|
|
def test_empty_target_raises_validation_error(self):
|
|
# Arrange
|
|
operator = _make_operator()
|
|
content = _make_content()
|
|
# Act / Assert
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
AdminSendCmd(
|
|
target="",
|
|
content=content,
|
|
conversation_policy="reuse",
|
|
idempotency_key="idem-1",
|
|
operator=operator,
|
|
)
|
|
assert exc_info.value.field == "target"
|
|
|
|
def test_empty_idempotency_key_raises_validation_error(self):
|
|
# Arrange
|
|
operator = _make_operator()
|
|
content = _make_content()
|
|
# Act / Assert
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
AdminSendCmd(
|
|
target="sess-1",
|
|
content=content,
|
|
conversation_policy="reuse",
|
|
idempotency_key="",
|
|
operator=operator,
|
|
)
|
|
assert exc_info.value.field == "idempotency_key"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestAdminSendResult:
|
|
"""管理员发送结果 DTO 测试。"""
|
|
|
|
def test_required_fields_are_assigned(self):
|
|
# Arrange
|
|
# Act
|
|
result = AdminSendResult(message_ids=("msg-1", "msg-2"))
|
|
# Assert
|
|
assert result.message_ids == ("msg-1", "msg-2")
|
|
assert result.failures == ()
|
|
assert result.skipped == ()
|
|
|
|
def test_with_failures_and_skipped(self):
|
|
# Arrange
|
|
failures = (FailureDetail(target="peer-1", error_code="ERR", message="fail"),)
|
|
skipped = (SkipDetail(target="peer-2", reason="in_denylist", policy="deny"),)
|
|
# Act
|
|
result = AdminSendResult(
|
|
message_ids=("msg-1",),
|
|
failures=failures,
|
|
skipped=skipped,
|
|
)
|
|
# Assert
|
|
assert result.failures == failures
|
|
assert result.skipped == skipped
|
|
|
|
def test_is_frozen(self):
|
|
# Arrange
|
|
result = AdminSendResult(message_ids=("msg-1",))
|
|
# Act / Assert
|
|
with pytest.raises(dataclasses.FrozenInstanceError):
|
|
result.message_ids = () # type: ignore[misc]
|