Spaces:
Running
Running
File size: 9,435 Bytes
c8365f5 | 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 | import unittest
import httpx
from benchmarks.model_watch_adapter import (
CatalogStatus,
GeminiModelsAdapter,
ObserveOnlyModelsAdapter,
ProviderProfile,
models_url,
)
class ModelWatchAdapterTests(unittest.IsolatedAsyncioTestCase):
def profile(self, **overrides):
values = {
"provider": "groq",
"profile": "A",
"base_url": "https://api.example.test/openai/v1",
"api_key": "secret-not-logged",
"default_model": "openai/gpt-oss-120b",
}
values.update(overrides)
return ProviderProfile(**values)
def adapter(self, handler):
transport = httpx.MockTransport(handler)
return ObserveOnlyModelsAdapter(client=httpx.AsyncClient(transport=transport))
async def test_auto_apply_is_disabled_by_default(self):
adapter = ObserveOnlyModelsAdapter()
calls = []
async def callback(provider, old_model, new_model):
calls.append((provider, old_model, new_model))
result = await adapter.apply_updates([("groq", "old", "new")], callback)
self.assertFalse(adapter.can_auto_apply)
self.assertEqual(result["reason"], "auto_apply_disabled")
self.assertEqual(calls, [])
async def test_auto_apply_requires_marker_and_allowlist(self):
from benchmarks.model_watch_adapter import ModelWatchConfig
config = ModelWatchConfig(
auto_apply_enabled=True,
approval_marker="I_UNDERSTAND_MODEL_UPDATES",
approved_updates=(("groq", "old", "new"),),
)
adapter = ObserveOnlyModelsAdapter(config=config)
calls = []
async def callback(provider, old_model, new_model):
calls.append((provider, old_model, new_model))
result = await adapter.apply_updates(
[("groq", "old", "new"), ("gemini", "old", "new")],
callback,
)
self.assertTrue(adapter.can_auto_apply)
self.assertEqual(result["applied_count"], 1)
self.assertEqual(result["skipped_count"], 1)
self.assertEqual(calls, [("groq", "old", "new")])
async def test_catalog_available_and_default_present(self):
async def handler(request):
self.assertEqual(request.url.path, "/openai/v1/models")
self.assertEqual(request.headers["Authorization"], "Bearer secret-not-logged")
return httpx.Response(200, json={"data": [{"id": "openai/gpt-oss-120b"}, {"id": "other"}]})
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile())
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.AVAILABLE)
self.assertTrue(result.default_available)
self.assertFalse(result.should_auto_apply)
self.assertEqual(result.as_audit_record()["model_count"], 2)
async def test_unauthorized_never_suggests_apply(self):
async def handler(_request):
return httpx.Response(401, json={"error": "invalid key"})
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile())
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.UNAUTHORIZED)
self.assertEqual(result.http_status, 401)
self.assertFalse(result.should_auto_apply)
async def test_forbidden_is_distinct_from_unauthorized(self):
async def handler(_request):
return httpx.Response(403, json={"error": "forbidden"})
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile())
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.FORBIDDEN)
async def test_rate_limit_preserves_retry_after_without_secret(self):
async def handler(_request):
return httpx.Response(
429,
headers={"Retry-After": "37"},
json={"error": "quota exceeded", "key": "must-not-be-recorded"},
)
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile())
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.RATE_LIMITED)
self.assertEqual(result.retry_after_seconds, 37)
self.assertNotIn("must-not-be-recorded", result.detail)
async def test_server_error_is_provider_error(self):
async def handler(_request):
return httpx.Response(503, text="temporarily unavailable")
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile())
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.PROVIDER_ERROR)
self.assertEqual(result.http_status, 503)
async def test_malformed_catalog_is_not_empty_catalog(self):
async def handler(_request):
return httpx.Response(200, json={"models": [{"id": "x"}]})
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile())
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.MALFORMED)
self.assertIsNone(result.default_available)
async def test_timeout_is_classified(self):
async def handler(_request):
raise httpx.ReadTimeout("provider timeout")
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile())
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.TIMEOUT)
async def test_query_key_auth_does_not_use_bearer(self):
async def handler(request):
self.assertEqual(request.url.params.get("key"), "secret-not-logged")
self.assertNotIn("authorization", request.headers)
return httpx.Response(200, json=[{"id": "gemini-3.6-flash"}])
adapter = self.adapter(handler)
result = await adapter.list_models(self.profile(auth_mode="query_key", default_model="gemini-3.6-flash"))
await adapter._client.aclose()
self.assertEqual(result.status, CatalogStatus.AVAILABLE)
self.assertTrue(result.default_available)
class ModelWatchUrlTests(unittest.TestCase):
def test_normalizes_completion_url(self):
self.assertEqual(models_url("https://x/v1/chat/completions"), "https://x/v1/models")
def test_keeps_existing_models_suffix(self):
self.assertEqual(models_url("https://x/v1/models"), "https://x/v1/models")
if __name__ == "__main__":
unittest.main()
class GeminiModelsAdapterTests(unittest.IsolatedAsyncioTestCase):
def profile(self, profile="A", default_model="gemini-3.6-flash"):
return ProviderProfile(
provider="gemini",
profile=profile,
base_url="https://generativelanguage.googleapis.com/v1beta",
api_key="gemini-secret-not-logged",
default_model=default_model,
auth_mode="query_key",
)
async def test_native_models_payload_is_parsed_and_prefix_removed(self):
async def handler(request):
self.assertEqual(request.url.params.get("key"), "gemini-secret-not-logged")
return httpx.Response(200, json={"models": [
{"name": "models/gemini-3.6-flash"},
{"name": "models/gemini-3.5-flash"},
]})
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
result = await GeminiModelsAdapter(client=client).list_models(self.profile())
await client.aclose()
self.assertEqual(result.status, CatalogStatus.AVAILABLE)
self.assertEqual(result.models, ("gemini-3.6-flash", "gemini-3.5-flash"))
self.assertTrue(result.default_available)
self.assertEqual(result.metadata["catalog_format"], "gemini_native")
async def test_rate_limited_gemini_profile_is_classified(self):
async def handler(_request):
return httpx.Response(429, headers={"Retry-After": "60"}, json={"error": "quota"})
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
result = await GeminiModelsAdapter(client=client).list_models(self.profile())
await client.aclose()
self.assertEqual(result.status, CatalogStatus.RATE_LIMITED)
self.assertEqual(result.retry_after_seconds, 60)
async def test_scan_profiles_skips_429_but_keeps_healthy_profiles(self):
calls = []
async def handler(request):
profile = request.url.params.get("profile")
calls.append(request.url.path)
if len(calls) == 1:
return httpx.Response(429, json={"error": "quota"})
return httpx.Response(200, json={"models": [{"name": "models/gemini-3.6-flash"}]})
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
adapter = GeminiModelsAdapter(client=client)
from benchmarks.model_watch_adapter import scan_profiles
scan = await scan_profiles([self.profile("A"), self.profile("B")], adapter=adapter)
await client.aclose()
self.assertEqual(len(scan.skipped_rate_limited), 1)
self.assertEqual(scan.skipped_rate_limited[0].metadata["skip_reason"], "rate_limited")
self.assertEqual(len(scan.results), 1)
self.assertEqual(scan.results[0].status, CatalogStatus.AVAILABLE)
if __name__ == "__main__":
unittest.main()
|