Initial Archon Kion implementation
- 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
This commit is contained in:
1
tests/__init__.py
Normal file
1
tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# Tests package
|
||||
BIN
tests/__pycache__/__init__.cpython-312.pyc
Normal file
BIN
tests/__pycache__/__init__.cpython-312.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_archon.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_archon.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
260
tests/test_archon.py
Normal file
260
tests/test_archon.py
Normal file
@@ -0,0 +1,260 @@
|
||||
#!/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"])
|
||||
Reference in New Issue
Block a user