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

1""" 

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

3 

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

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

6""" 

7 

8import json 

9import os 

10import tempfile 

11from contextlib import contextmanager 

12from dataclasses import dataclass, field 

13from pathlib import Path 

14from typing import Any, Dict, List, Optional 

15from unittest.mock import MagicMock, patch 

16 

17 

18# ─── Mock LLM ─────────────────────────────────────────────── 

19 

20@dataclass 

21class MockLLMResponse: 

22 """Mock LLM 响应。""" 

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

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

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

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

27 }) 

28 finish_reason: str = "stop" 

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

30 

31 

32class MockLLMClient: 

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

34 

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

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

37 self._idx = 0 

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

39 

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

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

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

43 self._idx += 1 

44 return resp 

45 

46 def reset(self): 

47 self._idx = 0 

48 self.calls.clear() 

49 

50 

51# ─── Fixture 工厂 ──────────────────────────────────────────── 

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: Optional[Dict] = 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: Optional[Dict] = 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 shutil.rmtree(d, ignore_errors=True) 

102 

103 

104def mock_memory_store(): 

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

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

107 return store 

108 

109 

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

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

112 return { 

113 "state": state, 

114 "iteration": 0, 

115 "total_tokens": 0, 

116 "total_cost": 0.0, 

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

118 "history": [], 

119 } 

120 

121 

122def sample_audit_report(): 

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

124 return { 

125 "findings": [ 

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

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

128 ], 

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

130 "score": 85, 

131 } 

132 

133 

134def sample_health_status(healthy: bool = True): 

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

136 return { 

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

138 "checks": [ 

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

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

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

142 ], 

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

144 } 

145 

146 

147def sample_docker_config(): 

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

149 return { 

150 "image": "agentos:latest", 

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

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

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

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

155 } 

156 

157 

158def sample_middleware_stack(): 

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

160 return { 

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

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

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

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

165 } 

166 

167 

168def sample_alert_config(): 

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

170 return { 

171 "rules": [ 

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

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

174 ], 

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

176 } 

177 

178 

179# ─── 辅助 ──────────────────────────────────────────────────── 

180 

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

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

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

184 _deep_merge(base[k], v) 

185 else: 

186 base[k] = v