-
Notifications
You must be signed in to change notification settings - Fork 266
Expand file tree
/
Copy pathtest_ssl_config.py
More file actions
370 lines (289 loc) · 14.9 KB
/
test_ssl_config.py
File metadata and controls
370 lines (289 loc) · 14.9 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
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
"""
Unit tests for SSL configuration support in optillm.
Tests verify that SSL certificate verification can be configured via:
- Command-line arguments (--ssl-verify, --no-ssl-verify, --ssl-cert-path)
- Environment variables (OPTILLM_SSL_VERIFY, OPTILLM_SSL_CERT_PATH)
- And that SSL settings are properly propagated to HTTP clients
"""
import unittest
from unittest.mock import Mock, patch, MagicMock, call
import sys
import os
import tempfile
import httpx
# Add parent directory to path to import optillm modules
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from optillm import server_config, parse_args
class TestSSLConfiguration(unittest.TestCase):
"""Test SSL configuration via CLI arguments and environment variables."""
def setUp(self):
"""Reset server_config before each test."""
# Save original config
self.original_config = server_config.copy()
# Clear SSL-related environment variables
for key in ['OPTILLM_SSL_VERIFY', 'OPTILLM_SSL_CERT_PATH']:
if key in os.environ:
del os.environ[key]
def tearDown(self):
"""Restore original server_config after each test."""
server_config.clear()
server_config.update(self.original_config)
def test_default_ssl_verify_enabled(self):
"""Test that SSL verification is enabled by default."""
self.assertTrue(server_config.get('ssl_verify', True))
self.assertEqual(server_config.get('ssl_cert_path', ''), '')
def test_cli_no_ssl_verify_flag(self):
"""Test --no-ssl-verify CLI flag disables SSL verification."""
with patch('sys.argv', ['optillm', '--no-ssl-verify']):
args = parse_args()
self.assertFalse(args.ssl_verify)
def test_cli_ssl_cert_path(self):
"""Test --ssl-cert-path CLI argument."""
test_cert_path = '/path/to/ca-bundle.crt'
with patch('sys.argv', ['optillm', '--ssl-cert-path', test_cert_path]):
args = parse_args()
self.assertEqual(args.ssl_cert_path, test_cert_path)
def test_env_ssl_verify_false(self):
"""Test OPTILLM_SSL_VERIFY=false environment variable."""
os.environ['OPTILLM_SSL_VERIFY'] = 'false'
with patch('sys.argv', ['optillm']):
args = parse_args()
self.assertFalse(args.ssl_verify)
def test_env_ssl_verify_true(self):
"""Test OPTILLM_SSL_VERIFY=true environment variable."""
os.environ['OPTILLM_SSL_VERIFY'] = 'true'
with patch('sys.argv', ['optillm']):
args = parse_args()
self.assertTrue(args.ssl_verify)
def test_env_ssl_cert_path(self):
"""Test OPTILLM_SSL_CERT_PATH environment variable."""
test_cert_path = '/etc/ssl/certs/custom-ca.pem'
os.environ['OPTILLM_SSL_CERT_PATH'] = test_cert_path
with patch('sys.argv', ['optillm']):
args = parse_args()
self.assertEqual(args.ssl_cert_path, test_cert_path)
def test_cli_overrides_env(self):
"""Test that CLI arguments override environment variables."""
os.environ['OPTILLM_SSL_VERIFY'] = 'true'
with patch('sys.argv', ['optillm', '--no-ssl-verify']):
args = parse_args()
self.assertFalse(args.ssl_verify)
class TestHTTPClientSSLConfiguration(unittest.TestCase):
"""Test that SSL configuration is properly applied to HTTP clients."""
def setUp(self):
"""Set up test environment."""
self.original_config = server_config.copy()
def tearDown(self):
"""Restore original server_config."""
server_config.clear()
server_config.update(self.original_config)
@patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'})
def test_httpx_client_ssl_verify_disabled(self):
"""Test httpx.Client created with verify=False when SSL disabled."""
from optillm.server import get_config
# Configure to disable SSL verification
server_config['ssl_verify'] = False
server_config['ssl_cert_path'] = ''
# Create client
with patch('httpx.Client') as mock_httpx_client, \
patch('optillm.server.OpenAI') as mock_openai:
get_config()
# Verify httpx.Client was called with verify=False
mock_httpx_client.assert_called_once_with(verify=False)
@patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'})
def test_httpx_client_ssl_verify_enabled(self):
"""Test httpx.Client created with verify=True by default."""
from optillm.server import get_config
# Configure to enable SSL verification (default)
server_config['ssl_verify'] = True
server_config['ssl_cert_path'] = ''
# Create client
with patch('httpx.Client') as mock_httpx_client, \
patch('optillm.server.OpenAI') as mock_openai:
get_config()
# Verify httpx.Client was called with verify=True
mock_httpx_client.assert_called_once_with(verify=True)
@patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'})
def test_httpx_client_custom_cert_path(self):
"""Test httpx.Client created with custom certificate path."""
from optillm.server import get_config
# Configure custom certificate path
test_cert_path = '/path/to/custom-ca.pem'
server_config['ssl_verify'] = True
server_config['ssl_cert_path'] = test_cert_path
# Create client
with patch('httpx.Client') as mock_httpx_client, \
patch('optillm.server.OpenAI') as mock_openai:
get_config()
# Verify httpx.Client was called with custom cert path
mock_httpx_client.assert_called_once_with(verify=test_cert_path)
@patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key', 'OPTILLM_API_KEY': ''}, clear=False)
def test_openai_client_receives_http_client(self):
"""Test that OpenAI client receives the configured httpx client."""
from optillm.server import get_config
# Ensure OPTILLM_API_KEY is not set (it takes precedence)
if 'OPTILLM_API_KEY' in os.environ:
del os.environ['OPTILLM_API_KEY']
server_config['ssl_verify'] = False
server_config['ssl_cert_path'] = ''
server_config['base_url'] = ''
mock_http_client_instance = MagicMock()
with patch('httpx.Client', return_value=mock_http_client_instance) as mock_httpx_client, \
patch('optillm.server.OpenAI') as mock_openai:
get_config()
# Verify OpenAI was called with http_client parameter
mock_openai.assert_called_once()
call_kwargs = mock_openai.call_args[1]
self.assertIn('http_client', call_kwargs)
self.assertEqual(call_kwargs['http_client'], mock_http_client_instance)
@patch.dict(os.environ, {'CEREBRAS_API_KEY': 'test-key', 'OPTILLM_API_KEY': ''}, clear=False)
def test_cerebras_client_receives_http_client(self):
"""Test that Cerebras client receives the configured httpx client."""
from optillm.server import get_config
# Ensure OPTILLM_API_KEY is not set (it takes precedence)
if 'OPTILLM_API_KEY' in os.environ:
del os.environ['OPTILLM_API_KEY']
server_config['ssl_verify'] = False
server_config['ssl_cert_path'] = ''
server_config['base_url'] = ''
mock_http_client_instance = MagicMock()
with patch('httpx.Client', return_value=mock_http_client_instance) as mock_httpx_client, \
patch('optillm.server.Cerebras') as mock_cerebras:
get_config()
# Verify Cerebras was called with http_client parameter
mock_cerebras.assert_called_once()
call_kwargs = mock_cerebras.call_args[1]
self.assertIn('http_client', call_kwargs)
self.assertEqual(call_kwargs['http_client'], mock_http_client_instance)
@patch.dict(os.environ, {'AZURE_OPENAI_API_KEY': 'test-key', 'AZURE_API_VERSION': '2024-02-15-preview', 'AZURE_API_BASE': 'https://test.openai.azure.com', 'OPTILLM_API_KEY': ''}, clear=False)
def test_azure_client_receives_http_client(self):
"""Test that AzureOpenAI client receives the configured httpx client."""
from optillm.server import get_config
# Ensure OPTILLM_API_KEY is not set (it takes precedence)
if 'OPTILLM_API_KEY' in os.environ:
del os.environ['OPTILLM_API_KEY']
server_config['ssl_verify'] = False
server_config['ssl_cert_path'] = ''
mock_http_client_instance = MagicMock()
with patch('httpx.Client', return_value=mock_http_client_instance) as mock_httpx_client, \
patch('optillm.server.AzureOpenAI') as mock_azure:
get_config()
# Verify AzureOpenAI was called with http_client parameter
mock_azure.assert_called_once()
call_kwargs = mock_azure.call_args[1]
self.assertIn('http_client', call_kwargs)
self.assertEqual(call_kwargs['http_client'], mock_http_client_instance)
class TestPluginSSLConfiguration(unittest.TestCase):
"""Test that plugins properly use SSL configuration."""
def setUp(self):
"""Set up test environment."""
self.original_config = server_config.copy()
def tearDown(self):
"""Restore original server_config."""
server_config.clear()
server_config.update(self.original_config)
@patch('optillm.plugins.readurls_plugin.requests.get')
def test_readurls_plugin_ssl_verify_disabled(self, mock_requests_get):
"""Test readurls plugin respects SSL verification disabled."""
from optillm.plugins.readurls_plugin import fetch_webpage_content
# Configure to disable SSL verification
server_config['ssl_verify'] = False
server_config['ssl_cert_path'] = ''
# Mock response
mock_response = MagicMock()
mock_response.content = b'<html><body><p>Test content</p></body></html>'
mock_response.raise_for_status = MagicMock()
mock_requests_get.return_value = mock_response
# Fetch webpage
fetch_webpage_content('https://example.com')
# Verify requests.get was called with verify=False
mock_requests_get.assert_called_once()
call_kwargs = mock_requests_get.call_args[1]
self.assertIn('verify', call_kwargs)
self.assertFalse(call_kwargs['verify'])
@patch('optillm.plugins.readurls_plugin.requests.get')
def test_readurls_plugin_ssl_verify_enabled(self, mock_requests_get):
"""Test readurls plugin respects SSL verification enabled."""
from optillm.plugins.readurls_plugin import fetch_webpage_content
# Configure to enable SSL verification
server_config['ssl_verify'] = True
server_config['ssl_cert_path'] = ''
# Mock response
mock_response = MagicMock()
mock_response.content = b'<html><body><p>Test content</p></body></html>'
mock_response.raise_for_status = MagicMock()
mock_requests_get.return_value = mock_response
# Fetch webpage
fetch_webpage_content('https://example.com')
# Verify requests.get was called with verify=True
mock_requests_get.assert_called_once()
call_kwargs = mock_requests_get.call_args[1]
self.assertIn('verify', call_kwargs)
self.assertTrue(call_kwargs['verify'])
@patch('optillm.plugins.readurls_plugin.requests.get')
def test_readurls_plugin_custom_cert_path(self, mock_requests_get):
"""Test readurls plugin uses custom certificate path."""
from optillm.plugins.readurls_plugin import fetch_webpage_content
# Configure custom certificate path
test_cert_path = '/path/to/custom-ca.pem'
server_config['ssl_verify'] = True
server_config['ssl_cert_path'] = test_cert_path
# Mock response
mock_response = MagicMock()
mock_response.content = b'<html><body><p>Test content</p></body></html>'
mock_response.raise_for_status = MagicMock()
mock_requests_get.return_value = mock_response
# Fetch webpage
fetch_webpage_content('https://example.com')
# Verify requests.get was called with custom cert path
mock_requests_get.assert_called_once()
call_kwargs = mock_requests_get.call_args[1]
self.assertIn('verify', call_kwargs)
self.assertEqual(call_kwargs['verify'], test_cert_path)
class TestSSLWarnings(unittest.TestCase):
"""Test that appropriate warnings are shown when SSL is disabled."""
def setUp(self):
"""Set up test environment."""
self.original_config = server_config.copy()
def tearDown(self):
"""Restore original server_config."""
server_config.clear()
server_config.update(self.original_config)
@patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key'})
def test_warning_when_ssl_disabled(self):
"""Test that a warning is logged when SSL verification is disabled."""
from optillm.server import get_config
# Configure to disable SSL verification
server_config['ssl_verify'] = False
server_config['ssl_cert_path'] = ''
with patch('httpx.Client') as mock_httpx_client, \
patch('optillm.server.OpenAI') as mock_openai, \
patch('optillm.server.logger.warning') as mock_logger_warning:
get_config()
# Verify warning was logged
mock_logger_warning.assert_called()
warning_message = mock_logger_warning.call_args[0][0]
self.assertIn('SSL certificate verification is DISABLED', warning_message)
self.assertIn('insecure', warning_message.lower())
@patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key', 'OPTILLM_API_KEY': ''}, clear=False)
def test_info_when_custom_cert_used(self):
"""Test that an info message is logged when using custom certificate."""
from optillm.server import get_config
# Ensure OPTILLM_API_KEY is not set (it takes precedence)
if 'OPTILLM_API_KEY' in os.environ:
del os.environ['OPTILLM_API_KEY']
# Configure custom certificate path
test_cert_path = '/path/to/custom-ca.pem'
server_config['ssl_verify'] = True
server_config['ssl_cert_path'] = test_cert_path
with patch('httpx.Client') as mock_httpx_client, \
patch('optillm.server.OpenAI') as mock_openai, \
patch('optillm.server.logger.info') as mock_logger_info:
get_config()
# Verify info message was logged about custom cert
# The logger.info is called multiple times, check all calls
all_info_messages = [call[0][0] for call in mock_logger_info.call_args_list if call[0]]
cert_message_found = any('custom CA certificate bundle' in msg for msg in all_info_messages)
self.assertTrue(cert_message_found, f"Expected 'custom CA certificate bundle' in one of: {all_info_messages}")
if __name__ == '__main__':
unittest.main()