Coverage for agentos/testing/fixtures.py: 45%

1""" 

2AgentOS v0.95 Testing Fixtures — 可复用测试基础设施。 

3 

4提供 mock 对象工厂、预设配置 fixtures、临时文件上下文， 

5供单元测试和集成测试共用。 

6""" 

7 

8import tempfile 

9from contextlib import contextmanager 

10from dataclasses import dataclass, field 

11from pathlib import Path 

12from typing import Any 

13from unittest.mock import MagicMock 

14 

15# ─── Mock LLM ─────────────────────────────────────────────── 

16 

17 

18@dataclass 

19class MockLLMResponse: 

20 """Mock LLM 响应。""" 

21 

22 content: str = "This is a mock LLM response." 

23 model: str = "mock-gpt-4" 

24 usage: dict[str, int] = field( 

25 default_factory=lambda: {"prompt_tokens": 50, "completion_tokens": 30, "total_tokens": 80} 

26 ) 

27 finish_reason: str = "stop" 

28 tool_calls: list[dict] | None = None 

29 

30 

31class MockLLMClient: 

32 """可配置的 Mock LLM 客户端，支持预设响应序列和工具调用。""" 

33 

34 def __init__(self, responses: list[MockLLMResponse] | None = None): 

35 self.responses = responses or [MockLLMResponse()] 

36 self._idx = 0 

37 self.calls: list[dict] = [] 

38 

39 async def chat(self, messages: list[dict], **kwargs) -> MockLLMResponse: 

40 self.calls.append({"messages": messages, "kwargs": kwargs}) 

41 resp = self.responses[min(self._idx, len(self.responses) - 1)] 

42 self._idx += 1 

43 return resp 

44 

45 def reset(self): 

46 self._idx = 0 

47 self.calls.clear() 

48 

49 

50# ─── Fixture 工厂 ──────────────────────────────────────────── 

51 

52 

53def mock_openai_client(): 

54 """创建一个完整的 mock OpenAI client。""" 

55 client = MagicMock() 

56 client.chat.completions.create.return_value = MagicMock( 

57 choices=[MagicMock(message=MagicMock(content="mock response"))], 

58 model="mock-gpt-4", 

59 usage=MagicMock(prompt_tokens=10, completion_tokens=5, total_tokens=15), 

60 ) 

61 return client 

62 

63 

64def mock_model_response(content: str = "ok", model: str = "mock-model"): 

65 return MockLLMResponse(content=content, model=model) 

66 

67 

68def sample_config(overrides: dict | None = None) -> dict[str, Any]: 

69 """返回一份可用于测试的完整 AgentOSConfig 字典。""" 

70 base = { 

71 "models": { 

72 "default": {"provider": "openai", "model": "gpt-4o-mini", "temperature": 0.7}, 

73 "fast": {"provider": "openai", "model": "gpt-4o-mini", "temperature": 0.3}, 

74 }, 

75 "loop": {"max_iterations": 10, "timeout_seconds": 30}, 

76 "memory": {"backend": "short_term", "max_tokens": 8000}, 

77 "security": {"guardrails_enabled": True, "pii_sanitize": True}, 

78 "observability": {"metrics_enabled": False, "tracing_enabled": False}, 

79 } 

80 if overrides: 

81 _deep_merge(base, overrides) 

82 return base 

83 

84 

85def sample_loop_config(overrides: dict | None = None) -> dict[str, Any]: 

86 """返回 LoopConfig 字典。""" 

87 base = {"max_iterations": 5, "timeout_seconds": 15, "reflection_enabled": True} 

88 if overrides: 

89 base.update(overrides) 

90 return base 

91 

92 

93@contextmanager 

94def temp_workspace(suffix: str = ""): 

95 """创建临时工作目录，yield Path 对象，退出时清理。""" 

96 d = tempfile.mkdtemp(suffix=f"_agentos_test{suffix}") 

97 try: 

98 yield Path(d) 

99 finally: 

100 import shutil 

101 

102 shutil.rmtree(d, ignore_errors=True) 

103 

104 

105def mock_memory_store(): 

106 """返回一个 dict-backed 模拟 memory store。""" 

107 store = {"messages": [], "summary": "", "entities": {}} 

108 return store 

109 

110 

111def sample_agent_state(state: str = "idle", context: dict | None = None): 

112 """返回一份预设的 AgentState 字典。""" 

113 return { 

114 "state": state, 

115 "iteration": 0, 

116 "total_tokens": 0, 

117 "total_cost": 0.0, 

118 "context": context or {"task": "test task"}, 

119 "history": [], 

120 } 

121 

122 

123def sample_audit_report(): 

124 """返回一份预设的 AuditReport 字典。""" 

125 return { 

126 "findings": [ 

127 { 

128 "severity": "low", 

129 "category": "code_injection", 

130 "description": "eval() usage detected", 

131 "location": "test.py:42", 

132 }, 

133 { 

134 "severity": "info", 

135 "category": "best_practice", 

136 "description": "hardcoded secret pattern", 

137 "location": "config.py:11", 

138 }, 

139 ], 

140 "summary": {"critical": 0, "high": 0, "medium": 0, "low": 1, "info": 1}, 

141 "score": 85, 

142 } 

143 

144 

145def sample_health_status(healthy: bool = True): 

146 """返回一份预设的 HealthStatus 字典。""" 

147 return { 

148 "status": "healthy" if healthy else "degraded", 

149 "checks": [ 

150 {"name": "openai_connectivity", "pass": True, "latency_ms": 120}, 

151 {"name": "disk_space", "pass": True, "free_gb": 42.0}, 

152 {"name": "memory", "pass": True, "used_percent": 35.0}, 

153 ], 

154 "timestamp": "2025-01-01T00:00:00Z", 

155 } 

156 

157 

158def sample_docker_config(): 

159 """返回一份预设的 DockerConfig 字典。""" 

160 return { 

161 "image": "agentos:latest", 

162 "ports": {"8000/tcp": 8000}, 

163 "volumes": {"./data": "/app/data"}, 

164 "environment": {"LOG_LEVEL": "INFO"}, 

165 "healthcheck": {"test": "curl -f localhost:8000/health", "interval": "30s"}, 

166 } 

167 

168 

169def sample_middleware_stack(): 

170 """返回一份预设的 MiddlewareStack 配置字典。""" 

171 return { 

172 "cors": {"allowed_origins": ["*"], "allowed_methods": ["GET", "POST"]}, 

173 "auth": {"enabled": True, "token_header": "X-API-Key"}, 

174 "request_id": {"enabled": True, "header_name": "X-Request-ID"}, 

175 "request_log": {"enabled": True, "log_body": False}, 

176 } 

177 

178 

179def sample_alert_config(): 

180 """返回一份预设的 AlertConfig 字典。""" 

181 return { 

182 "rules": [ 

183 {"name": "high_latency", "condition": "latency_p95 > 5000", "severity": "warning"}, 

184 {"name": "error_rate", "condition": "error_rate > 0.05", "severity": "critical"}, 

185 ], 

186 "webhooks": [{"url": "https://hooks.slack.com/test", "channel": "#alerts"}], 

187 } 

188 

189 

190# ─── 辅助 ──────────────────────────────────────────────────── 

191 

192 

193def _deep_merge(base: dict, override: dict): 

194 for k, v in override.items(): 

195 if isinstance(v, dict) and isinstance(base.get(k), dict): 

196 _deep_merge(base[k], v) 

197 else: 

198 base[k] = v