47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from core.provider_health import probe_provider
|
|
|
|
|
|
class ProviderHealthTests(unittest.TestCase):
|
|
@patch('core.provider_health.requests.get')
|
|
def test_falls_back_to_next_endpoint_and_validates_json_shape(self, get):
|
|
first = Mock(status_code=503)
|
|
second = Mock(status_code=200)
|
|
second.json.return_value = {'data': [{'id': 'model-a'}]}
|
|
get.side_effect = [first, second]
|
|
|
|
result = probe_provider('http://engine', [('/health', None), ('/v1/models', 'data')])
|
|
|
|
self.assertTrue(result['ok'])
|
|
self.assertEqual(result['endpoint'], '/v1/models')
|
|
self.assertEqual(result['payload']['data'][0]['id'], 'model-a')
|
|
self.assertEqual(get.call_args_list[1].kwargs['timeout'], 3.0)
|
|
|
|
@patch('core.provider_health.requests.get')
|
|
def test_normalizes_connection_and_timeout_errors(self, get):
|
|
import requests
|
|
get.side_effect = requests.exceptions.ConnectionError()
|
|
result = probe_provider('http://engine', [('/health', None)])
|
|
self.assertFalse(result['ok'])
|
|
self.assertEqual(result['error'], 'Connection refused')
|
|
|
|
get.side_effect = requests.exceptions.Timeout()
|
|
result = probe_provider('http://engine', [('/health', None)], timeout=2)
|
|
self.assertFalse(result['ok'])
|
|
self.assertEqual(result['error'], 'Timeout (2 s)')
|
|
|
|
@patch('core.provider_health.requests.get')
|
|
def test_does_not_accept_missing_json_key(self, get):
|
|
response = Mock(status_code=200)
|
|
response.json.return_value = {'models': []}
|
|
get.return_value = response
|
|
result = probe_provider('http://engine', [('/v1/models', 'data')])
|
|
self.assertFalse(result['ok'])
|
|
self.assertIn("missing 'data'", result['error'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|