tests/test_model_watch_adapter.py

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