9e0dc4e9fe
Define the LLM adapter interface in backend/engine/adapters/base.py with async methods complete(), list_models(), and test_connection(). The AdapterResponse dataclass holds response text, token counts, latency, model name, and raw metadata. Includes 11 tests covering instantiation guards, concrete subclass behavior, and dataclass semantics.
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
"""Tests for the base adapter interface."""
|
|
|
|
import pytest
|
|
from dataclasses import asdict
|
|
from typing import Any
|
|
|
|
from engine.adapters.base import AdapterResponse, BaseAdapter
|
|
|
|
|
|
class ConcreteAdapter(BaseAdapter):
|
|
"""Minimal concrete implementation for testing."""
|
|
|
|
async def complete(
|
|
self, prompt: str, model: str, params: dict[str, Any]
|
|
) -> AdapterResponse:
|
|
return AdapterResponse(
|
|
text=f"response to: {prompt}",
|
|
tokens_in=len(prompt.split()),
|
|
tokens_out=5,
|
|
latency_ms=42.0,
|
|
model=model,
|
|
)
|
|
|
|
async def list_models(self) -> list[str]:
|
|
return ["model-a", "model-b"]
|
|
|
|
async def test_connection(self) -> bool:
|
|
return True
|
|
|
|
|
|
class IncompleteAdapter(BaseAdapter):
|
|
"""Adapter missing required methods — should not be instantiable."""
|
|
|
|
pass
|
|
|
|
|
|
class TestAdapterResponse:
|
|
def test_basic_creation(self):
|
|
resp = AdapterResponse(
|
|
text="hello", tokens_in=10, tokens_out=5, latency_ms=100.0
|
|
)
|
|
assert resp.text == "hello"
|
|
assert resp.tokens_in == 10
|
|
assert resp.tokens_out == 5
|
|
assert resp.latency_ms == 100.0
|
|
|
|
def test_defaults(self):
|
|
resp = AdapterResponse(text="hi", tokens_in=1, tokens_out=1, latency_ms=1.0)
|
|
assert resp.model == ""
|
|
assert resp.raw == {}
|
|
|
|
def test_raw_metadata(self):
|
|
raw = {"id": "chatcmpl-123", "finish_reason": "stop"}
|
|
resp = AdapterResponse(
|
|
text="ok", tokens_in=1, tokens_out=1, latency_ms=1.0, raw=raw
|
|
)
|
|
assert resp.raw["id"] == "chatcmpl-123"
|
|
|
|
def test_is_dataclass(self):
|
|
resp = AdapterResponse(text="x", tokens_in=0, tokens_out=0, latency_ms=0.0)
|
|
d = asdict(resp)
|
|
assert isinstance(d, dict)
|
|
assert "text" in d
|
|
assert "tokens_in" in d
|
|
|
|
def test_raw_default_not_shared(self):
|
|
r1 = AdapterResponse(text="a", tokens_in=0, tokens_out=0, latency_ms=0.0)
|
|
r2 = AdapterResponse(text="b", tokens_in=0, tokens_out=0, latency_ms=0.0)
|
|
r1.raw["key"] = "value"
|
|
assert "key" not in r2.raw
|
|
|
|
|
|
class TestBaseAdapter:
|
|
def test_cannot_instantiate_abstract(self):
|
|
with pytest.raises(TypeError):
|
|
BaseAdapter()
|
|
|
|
def test_cannot_instantiate_incomplete(self):
|
|
with pytest.raises(TypeError):
|
|
IncompleteAdapter()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concrete_complete(self):
|
|
adapter = ConcreteAdapter()
|
|
resp = await adapter.complete("hello world", "model-a", {})
|
|
assert isinstance(resp, AdapterResponse)
|
|
assert "hello world" in resp.text
|
|
assert resp.model == "model-a"
|
|
assert resp.tokens_in == 2
|
|
assert resp.tokens_out == 5
|
|
assert resp.latency_ms == 42.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concrete_list_models(self):
|
|
adapter = ConcreteAdapter()
|
|
models = await adapter.list_models()
|
|
assert models == ["model-a", "model-b"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concrete_test_connection(self):
|
|
adapter = ConcreteAdapter()
|
|
assert await adapter.test_connection() is True
|
|
|
|
def test_is_subclass(self):
|
|
assert issubclass(ConcreteAdapter, BaseAdapter)
|
|
adapter = ConcreteAdapter()
|
|
assert isinstance(adapter, BaseAdapter)
|