210 lines
7.1 KiB
Python
210 lines
7.1 KiB
Python
"""Tests for LLM adapters."""
|
|
|
|
import pytest
|
|
|
|
from fusionagi.adapters.base import LLMAdapter
|
|
from fusionagi.adapters.stub_adapter import StubAdapter
|
|
from fusionagi.adapters.cache import CachedAdapter
|
|
|
|
|
|
class TestStubAdapter:
|
|
"""Test StubAdapter functionality."""
|
|
|
|
def test_complete_returns_configured_response(self):
|
|
"""Test that complete() returns the configured response."""
|
|
adapter = StubAdapter(response="Test response")
|
|
|
|
result = adapter.complete([{"role": "user", "content": "Hello"}])
|
|
|
|
assert result == "Test response"
|
|
|
|
def test_complete_structured_with_dict_response(self):
|
|
"""Test complete_structured with configured dict response."""
|
|
adapter = StubAdapter(
|
|
response="ignored",
|
|
structured_response={"key": "value", "number": 42},
|
|
)
|
|
|
|
result = adapter.complete_structured([{"role": "user", "content": "Hello"}])
|
|
|
|
assert result == {"key": "value", "number": 42}
|
|
|
|
def test_complete_structured_parses_json_response(self):
|
|
"""Test complete_structured parses JSON from text response."""
|
|
adapter = StubAdapter(response='{"parsed": true}')
|
|
|
|
result = adapter.complete_structured([{"role": "user", "content": "Hello"}])
|
|
|
|
assert result == {"parsed": True}
|
|
|
|
def test_complete_structured_returns_none_for_non_json(self):
|
|
"""Test complete_structured returns None for non-JSON text."""
|
|
adapter = StubAdapter(response="Not JSON at all")
|
|
|
|
result = adapter.complete_structured([{"role": "user", "content": "Hello"}])
|
|
|
|
assert result is None
|
|
|
|
def test_set_response(self):
|
|
"""Test dynamically changing the response."""
|
|
adapter = StubAdapter(response="Initial")
|
|
|
|
assert adapter.complete([]) == "Initial"
|
|
|
|
adapter.set_response("Changed")
|
|
assert adapter.complete([]) == "Changed"
|
|
|
|
def test_set_structured_response(self):
|
|
"""Test dynamically changing the structured response."""
|
|
adapter = StubAdapter()
|
|
|
|
adapter.set_structured_response({"dynamic": True})
|
|
result = adapter.complete_structured([])
|
|
|
|
assert result == {"dynamic": True}
|
|
|
|
|
|
class TestCachedAdapter:
|
|
"""Test CachedAdapter functionality."""
|
|
|
|
def test_caches_responses(self):
|
|
"""Test that responses are cached."""
|
|
# Track how many times the underlying adapter is called
|
|
call_count = 0
|
|
|
|
class CountingAdapter(LLMAdapter):
|
|
def complete(self, messages, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return f"Response {call_count}"
|
|
|
|
underlying = CountingAdapter()
|
|
cached = CachedAdapter(underlying, max_entries=10)
|
|
|
|
messages = [{"role": "user", "content": "Hello"}]
|
|
|
|
# First call - cache miss
|
|
result1 = cached.complete(messages)
|
|
assert call_count == 1
|
|
|
|
# Second call with same messages - cache hit
|
|
result2 = cached.complete(messages)
|
|
assert call_count == 1 # Not incremented
|
|
assert result1 == result2
|
|
|
|
def test_cache_eviction(self):
|
|
"""Test LRU cache eviction when at capacity."""
|
|
underlying = StubAdapter(response="cached")
|
|
cached = CachedAdapter(underlying, max_entries=2)
|
|
|
|
# Fill the cache
|
|
cached.complete([{"role": "user", "content": "msg1"}])
|
|
cached.complete([{"role": "user", "content": "msg2"}])
|
|
|
|
# This should trigger eviction
|
|
cached.complete([{"role": "user", "content": "msg3"}])
|
|
|
|
stats = cached.get_stats()
|
|
assert stats["text_cache_size"] == 2
|
|
|
|
def test_cache_stats(self):
|
|
"""Test cache statistics."""
|
|
underlying = StubAdapter(response="test")
|
|
cached = CachedAdapter(underlying, max_entries=10)
|
|
|
|
messages = [{"role": "user", "content": "Hello"}]
|
|
|
|
cached.complete(messages) # Miss
|
|
cached.complete(messages) # Hit
|
|
cached.complete(messages) # Hit
|
|
|
|
stats = cached.get_stats()
|
|
|
|
assert stats["hits"] == 2
|
|
assert stats["misses"] == 1
|
|
assert stats["hit_rate"] == 2/3
|
|
|
|
def test_clear_cache(self):
|
|
"""Test clearing the cache."""
|
|
underlying = StubAdapter(response="test")
|
|
cached = CachedAdapter(underlying, max_entries=10)
|
|
|
|
cached.complete([{"role": "user", "content": "msg"}])
|
|
|
|
stats = cached.get_stats()
|
|
assert stats["text_cache_size"] == 1
|
|
|
|
cached.clear_cache()
|
|
|
|
stats = cached.get_stats()
|
|
assert stats["text_cache_size"] == 0
|
|
assert stats["hits"] == 0
|
|
assert stats["misses"] == 0
|
|
|
|
def test_structured_cache_separate(self):
|
|
"""Test that structured responses are cached separately."""
|
|
underlying = StubAdapter(
|
|
response="text",
|
|
structured_response={"structured": True},
|
|
)
|
|
cached = CachedAdapter(underlying, max_entries=10)
|
|
|
|
messages = [{"role": "user", "content": "Hello"}]
|
|
|
|
# Text and structured have separate caches
|
|
cached.complete(messages)
|
|
cached.complete_structured(messages)
|
|
|
|
stats = cached.get_stats()
|
|
assert stats["text_cache_size"] == 1
|
|
assert stats["structured_cache_size"] == 1
|
|
|
|
def test_kwargs_affect_cache_key(self):
|
|
"""Test that different kwargs produce different cache keys."""
|
|
call_count = 0
|
|
|
|
class CountingAdapter(LLMAdapter):
|
|
def complete(self, messages, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return f"Response with temp={kwargs.get('temperature')}"
|
|
|
|
underlying = CountingAdapter()
|
|
cached = CachedAdapter(underlying, max_entries=10)
|
|
|
|
messages = [{"role": "user", "content": "Hello"}]
|
|
|
|
# Different temperature values should be separate cache entries
|
|
cached.complete(messages, temperature=0.5)
|
|
cached.complete(messages, temperature=0.7)
|
|
cached.complete(messages, temperature=0.5) # Should hit cache
|
|
|
|
assert call_count == 2
|
|
|
|
|
|
class TestLLMAdapterInterface:
|
|
"""Test that adapters conform to the LLMAdapter interface."""
|
|
|
|
def test_stub_adapter_is_llm_adapter(self):
|
|
"""Test StubAdapter is an LLMAdapter."""
|
|
adapter = StubAdapter()
|
|
assert isinstance(adapter, LLMAdapter)
|
|
|
|
def test_cached_adapter_is_llm_adapter(self):
|
|
"""Test CachedAdapter is an LLMAdapter."""
|
|
underlying = StubAdapter()
|
|
cached = CachedAdapter(underlying)
|
|
assert isinstance(cached, LLMAdapter)
|
|
|
|
def test_complete_structured_default(self):
|
|
"""Test that complete_structured has a default implementation."""
|
|
class MinimalAdapter(LLMAdapter):
|
|
def complete(self, messages, **kwargs):
|
|
return "text"
|
|
|
|
adapter = MinimalAdapter()
|
|
|
|
# Should return None by default (base implementation)
|
|
result = adapter.complete_structured([])
|
|
assert result is None
|