forked from HKUDS/OpenSpace
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_llm_factory.py
More file actions
217 lines (163 loc) · 7.48 KB
/
Copy pathtest_llm_factory.py
File metadata and controls
217 lines (163 loc) · 7.48 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
"""Tests for LLMFactory — extracted LLM client creation from OpenSpace."""
from __future__ import annotations
import pytest
from unittest.mock import MagicMock, patch, call
try:
from openspace.tool_layer import OpenSpace, OpenSpaceConfig
from openspace.llm_factory import LLMFactory
_HAS_TOOL_LAYER = True
except Exception:
_HAS_TOOL_LAYER = False
pytestmark = pytest.mark.skipif(not _HAS_TOOL_LAYER, reason="tool_layer deps unavailable")
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def config():
return OpenSpaceConfig(
llm_model="openrouter/anthropic/claude-sonnet-4.5",
llm_enable_thinking=True,
llm_timeout=60.0,
llm_max_retries=5,
llm_rate_limit_delay=0.5,
llm_kwargs={"api_key": "sk-test"},
tool_retrieval_model="openrouter/openai/gpt-4o",
)
@pytest.fixture
def minimal_config():
return OpenSpaceConfig()
# ---------------------------------------------------------------------------
# LLMFactory.__init__
# ---------------------------------------------------------------------------
class TestLLMFactoryInit:
def test_initial_state(self, config):
factory = LLMFactory(config=config)
assert factory.llm_client is None
assert factory.tool_retrieval_llm is None
def test_stores_config(self, config):
factory = LLMFactory(config=config)
assert factory._config is config
# ---------------------------------------------------------------------------
# LLMFactory.create_main()
# ---------------------------------------------------------------------------
class TestCreateMain:
def test_creates_llm_client(self, config):
with patch("openspace.llm_factory.LLMClient") as MockLLM:
mock_client = MagicMock()
MockLLM.return_value = mock_client
factory = LLMFactory(config=config)
result = factory.create_main()
assert result is mock_client
assert factory.llm_client is mock_client
def test_passes_all_config_fields(self, config):
with patch("openspace.llm_factory.LLMClient") as MockLLM:
MockLLM.return_value = MagicMock()
factory = LLMFactory(config=config)
factory.create_main()
MockLLM.assert_called_once_with(
model="openrouter/anthropic/claude-sonnet-4.5",
enable_thinking=True,
rate_limit_delay=0.5,
max_retries=5,
timeout=60.0,
api_key="sk-test",
)
def test_uses_defaults_when_minimal_config(self, minimal_config):
with patch("openspace.llm_factory.LLMClient") as MockLLM:
MockLLM.return_value = MagicMock()
factory = LLMFactory(config=minimal_config)
factory.create_main()
MockLLM.assert_called_once_with(
model="openrouter/anthropic/claude-sonnet-4.5",
enable_thinking=False,
rate_limit_delay=0.0,
max_retries=3,
timeout=120.0,
)
def test_create_main_twice_replaces_client(self, config):
with patch("openspace.llm_factory.LLMClient") as MockLLM:
first = MagicMock()
second = MagicMock()
MockLLM.side_effect = [first, second]
factory = LLMFactory(config=config)
factory.create_main()
assert factory.llm_client is first
factory.create_main()
assert factory.llm_client is second
# ---------------------------------------------------------------------------
# LLMFactory.create_tool_retrieval()
# ---------------------------------------------------------------------------
class TestCreateToolRetrieval:
def test_creates_when_model_configured(self, config):
with patch("openspace.llm_factory.LLMClient") as MockLLM:
mock_client = MagicMock()
MockLLM.return_value = mock_client
factory = LLMFactory(config=config)
result = factory.create_tool_retrieval()
assert result is mock_client
assert factory.tool_retrieval_llm is mock_client
def test_returns_none_when_no_model(self, minimal_config):
factory = LLMFactory(config=minimal_config)
result = factory.create_tool_retrieval()
assert result is None
assert factory.tool_retrieval_llm is None
def test_passes_correct_config(self, config):
with patch("openspace.llm_factory.LLMClient") as MockLLM:
MockLLM.return_value = MagicMock()
factory = LLMFactory(config=config)
factory.create_tool_retrieval()
MockLLM.assert_called_once_with(
model="openrouter/openai/gpt-4o",
timeout=60.0,
max_retries=5,
api_key="sk-test",
)
def test_inherits_llm_kwargs(self, config):
"""Tool retrieval LLM inherits credentials from llm_kwargs."""
config.llm_kwargs = {"api_key": "sk-shared", "api_base": "https://custom"}
with patch("openspace.llm_factory.LLMClient") as MockLLM:
MockLLM.return_value = MagicMock()
factory = LLMFactory(config=config)
factory.create_tool_retrieval()
kwargs = MockLLM.call_args.kwargs
assert kwargs["api_key"] == "sk-shared"
assert kwargs["api_base"] == "https://custom"
def test_returns_none_when_empty_string_model(self, config):
"""Empty string model treated as unconfigured."""
config.tool_retrieval_model = ""
factory = LLMFactory(config=config)
result = factory.create_tool_retrieval()
assert result is None
assert factory.tool_retrieval_llm is None
def test_create_tool_retrieval_twice_replaces(self, config):
"""Second call replaces the tool retrieval client."""
with patch("openspace.llm_factory.LLMClient") as MockLLM:
first, second = MagicMock(), MagicMock()
MockLLM.side_effect = [first, second]
factory = LLMFactory(config=config)
factory.create_tool_retrieval()
assert factory.tool_retrieval_llm is first
factory.create_tool_retrieval()
assert factory.tool_retrieval_llm is second
# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------
class TestErrorHandling:
def test_create_main_exception_propagates(self, config):
"""LLMClient constructor failure propagates, client stays None."""
with patch("openspace.llm_factory.LLMClient", side_effect=RuntimeError("boom")):
factory = LLMFactory(config=config)
with pytest.raises(RuntimeError, match="boom"):
factory.create_main()
assert factory.llm_client is None
# ---------------------------------------------------------------------------
# OpenSpace backward compatibility
# ---------------------------------------------------------------------------
class TestOpenSpaceDelegation:
def test_openspace_has_llm_factory_attr(self):
os_instance = OpenSpace()
assert hasattr(os_instance, "_llm_factory")
def test_llm_client_still_accessible(self):
os_instance = OpenSpace()
assert hasattr(os_instance, "_llm_client")
assert os_instance._llm_client is None