Spaces:
Build error
Build error
File size: 5,171 Bytes
7a05808 e4a41fa 7a05808 e4a41fa 7a05808 e4a41fa 7a05808 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | """Tests for CacheOptimizerRegistry."""
import pytest
from headroom.cache import (
AnthropicCacheOptimizer,
CacheConfig,
CacheOptimizerRegistry,
GoogleCacheOptimizer,
OpenAICacheOptimizer,
)
from headroom.cache.base import BaseCacheOptimizer, CacheResult, CacheStrategy
class MockOptimizer(BaseCacheOptimizer):
"""Mock optimizer for testing."""
@property
def name(self) -> str:
return "mock-optimizer"
@property
def provider(self) -> str:
return "mock"
@property
def strategy(self) -> CacheStrategy:
return CacheStrategy.NONE
def optimize(self, messages, context, config=None):
return CacheResult(messages=messages)
class TestCacheOptimizerRegistry:
"""Test CacheOptimizerRegistry functionality."""
def test_default_providers_registered(self):
"""Test that default providers are registered on import."""
providers = CacheOptimizerRegistry.list_all()
assert "anthropic" in providers
assert "openai" in providers
assert "google" in providers
def test_get_anthropic(self):
"""Test getting Anthropic optimizer."""
optimizer = CacheOptimizerRegistry.get("anthropic")
assert isinstance(optimizer, AnthropicCacheOptimizer)
assert optimizer.provider == "anthropic"
assert optimizer.strategy == CacheStrategy.EXPLICIT_BREAKPOINTS
def test_get_openai(self):
"""Test getting OpenAI optimizer."""
optimizer = CacheOptimizerRegistry.get("openai")
assert isinstance(optimizer, OpenAICacheOptimizer)
assert optimizer.provider == "openai"
assert optimizer.strategy == CacheStrategy.PREFIX_STABILIZATION
def test_get_google(self):
"""Test getting Google optimizer."""
optimizer = CacheOptimizerRegistry.get("google")
assert isinstance(optimizer, GoogleCacheOptimizer)
assert optimizer.provider == "google"
assert optimizer.strategy == CacheStrategy.CACHED_CONTENT
def test_get_with_config(self):
"""Test getting optimizer with custom config."""
config = CacheConfig(min_cacheable_tokens=2048)
optimizer = CacheOptimizerRegistry.get("anthropic", config=config, cached=False)
assert optimizer.config.min_cacheable_tokens >= 1024 # Anthropic enforces minimum
def test_register_custom_optimizer(self):
"""Test registering a custom optimizer."""
CacheOptimizerRegistry.register("mock", MockOptimizer)
try:
optimizer = CacheOptimizerRegistry.get("mock")
assert isinstance(optimizer, MockOptimizer)
finally:
CacheOptimizerRegistry.unregister("mock")
def test_register_duplicate_raises(self):
"""Test that registering duplicate without override raises."""
CacheOptimizerRegistry.register("test-dup", MockOptimizer)
try:
with pytest.raises(ValueError):
CacheOptimizerRegistry.register("test-dup", MockOptimizer)
finally:
CacheOptimizerRegistry.unregister("test-dup")
def test_register_with_override(self):
"""Test registering with override."""
CacheOptimizerRegistry.register("test-override", MockOptimizer)
try:
CacheOptimizerRegistry.register("test-override", MockOptimizer, override=True)
optimizer = CacheOptimizerRegistry.get("test-override")
assert isinstance(optimizer, MockOptimizer)
finally:
CacheOptimizerRegistry.unregister("test-override")
def test_get_unknown_provider_raises(self):
"""Test getting unknown provider raises KeyError."""
with pytest.raises(KeyError):
CacheOptimizerRegistry.get("unknown-provider")
def test_list_providers(self):
"""Test listing providers."""
providers = CacheOptimizerRegistry.list_providers()
assert "anthropic" in providers
assert "openai" in providers
assert "google" in providers
def test_is_registered(self):
"""Test is_registered check."""
assert CacheOptimizerRegistry.is_registered("anthropic")
assert not CacheOptimizerRegistry.is_registered("nonexistent")
def test_cached_instances(self):
"""Test that cached instances are reused."""
opt1 = CacheOptimizerRegistry.get("anthropic", cached=True)
opt2 = CacheOptimizerRegistry.get("anthropic", cached=True)
assert opt1 is opt2
def test_uncached_instances(self):
"""Test that uncached instances are not reused."""
opt1 = CacheOptimizerRegistry.get("anthropic", cached=False)
opt2 = CacheOptimizerRegistry.get("anthropic", cached=False)
assert opt1 is not opt2
def test_tier_based_selection(self):
"""Test tier-based optimizer selection."""
# OSS tier should work
oss_opt = CacheOptimizerRegistry.get("anthropic", tier="oss")
assert oss_opt is not None
# Enterprise tier falls back to OSS if not registered
ent_opt = CacheOptimizerRegistry.get("anthropic", tier="enterprise")
assert ent_opt is not None
|