Ezra
604d7f6c70
- Complete daemon with FastAPI - Ollama client for local AI (gemma3:4b) - Telegram webhook handler - Hermes bridge (thin profile) - Systemd service definition - All unit tests passing
261 lines
7.5 KiB
Python
261 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Unit tests for Archon Kion
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from ollama_client import OllamaClient
|
|
from hermes_bridge import HermesBridge
|
|
|
|
|
|
# ============================================================================
|
|
# Fixtures
|
|
# ============================================================================
|
|
|
|
@pytest.fixture
|
|
def temp_profile():
|
|
"""Create temporary profile file"""
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
|
|
profile = {
|
|
'identity': {'name': 'Test Kion', 'role': 'Test Assistant'},
|
|
'constraints': {'local_only': True, 'model': 'test-model'},
|
|
'routing': {'tag': '#test-archon'}
|
|
}
|
|
yaml.dump(profile, f)
|
|
path = f.name
|
|
yield path
|
|
os.unlink(path)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_config():
|
|
"""Create temporary config file"""
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
|
|
config = {
|
|
'ollama': {'host': 'localhost', 'port': 11434, 'model': 'gemma3:4b'},
|
|
'telegram': {'token': 'test-token', 'webhook_url': 'http://test/webhook'},
|
|
'hermes': {'profile_path': '/tmp/test-profile.yaml'}
|
|
}
|
|
yaml.dump(config, f)
|
|
path = f.name
|
|
yield path
|
|
os.unlink(path)
|
|
|
|
|
|
# ============================================================================
|
|
# Ollama Client Tests
|
|
# ============================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_client_initialization():
|
|
"""Test OllamaClient can be initialized"""
|
|
client = OllamaClient(host="localhost", port=11434, model="gemma3:4b")
|
|
assert client.host == "localhost"
|
|
assert client.port == 11434
|
|
assert client.model == "gemma3:4b"
|
|
assert client.base_url == "http://localhost:11434/api"
|
|
await client.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_health_check():
|
|
"""Test Ollama health check (requires running Ollama)"""
|
|
client = OllamaClient()
|
|
# This will fail if Ollama not running, but tests the method
|
|
result = await client.health_check()
|
|
# Result depends on whether Ollama is running
|
|
assert isinstance(result, bool)
|
|
await client.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_generate_sync():
|
|
"""Test synchronous generation (requires Ollama)"""
|
|
client = OllamaClient()
|
|
|
|
# Only test if Ollama is available
|
|
if await client.health_check():
|
|
response = await client.generate_sync("Say 'test' only.")
|
|
assert isinstance(response, str)
|
|
assert len(response) > 0
|
|
|
|
await client.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_list_models():
|
|
"""Test listing models (requires Ollama)"""
|
|
client = OllamaClient()
|
|
|
|
models = await client.list_models()
|
|
assert isinstance(models, list)
|
|
|
|
# If Ollama is running, should have models
|
|
if await client.health_check():
|
|
assert len(models) > 0
|
|
assert any('gemma' in m for m in models)
|
|
|
|
await client.close()
|
|
|
|
|
|
# ============================================================================
|
|
# Hermes Bridge Tests
|
|
# ============================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hermes_bridge_initialization(temp_profile):
|
|
"""Test HermesBridge loads profile"""
|
|
bridge = HermesBridge(profile_path=temp_profile)
|
|
|
|
identity = bridge.get_identity()
|
|
assert identity['name'] == 'Test Kion'
|
|
assert identity['role'] == 'Test Assistant'
|
|
|
|
constraints = bridge.get_constraints()
|
|
assert constraints['local_only'] is True
|
|
|
|
assert bridge.get_routing_tag() == '#test-archon'
|
|
|
|
|
|
def test_hermes_system_prompt(temp_profile):
|
|
"""Test system prompt generation"""
|
|
bridge = HermesBridge(profile_path=temp_profile)
|
|
prompt = bridge.get_system_prompt()
|
|
|
|
assert 'Test Kion' in prompt
|
|
assert 'Test Assistant' in prompt
|
|
assert 'local' in prompt.lower()
|
|
|
|
|
|
def test_hermes_should_handle(temp_profile):
|
|
"""Test message routing logic"""
|
|
bridge = HermesBridge(profile_path=temp_profile)
|
|
|
|
# Should handle commands
|
|
assert bridge.should_handle('/status') is True
|
|
|
|
# Should handle tagged messages
|
|
assert bridge.should_handle('Hello #test-archon') is True
|
|
|
|
# Should not handle regular messages
|
|
assert bridge.should_handle('Hello world') is False
|
|
|
|
|
|
def test_hermes_default_profile():
|
|
"""Test default profile when file missing"""
|
|
bridge = HermesBridge(profile_path='/nonexistent/path.yaml')
|
|
|
|
identity = bridge.get_identity()
|
|
assert 'name' in identity
|
|
assert identity.get('name') == 'Archon Kion'
|
|
|
|
|
|
# ============================================================================
|
|
# Integration Tests
|
|
# ============================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_pipeline():
|
|
"""Integration test: Full pipeline (requires Ollama)"""
|
|
client = OllamaClient()
|
|
|
|
# Skip if Ollama not available
|
|
if not await client.health_check():
|
|
pytest.skip("Ollama not available")
|
|
|
|
# Test generation pipeline
|
|
response = await client.generate_sync(
|
|
prompt="What is 2+2? Answer with just the number.",
|
|
system="You are a helpful assistant. Be concise."
|
|
)
|
|
|
|
assert '4' in response
|
|
|
|
await client.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_simulation():
|
|
"""Test memory handling in bot"""
|
|
from telegram_bot import TelegramBot
|
|
|
|
# Create mock components
|
|
memory = {}
|
|
client = OllamaClient()
|
|
bridge = HermesBridge(profile_path='/nonexistent.yaml')
|
|
|
|
bot = TelegramBot(
|
|
token="test-token",
|
|
webhook_url="http://test/webhook",
|
|
ollama_client=client,
|
|
hermes_bridge=bridge,
|
|
memory=memory
|
|
)
|
|
|
|
# Simulate message handling
|
|
chat_id = "12345"
|
|
if chat_id not in memory:
|
|
memory[chat_id] = []
|
|
|
|
memory[chat_id].append({"role": "user", "content": "Hello"})
|
|
memory[chat_id].append({"role": "assistant", "content": "Hi there!"})
|
|
|
|
assert len(memory[chat_id]) == 2
|
|
assert memory[chat_id][0]['role'] == 'user'
|
|
|
|
await client.close()
|
|
|
|
|
|
# ============================================================================
|
|
# Configuration Tests
|
|
# ============================================================================
|
|
|
|
def test_config_loading():
|
|
"""Test YAML config loading"""
|
|
config_path = Path(__file__).parent.parent / "config" / "archon-kion.yaml"
|
|
|
|
if config_path.exists():
|
|
with open(config_path) as f:
|
|
config = yaml.safe_load(f)
|
|
|
|
assert 'ollama' in config
|
|
assert 'telegram' in config
|
|
assert 'hermes' in config
|
|
|
|
assert config['ollama']['model'] == 'gemma3:4b'
|
|
|
|
|
|
def test_profile_loading():
|
|
"""Test YAML profile loading"""
|
|
profile_path = Path(__file__).parent.parent / "hermes-profile" / "profile.yaml"
|
|
|
|
if profile_path.exists():
|
|
with open(profile_path) as f:
|
|
profile = yaml.safe_load(f)
|
|
|
|
assert 'identity' in profile
|
|
assert 'constraints' in profile
|
|
assert 'routing' in profile
|
|
|
|
assert profile['routing']['tag'] == '#archon-kion'
|
|
|
|
|
|
# ============================================================================
|
|
# Main
|
|
# ============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|