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()