Spaces:
Running
Running
| import unittest | |
| from types import SimpleNamespace as Obj | |
| from provider_errors import provider_diagnostics, rate_limit_message | |
| class ProviderErrorTests(unittest.TestCase): | |
| def test_quota_fields_are_extracted_without_private_content(self): | |
| exc = Obj(status_code=429, body={'error': {'code': 'rate_limit_exceeded', | |
| 'message': 'Private prompt and organization details. tokens per day (TPD): Limit 0, Used 0, Requested 128'}}, | |
| response=Obj(headers={'retry-after': '2.5'})) | |
| d = provider_diagnostics(exc) | |
| self.assertEqual(d, dict(http_status=429, code='rate_limit_exceeded', limit_type='TPD', | |
| limit=0, used=0, requested=128, retry_after=3)) | |
| self.assertIn('zero allowance', rate_limit_message(d)) | |
| self.assertNotIn('Private', str(d)) | |
| def test_absent_and_malformed_diagnostics(self): | |
| for body in [None, 'private details', {'message': 12, 'code': 'malicious key!'}]: | |
| self.assertEqual(provider_diagnostics(Obj(body=body)), {}) | |
| for delay in ['NaN', 'inf', '-1', 'abc']: | |
| self.assertEqual(provider_diagnostics(Obj(response=Obj(headers={'retry-after': delay}))), {}) | |
| self.assertIn('did not provide', rate_limit_message({})) | |
| self.assertIn('wait 3 seconds', rate_limit_message({'retry_after': 3})) | |