Kris
d4770dfba4
1. 修复数据库配置测试中 schema 字段名不匹配问题 2. 替换多处 KeyError 断言为 DomainValidationError 校验 3. 优化 gRPC 测试中可选类型断言逻辑 4. 新增大量重试逻辑、认证校验、资产校验相关测试用例 5. 重构部分测试代码，移除冗余导入与代码 6. 调整 DTO 测试默认值与断言逻辑 7. 格式化部分测试文件代码格式 8. 新增适配器基类默认行为单元测试
391 lines
13 KiB
Python
391 lines
13 KiB
Python
"""integrations/registry.py 厂商集成目录注册表单元测试。
|
||
|
||
覆盖 register / filter_integrations / list_integrations / count_integrations /
|
||
get_integration / get_integration_or_raise / list_adapter_types / clear。
|
||
全局状态（_integrations）通过 _snapshot_registry fixture 在测试后恢复。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from yuxi.external_systems.exceptions import (
|
||
DomainValidationError,
|
||
EntityNotFoundError,
|
||
)
|
||
from yuxi.external_systems.integrations.registry import IntegrationRegistry
|
||
from yuxi.external_systems.integrations.schemas import IntegrationMetadata
|
||
|
||
|
||
# ─── 全局状态快照 fixture ─────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.fixture
|
||
def _snapshot_registry():
|
||
"""保存并恢复 _integrations，避免污染其他测试。"""
|
||
saved = dict(IntegrationRegistry._integrations)
|
||
IntegrationRegistry._integrations.clear()
|
||
try:
|
||
yield
|
||
finally:
|
||
IntegrationRegistry._integrations.clear()
|
||
IntegrationRegistry._integrations.update(saved)
|
||
|
||
|
||
# ─── 辅助构造 ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_metadata(**overrides: Any) -> IntegrationMetadata:
|
||
"""构造测试用 IntegrationMetadata。"""
|
||
base: dict[str, Any] = {
|
||
"key": "test",
|
||
"display_name": "Test Integration",
|
||
"description": "A test integration",
|
||
"adapter_type": "http",
|
||
"source_types": ["test"],
|
||
}
|
||
base.update(overrides)
|
||
return IntegrationMetadata(**base)
|
||
|
||
|
||
# ─── register ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_register_adds_metadata_to_registry(_snapshot_registry):
|
||
# Arrange
|
||
meta = _make_metadata(key="salesforce")
|
||
|
||
# Act
|
||
IntegrationRegistry.register(meta)
|
||
|
||
# Assert
|
||
assert IntegrationRegistry.get_integration("salesforce") is meta
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_register_idempotent_with_same_instance(_snapshot_registry):
|
||
# Arrange
|
||
meta = _make_metadata(key="salesforce")
|
||
IntegrationRegistry.register(meta)
|
||
|
||
# Act - 同一实例重复注册不应抛异常
|
||
IntegrationRegistry.register(meta)
|
||
|
||
# Assert
|
||
assert IntegrationRegistry.get_integration("salesforce") is meta
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_register_conflict_with_different_instance_raises(_snapshot_registry):
|
||
# Arrange
|
||
first = _make_metadata(key="salesforce", display_name="First")
|
||
IntegrationRegistry.register(first)
|
||
second = _make_metadata(key="salesforce", display_name="Second")
|
||
|
||
# Act / Assert - 不同实例重复注册应抛 DomainValidationError
|
||
with pytest.raises(DomainValidationError) as exc_info:
|
||
IntegrationRegistry.register(second)
|
||
assert "salesforce" in str(exc_info.value)
|
||
assert "First" in str(exc_info.value)
|
||
assert "Second" in str(exc_info.value)
|
||
# 原注册项保留
|
||
assert IntegrationRegistry.get_integration("salesforce") is first
|
||
|
||
|
||
# ─── filter_integrations ──────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_filterIntegrations_returns_all_sorted_by_key(_snapshot_registry):
|
||
# Arrange - 故意乱序注册
|
||
IntegrationRegistry.register(_make_metadata(key="zoho"))
|
||
IntegrationRegistry.register(_make_metadata(key="apple"))
|
||
IntegrationRegistry.register(_make_metadata(key="microsoft"))
|
||
|
||
# Act
|
||
result = IntegrationRegistry.filter_integrations()
|
||
|
||
# Assert - 按 key 升序稳定排序
|
||
assert [it.key for it in result] == ["apple", "microsoft", "zoho"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_filterIntegrations_by_adapter_type(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf", adapter_type="http"))
|
||
IntegrationRegistry.register(_make_metadata(key="grpc_pkg", adapter_type="grpc"))
|
||
IntegrationRegistry.register(_make_metadata(key="hub", adapter_type="http"))
|
||
|
||
# Act
|
||
result = IntegrationRegistry.filter_integrations(adapter_type="http")
|
||
|
||
# Assert - 精确匹配 adapter_type
|
||
assert [it.key for it in result] == ["hub", "sf"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_filterIntegrations_by_tags_or_relation(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf", tags=["crm", "sales"]))
|
||
IntegrationRegistry.register(_make_metadata(key="slack", tags=["messaging"]))
|
||
IntegrationRegistry.register(_make_metadata(key="hub", tags=["crm", "marketing"]))
|
||
|
||
# Act - 命中任一标签即返回（OR 关系）
|
||
result = IntegrationRegistry.filter_integrations(tags=["sales", "messaging"])
|
||
|
||
# Assert
|
||
assert [it.key for it in result] == ["sf", "slack"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_filterIntegrations_by_keyword_case_insensitive(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf", display_name="Salesforce"))
|
||
IntegrationRegistry.register(_make_metadata(key="hub", display_name="HubSpot"))
|
||
|
||
# Act - 大小写不敏感匹配 display_name
|
||
result = IntegrationRegistry.filter_integrations(keyword="SALES")
|
||
|
||
# Assert
|
||
assert [it.key for it in result] == ["sf"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_filterIntegrations_by_keyword_matches_description(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf", display_name="SF", description="CRM platform"))
|
||
IntegrationRegistry.register(_make_metadata(key="slack", display_name="Slack", description="Chat tool"))
|
||
|
||
# Act - 关键词匹配 description
|
||
result = IntegrationRegistry.filter_integrations(keyword="crm")
|
||
|
||
# Assert
|
||
assert [it.key for it in result] == ["sf"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_filterIntegrations_combined_filters(_snapshot_registry):
|
||
# Arrange - sf/hub 同时满足三条件，slack 仅满足 adapter_type+keyword 不满足 tags
|
||
IntegrationRegistry.register(
|
||
_make_metadata(key="sf", adapter_type="http", tags=["crm"], display_name="Salesforce CRM")
|
||
)
|
||
IntegrationRegistry.register(
|
||
_make_metadata(key="hub", adapter_type="http", tags=["crm"], display_name="HubSpot CRM")
|
||
)
|
||
IntegrationRegistry.register(
|
||
_make_metadata(key="slack", adapter_type="http", tags=["messaging"], display_name="Slack CRM")
|
||
)
|
||
|
||
# Act - adapter_type + tags + keyword 组合（AND 关系）
|
||
result = IntegrationRegistry.filter_integrations(adapter_type="http", tags=["crm"], keyword="crm")
|
||
|
||
# Assert - 同时满足所有条件：sf/hub 命中，slack 因 tags 不匹配被排除
|
||
assert [it.key for it in result] == ["hub", "sf"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_filterIntegrations_empty_when_no_match(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf"))
|
||
|
||
# Act
|
||
result = IntegrationRegistry.filter_integrations(keyword="nonexistent")
|
||
|
||
# Assert
|
||
assert result == []
|
||
|
||
|
||
# ─── list_integrations ────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_listIntegrations_paginates_results(_snapshot_registry):
|
||
# Arrange
|
||
for k in ("a", "b", "c", "d", "e"):
|
||
IntegrationRegistry.register(_make_metadata(key=k))
|
||
|
||
# Act - 第 2 页，每页 2 条
|
||
result = IntegrationRegistry.list_integrations(limit=2, offset=2)
|
||
|
||
# Assert
|
||
assert [it.key for it in result] == ["c", "d"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_listIntegrations_offset_negative_clamped_to_zero(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf"))
|
||
|
||
# Act - offset 为负数应归零
|
||
result = IntegrationRegistry.list_integrations(limit=10, offset=-5)
|
||
|
||
# Assert
|
||
assert [it.key for it in result] == ["sf"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_listIntegrations_limit_negative_clamped_to_zero(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf"))
|
||
|
||
# Act - limit 为负数应归零，返回空列表
|
||
result = IntegrationRegistry.list_integrations(limit=-1, offset=0)
|
||
|
||
# Assert
|
||
assert result == []
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_listIntegrations_offset_beyond_length_returns_empty(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf"))
|
||
|
||
# Act - offset 超过总数
|
||
result = IntegrationRegistry.list_integrations(limit=10, offset=100)
|
||
|
||
# Assert
|
||
assert result == []
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_listIntegrations_defaults_limit50_offset0(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf"))
|
||
|
||
# Act - 不传分页参数
|
||
result = IntegrationRegistry.list_integrations()
|
||
|
||
# Assert
|
||
assert [it.key for it in result] == ["sf"]
|
||
|
||
|
||
# ─── count_integrations ───────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_countIntegrations_returns_total(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf"))
|
||
IntegrationRegistry.register(_make_metadata(key="hub"))
|
||
|
||
# Act
|
||
total = IntegrationRegistry.count_integrations()
|
||
|
||
# Assert
|
||
assert total == 2
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_countIntegrations_respects_filters(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf", adapter_type="http"))
|
||
IntegrationRegistry.register(_make_metadata(key="grpc_pkg", adapter_type="grpc"))
|
||
|
||
# Act
|
||
total = IntegrationRegistry.count_integrations(adapter_type="http")
|
||
|
||
# Assert
|
||
assert total == 1
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_countIntegrations_zero_when_no_match(_snapshot_registry):
|
||
# Arrange / Act / Assert
|
||
assert IntegrationRegistry.count_integrations(keyword="nonexistent") == 0
|
||
|
||
|
||
# ─── get_integration ─────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_getIntegration_returns_metadata(_snapshot_registry):
|
||
# Arrange
|
||
meta = _make_metadata(key="salesforce")
|
||
IntegrationRegistry.register(meta)
|
||
|
||
# Act
|
||
result = IntegrationRegistry.get_integration("salesforce")
|
||
|
||
# Assert
|
||
assert result is meta
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_getIntegration_unknown_returns_none(_snapshot_registry):
|
||
# Arrange / Act / Assert
|
||
assert IntegrationRegistry.get_integration("nonexistent") is None
|
||
|
||
|
||
# ─── get_integration_or_raise ──────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_getIntegrationOrRaise_returns_metadata(_snapshot_registry):
|
||
# Arrange
|
||
meta = _make_metadata(key="salesforce")
|
||
IntegrationRegistry.register(meta)
|
||
|
||
# Act
|
||
result = IntegrationRegistry.get_integration_or_raise("salesforce")
|
||
|
||
# Assert
|
||
assert result is meta
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_getIntegrationOrRaise_unknown_raises_entityNotFoundError(_snapshot_registry):
|
||
# Arrange / Act / Assert
|
||
with pytest.raises(EntityNotFoundError) as exc_info:
|
||
IntegrationRegistry.get_integration_or_raise("nonexistent")
|
||
assert "nonexistent" in str(exc_info.value)
|
||
|
||
|
||
# ─── list_adapter_types ───────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_listAdapterTypes_returns_dedup_types(_snapshot_registry):
|
||
# Arrange - 多个集成共用 adapter_type
|
||
IntegrationRegistry.register(_make_metadata(key="sf", adapter_type="http"))
|
||
IntegrationRegistry.register(_make_metadata(key="hub", adapter_type="http"))
|
||
IntegrationRegistry.register(_make_metadata(key="grpc_pkg", adapter_type="grpc"))
|
||
|
||
# Act
|
||
types = IntegrationRegistry.list_adapter_types()
|
||
|
||
# Assert - 去重后按字母升序排序返回
|
||
assert types == ["grpc", "http"]
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_listAdapterTypes_empty_when_nothing_registered(_snapshot_registry):
|
||
# Arrange / Act / Assert
|
||
assert IntegrationRegistry.list_adapter_types() == []
|
||
|
||
|
||
# ─── clear ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_clear_empties_integrations(_snapshot_registry):
|
||
# Arrange
|
||
IntegrationRegistry.register(_make_metadata(key="sf"))
|
||
|
||
# Act
|
||
IntegrationRegistry.clear()
|
||
|
||
# Assert
|
||
assert IntegrationRegistry.list_adapter_types() == []
|
||
assert IntegrationRegistry.get_integration("sf") is None
|
||
assert IntegrationRegistry.count_integrations() == 0
|
||
|
||
|
||
@pytest.mark.unit
|
||
def test_clear_safe_when_already_empty(_snapshot_registry):
|
||
# Arrange / Act / Assert - 空注册表清空不抛异常
|
||
IntegrationRegistry.clear()
|
||
assert IntegrationRegistry.count_integrations() == 0
|