llm_council/backend/tests/test_openai_compat.py

import unittest

import httpx
import json

from backend.openai_compat import _resolve_chat_completions_url, _resolve_models_url, query_model, list_models


class TestOpenAICompatUrl(unittest.TestCase):
    def test_resolve_url_when_no_v1(self):
        self.assertEqual(
            _resolve_chat_completions_url("http://gpu:8000"),
            "http://gpu:8000/v1/chat/completions",
        )

    def test_resolve_url_when_v1(self):
        self.assertEqual(
            _resolve_chat_completions_url("http://gpu:8000/v1"),
            "http://gpu:8000/v1/chat/completions",
        )

    def test_resolve_url_when_v1_with_trailing_slash(self):
        self.assertEqual(
            _resolve_chat_completions_url("http://gpu:8000/v1/"),
            "http://gpu:8000/v1/chat/completions",
        )

    def test_resolve_models_url(self):
        self.assertEqual(
            _resolve_models_url("http://gpu:8000"),
            "http://gpu:8000/v1/models",
        )


class TestOpenAICompatRequest(unittest.IsolatedAsyncioTestCase):
    async def test_query_model_builds_payload_and_parses_response(self):
        captured = {}

        def handler(request: httpx.Request) -> httpx.Response:
            captured["url"] = str(request.url)
            captured["auth"] = request.headers.get("authorization")
            captured["json"] = json.loads(request.content.decode("utf-8"))
            return httpx.Response(
                200,
                json={
                    "choices": [
                        {
                            "message": {"content": "hello", "reasoning_details": None},
                        }
                    ]
                },
            )

        transport = httpx.MockTransport(handler)
        async with httpx.AsyncClient(transport=transport, timeout=10.0) as client:
            out = await query_model(
                "my-model",
                [{"role": "user", "content": "hi"}],
                base_url="http://gpu:8000",
                api_key="secret",
                max_tokens=123,
                timeout=10.0,
                client=client,
            )

        self.assertEqual(captured["url"], "http://gpu:8000/v1/chat/completions")
        self.assertEqual(captured["auth"], "Bearer secret")
        self.assertEqual(captured["json"]["model"], "my-model")
        self.assertEqual(captured["json"]["max_tokens"], 123)
        self.assertEqual(out["content"], "hello")

    async def test_list_models_parses_ids(self):
        def handler(request: httpx.Request) -> httpx.Response:
            return httpx.Response(
                200,
                json={"data": [{"id": "a"}, {"id": "b"}, {"nope": "c"}]},
            )

        transport = httpx.MockTransport(handler)
        async with httpx.AsyncClient(transport=transport, timeout=10.0) as client:
            ids = await list_models(
                base_url="http://gpu:8000",
                client=client,
            )
        self.assertEqual(ids, ["a", "b"])