Kris
08617091dc
- 新增多个业务域的__init__.py模块文件，规范包导出结构 - 调整多个DTO文件的导入路径，统一模块组织方式 - 移除测试文件中多余的空行与导入语句 - 优化部分业务模块的包层级划分
156 lines
4.9 KiB
Python
156 lines
4.9 KiB
Python
"""external_systems 单元测试共享 fixture。
|
||
|
||
本文件仅放置跨多个测试文件复用的 fixture 与工厂函数。单文件独有的 helper
|
||
应保留在对应测试文件内，避免过度集中（见 testing-guidelines.md）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
from yuxi.external_systems.core.models import (
|
||
Environment,
|
||
ExternalSystem,
|
||
ExternalTool,
|
||
)
|
||
|
||
# ─── 工厂函数 ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.fixture
|
||
def make_external_tool() -> type[ExternalTool]:
|
||
"""返回 ExternalTool 工厂类型，调用时按需覆盖字段。
|
||
|
||
用法::
|
||
|
||
tool = make_external_tool(slug="my-tool", adapter_type="http")
|
||
"""
|
||
|
||
def _build(**overrides: Any) -> ExternalTool:
|
||
defaults: dict[str, Any] = {
|
||
"slug": "test-tool",
|
||
"name": "Test Tool",
|
||
"description": "A tool for testing",
|
||
"adapter_type": "http",
|
||
"auth_type": "none",
|
||
"timeout": 30,
|
||
"enabled": True,
|
||
}
|
||
defaults.update(overrides)
|
||
return ExternalTool(**defaults)
|
||
|
||
return _build
|
||
|
||
|
||
@pytest.fixture
|
||
def make_external_system() -> type[ExternalSystem]:
|
||
"""返回 ExternalSystem 工厂类型，调用时按需覆盖字段。"""
|
||
|
||
def _build(**overrides: Any) -> ExternalSystem:
|
||
defaults: dict[str, Any] = {
|
||
"slug": "test-system",
|
||
"name": "Test System",
|
||
"description": "A system for testing",
|
||
"adapter_type": "http",
|
||
"auth_type": "none",
|
||
"enabled": True,
|
||
}
|
||
defaults.update(overrides)
|
||
return ExternalSystem(**defaults)
|
||
|
||
return _build
|
||
|
||
|
||
@pytest.fixture
|
||
def make_environment() -> type[Environment]:
|
||
"""返回 Environment 工厂类型。"""
|
||
|
||
def _build(**overrides: Any) -> Environment:
|
||
defaults: dict[str, Any] = {
|
||
"system_id": 1,
|
||
"env_key": "default",
|
||
"name": "Default Env",
|
||
"is_default": False,
|
||
"enabled": True,
|
||
}
|
||
defaults.update(overrides)
|
||
return Environment(**defaults)
|
||
|
||
return _build
|
||
|
||
|
||
# ─── Stub 端口实现 ─────────────────────────────────────────────────────────
|
||
|
||
|
||
class StubCryptoService:
|
||
"""内存版 CryptoService 桩，用于 webhook_verifier / secret_rotation_scheduler 等测试。
|
||
|
||
- encrypt_text / decrypt_text 互为逆操作（基于 base64）。
|
||
- mask 系列方法返回固定掩码占位。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._store: dict[str, str] = {}
|
||
|
||
def encrypt_text(self, plaintext: str) -> str:
|
||
import base64
|
||
|
||
encoded = base64.b64encode(plaintext.encode()).decode()
|
||
self._store[encoded] = plaintext
|
||
return encoded
|
||
|
||
def decrypt_text(self, ciphertext: str) -> str:
|
||
import base64
|
||
|
||
if ciphertext in self._store:
|
||
return self._store[ciphertext]
|
||
return base64.b64decode(ciphertext.encode()).decode()
|
||
|
||
def encrypt_sensitive_fields(self, data: dict[str, Any], sensitive_keys: set[str]) -> dict[str, Any]:
|
||
result = dict(data)
|
||
for key in sensitive_keys:
|
||
if key in result and isinstance(result[key], str):
|
||
result[key] = self.encrypt_text(result[key])
|
||
return result
|
||
|
||
def decrypt_sensitive_fields(self, data: dict[str, Any], sensitive_keys: set[str]) -> dict[str, Any]:
|
||
result = dict(data)
|
||
for key in sensitive_keys:
|
||
if key in result and isinstance(result[key], str):
|
||
result[key] = self.decrypt_text(result[key])
|
||
return result
|
||
|
||
def mask_sensitive_fields(self, data: dict[str, Any], sensitive_keys: set[str]) -> dict[str, Any]:
|
||
result = dict(data)
|
||
for key in sensitive_keys:
|
||
if key in result:
|
||
result[key] = "***"
|
||
return result
|
||
|
||
def mask_all_string_values(self, data: dict[str, Any]) -> dict[str, Any]:
|
||
return {k: "***" if isinstance(v, str) else v for k, v in data.items()}
|
||
|
||
@staticmethod
|
||
def is_sensitive_key(key: str) -> bool:
|
||
return key.lower() in {"password", "secret", "token", "api_key"}
|
||
|
||
@staticmethod
|
||
def sensitive_keywords() -> list[str]:
|
||
return ["password", "secret", "token", "api_key"]
|
||
|
||
|
||
@pytest.fixture
|
||
def stub_crypto() -> StubCryptoService:
|
||
"""提供内存版 CryptoService 桩。"""
|
||
return StubCryptoService()
|
||
|
||
|
||
@pytest.fixture
|
||
def stub_secret_resolver() -> AsyncMock:
|
||
"""提供 SecretResolver 异步桩，resolve 返回原始引用值。"""
|
||
resolver = AsyncMock()
|
||
resolver.resolve = AsyncMock(side_effect=lambda ref: ref)
|
||
return resolver
|