Felipe Cardoso
22ecb5e989
Implements the foundation for Context Management Engine: Types (backend/app/services/context/types/): - BaseContext: Abstract base with ID, content, priority, scoring - SystemContext: System prompts, personas, instructions - KnowledgeContext: RAG results from Knowledge Base MCP - ConversationContext: Chat history with role support - TaskContext: Task/issue context with acceptance criteria - ToolContext: Tool definitions and execution results - AssembledContext: Final assembled context result Configuration (config.py): - Token budget allocation (system 5%, task 10%, knowledge 40%, etc.) - Scoring weights (relevance 50%, recency 30%, priority 20%) - Cache settings (TTL, prefix) - Performance settings (max assembly time, parallel scoring) - Environment variable overrides with CTX_ prefix Exceptions (exceptions.py): - ContextError: Base exception - BudgetExceededError: Token budget violations - TokenCountError: Token counting failures - CompressionError: Compression failures - AssemblyTimeoutError: Assembly timeout - ScoringError, FormattingError, CacheError - ContextNotFoundError, InvalidContextError All 86 tests pass. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
244 lines
8.2 KiB
Python
244 lines
8.2 KiB
Python
"""Tests for context management configuration."""
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app.services.context.config import (
|
|
ContextSettings,
|
|
get_context_settings,
|
|
get_default_settings,
|
|
reset_context_settings,
|
|
)
|
|
|
|
|
|
class TestContextSettings:
|
|
"""Tests for ContextSettings."""
|
|
|
|
def test_default_values(self) -> None:
|
|
"""Test default settings values."""
|
|
settings = ContextSettings()
|
|
|
|
# Budget defaults should sum to 1.0
|
|
total = (
|
|
settings.budget_system
|
|
+ settings.budget_task
|
|
+ settings.budget_knowledge
|
|
+ settings.budget_conversation
|
|
+ settings.budget_tools
|
|
+ settings.budget_response
|
|
+ settings.budget_buffer
|
|
)
|
|
assert abs(total - 1.0) < 0.001
|
|
|
|
# Scoring weights should sum to 1.0
|
|
weights_total = (
|
|
settings.scoring_relevance_weight
|
|
+ settings.scoring_recency_weight
|
|
+ settings.scoring_priority_weight
|
|
)
|
|
assert abs(weights_total - 1.0) < 0.001
|
|
|
|
def test_budget_allocation_values(self) -> None:
|
|
"""Test specific budget allocation values."""
|
|
settings = ContextSettings()
|
|
|
|
assert settings.budget_system == 0.05
|
|
assert settings.budget_task == 0.10
|
|
assert settings.budget_knowledge == 0.40
|
|
assert settings.budget_conversation == 0.20
|
|
assert settings.budget_tools == 0.05
|
|
assert settings.budget_response == 0.15
|
|
assert settings.budget_buffer == 0.05
|
|
|
|
def test_scoring_weights(self) -> None:
|
|
"""Test scoring weights."""
|
|
settings = ContextSettings()
|
|
|
|
assert settings.scoring_relevance_weight == 0.5
|
|
assert settings.scoring_recency_weight == 0.3
|
|
assert settings.scoring_priority_weight == 0.2
|
|
|
|
def test_cache_settings(self) -> None:
|
|
"""Test cache settings."""
|
|
settings = ContextSettings()
|
|
|
|
assert settings.cache_enabled is True
|
|
assert settings.cache_ttl_seconds == 3600
|
|
assert settings.cache_prefix == "ctx"
|
|
|
|
def test_performance_settings(self) -> None:
|
|
"""Test performance settings."""
|
|
settings = ContextSettings()
|
|
|
|
assert settings.max_assembly_time_ms == 100
|
|
assert settings.parallel_scoring is True
|
|
assert settings.max_parallel_scores == 10
|
|
|
|
def test_get_budget_allocation(self) -> None:
|
|
"""Test get_budget_allocation method."""
|
|
settings = ContextSettings()
|
|
allocation = settings.get_budget_allocation()
|
|
|
|
assert isinstance(allocation, dict)
|
|
assert "system" in allocation
|
|
assert "knowledge" in allocation
|
|
assert allocation["system"] == 0.05
|
|
assert allocation["knowledge"] == 0.40
|
|
|
|
def test_get_scoring_weights(self) -> None:
|
|
"""Test get_scoring_weights method."""
|
|
settings = ContextSettings()
|
|
weights = settings.get_scoring_weights()
|
|
|
|
assert isinstance(weights, dict)
|
|
assert "relevance" in weights
|
|
assert "recency" in weights
|
|
assert "priority" in weights
|
|
assert weights["relevance"] == 0.5
|
|
|
|
def test_to_dict(self) -> None:
|
|
"""Test to_dict method."""
|
|
settings = ContextSettings()
|
|
result = settings.to_dict()
|
|
|
|
assert "budget" in result
|
|
assert "scoring" in result
|
|
assert "compression" in result
|
|
assert "cache" in result
|
|
assert "performance" in result
|
|
assert "knowledge" in result
|
|
assert "conversation" in result
|
|
|
|
def test_budget_validation_fails_on_wrong_sum(self) -> None:
|
|
"""Test that budget validation fails when sum != 1.0."""
|
|
with pytest.raises(ValueError) as exc_info:
|
|
ContextSettings(
|
|
budget_system=0.5,
|
|
budget_task=0.5,
|
|
# Other budgets default to non-zero, so total > 1.0
|
|
)
|
|
|
|
assert "sum to 1.0" in str(exc_info.value)
|
|
|
|
def test_scoring_validation_fails_on_wrong_sum(self) -> None:
|
|
"""Test that scoring validation fails when sum != 1.0."""
|
|
with pytest.raises(ValueError) as exc_info:
|
|
ContextSettings(
|
|
scoring_relevance_weight=0.8,
|
|
scoring_recency_weight=0.8,
|
|
scoring_priority_weight=0.8,
|
|
)
|
|
|
|
assert "sum to 1.0" in str(exc_info.value)
|
|
|
|
def test_search_type_validation(self) -> None:
|
|
"""Test search type validation."""
|
|
# Valid types should work
|
|
ContextSettings(knowledge_search_type="semantic")
|
|
ContextSettings(knowledge_search_type="keyword")
|
|
ContextSettings(knowledge_search_type="hybrid")
|
|
|
|
# Invalid type should fail
|
|
with pytest.raises(ValueError):
|
|
ContextSettings(knowledge_search_type="invalid")
|
|
|
|
def test_custom_budget_allocation(self) -> None:
|
|
"""Test custom budget allocation that sums to 1.0."""
|
|
settings = ContextSettings(
|
|
budget_system=0.10,
|
|
budget_task=0.10,
|
|
budget_knowledge=0.30,
|
|
budget_conversation=0.25,
|
|
budget_tools=0.05,
|
|
budget_response=0.15,
|
|
budget_buffer=0.05,
|
|
)
|
|
|
|
total = (
|
|
settings.budget_system
|
|
+ settings.budget_task
|
|
+ settings.budget_knowledge
|
|
+ settings.budget_conversation
|
|
+ settings.budget_tools
|
|
+ settings.budget_response
|
|
+ settings.budget_buffer
|
|
)
|
|
assert abs(total - 1.0) < 0.001
|
|
|
|
|
|
class TestSettingsSingleton:
|
|
"""Tests for settings singleton pattern."""
|
|
|
|
def setup_method(self) -> None:
|
|
"""Reset settings before each test."""
|
|
reset_context_settings()
|
|
|
|
def teardown_method(self) -> None:
|
|
"""Clean up after each test."""
|
|
reset_context_settings()
|
|
|
|
def test_get_context_settings_returns_instance(self) -> None:
|
|
"""Test that get_context_settings returns a settings instance."""
|
|
settings = get_context_settings()
|
|
assert isinstance(settings, ContextSettings)
|
|
|
|
def test_get_context_settings_returns_same_instance(self) -> None:
|
|
"""Test that get_context_settings returns the same instance."""
|
|
settings1 = get_context_settings()
|
|
settings2 = get_context_settings()
|
|
assert settings1 is settings2
|
|
|
|
def test_reset_creates_new_instance(self) -> None:
|
|
"""Test that reset creates a new instance."""
|
|
settings1 = get_context_settings()
|
|
reset_context_settings()
|
|
settings2 = get_context_settings()
|
|
|
|
# Should be different instances
|
|
assert settings1 is not settings2
|
|
|
|
def test_get_default_settings_cached(self) -> None:
|
|
"""Test that get_default_settings is cached."""
|
|
settings1 = get_default_settings()
|
|
settings2 = get_default_settings()
|
|
assert settings1 is settings2
|
|
|
|
|
|
class TestEnvironmentOverrides:
|
|
"""Tests for environment variable overrides."""
|
|
|
|
def setup_method(self) -> None:
|
|
"""Reset settings before each test."""
|
|
reset_context_settings()
|
|
|
|
def teardown_method(self) -> None:
|
|
"""Clean up after each test."""
|
|
reset_context_settings()
|
|
# Clean up any env vars we set
|
|
for key in list(os.environ.keys()):
|
|
if key.startswith("CTX_"):
|
|
del os.environ[key]
|
|
|
|
def test_env_override_cache_enabled(self) -> None:
|
|
"""Test that CTX_CACHE_ENABLED env var works."""
|
|
with patch.dict(os.environ, {"CTX_CACHE_ENABLED": "false"}):
|
|
reset_context_settings()
|
|
settings = ContextSettings()
|
|
assert settings.cache_enabled is False
|
|
|
|
def test_env_override_cache_ttl(self) -> None:
|
|
"""Test that CTX_CACHE_TTL_SECONDS env var works."""
|
|
with patch.dict(os.environ, {"CTX_CACHE_TTL_SECONDS": "7200"}):
|
|
reset_context_settings()
|
|
settings = ContextSettings()
|
|
assert settings.cache_ttl_seconds == 7200
|
|
|
|
def test_env_override_max_assembly_time(self) -> None:
|
|
"""Test that CTX_MAX_ASSEMBLY_TIME_MS env var works."""
|
|
with patch.dict(os.environ, {"CTX_MAX_ASSEMBLY_TIME_MS": "200"}):
|
|
reset_context_settings()
|
|
settings = ContextSettings()
|
|
assert settings.max_assembly_time_ms == 200
|