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, Dict, List, Optional 

13from unittest.mock import MagicMock 

14 

15 

16# ─── Mock LLM ─────────────────────────────────────────────── 

17 

18@dataclass 

19class MockLLMResponse: 

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

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

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

23 usage: Dict[str, int] = field(default_factory=lambda: { 

24 "prompt_tokens": 50, "completion_tokens": 30, "total_tokens": 80 

25 }) 

26 finish_reason: str = "stop" 

27 tool_calls: Optional[List[Dict]] = None 

28 

29 

30class MockLLMClient: 

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

32 

33 def __init__(self, responses: Optional[List[MockLLMResponse]] = None): 

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

35 self._idx = 0 

36 self.calls: List[Dict] = [] 

37 

38 async def chat(self, messages: List[Dict], **kwargs) -> MockLLMResponse: 

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

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

41 self._idx += 1 

42 return resp 

43 

44 def reset(self): 

45 self._idx = 0 

46 self.calls.clear() 

47 

48 

49# ─── Fixture 工厂 ──────────────────────────────────────────── 

50 

51def mock_openai_client(): 

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

53 client = MagicMock() 

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

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

56 model="mock-gpt-4", 

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

58 ) 

59 return client 

60 

61 

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

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

64 

65 

66def sample_config(overrides: Optional[Dict] = None) -> Dict[str, Any]: 

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

68 base = { 

69 "models": { 

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

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

72 }, 

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

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

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

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

77 } 

78 if overrides: 

79 _deep_merge(base, overrides) 

80 return base 

81 

82 

83def sample_loop_config(overrides: Optional[Dict] = None) -> Dict[str, Any]: 

84 """返回 LoopConfig 字典。""" 

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

86 if overrides: 

87 base.update(overrides) 

88 return base 

89 

90 

91@contextmanager 

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

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

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

95 try: 

96 yield Path(d) 

97 finally: 

98 import shutil 

99 shutil.rmtree(d, ignore_errors=True) 

100 

101 

102def mock_memory_store(): 

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

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

105 return store 

106 

107 

108def sample_agent_state(state: str = "idle", context: Optional[Dict] = None): 

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

110 return { 

111 "state": state, 

112 "iteration": 0, 

113 "total_tokens": 0, 

114 "total_cost": 0.0, 

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

116 "history": [], 

117 } 

118 

119 

120def sample_audit_report(): 

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

122 return { 

123 "findings": [ 

124 {"severity": "low", "category": "code_injection", "description": "eval() usage detected", "location": "test.py:42"}, 

125 {"severity": "info", "category": "best_practice", "description": "hardcoded secret pattern", "location": "config.py:11"}, 

126 ], 

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

128 "score": 85, 

129 } 

130 

131 

132def sample_health_status(healthy: bool = True): 

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

134 return { 

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

136 "checks": [ 

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

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

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

140 ], 

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

142 } 

143 

144 

145def sample_docker_config(): 

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

147 return { 

148 "image": "agentos:latest", 

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

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

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

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

153 } 

154 

155 

156def sample_middleware_stack(): 

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

158 return { 

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

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

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

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

163 } 

164 

165 

166def sample_alert_config(): 

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

168 return { 

169 "rules": [ 

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

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

172 ], 

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

174 } 

175 

176 

177# ─── 辅助 ──────────────────────────────────────────────────── 

178 

179def _deep_merge(base: Dict, override: Dict): 

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

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

182 _deep_merge(base[k], v) 

183 else: 

184 base[k] = v