ec33ac1936
Based on claude-code-tools TmuxCLIController, this refactor: - Added DockerTmuxController class for robust tmux session management - Implements send_keys() with configurable delay_enter - Implements capture_pane() for output retrieval - Implements wait_for_prompt() for pattern-based completion detection - Implements wait_for_idle() for content-hash-based idle detection - Implements wait_for_shell_prompt() for shell prompt detection Also includes workflow improvements: - Pre-task git snapshot before agent execution - Post-task commit protocol in agent guidelines Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
335 lines
9.4 KiB
Python
335 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for Luzia Status Integration
|
|
Verifies all components are properly installed and functional
|
|
|
|
Run with: python3 test_status_integration.py
|
|
"""
|
|
|
|
import sys
|
|
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='[%(levelname)s] %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def test_imports():
|
|
"""Test that all required modules can be imported"""
|
|
print("\n" + "=" * 60)
|
|
print("TEST 1: Module Imports")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
import toml
|
|
logger.info("✓ toml module available")
|
|
except ImportError:
|
|
logger.error("✗ toml module not available")
|
|
return False
|
|
|
|
try:
|
|
from luzia_status_publisher_impl import (
|
|
LuziaStatusPublisher,
|
|
StatusMessage,
|
|
StatusMessageType,
|
|
Severity
|
|
)
|
|
logger.info("✓ luzia_status_publisher_impl imported")
|
|
except ImportError as e:
|
|
logger.error(f"✗ Failed to import publisher: {e}")
|
|
return False
|
|
|
|
try:
|
|
from luzia_claude_bridge_impl import LuziaClaudeBridge, CLIStatusHelper
|
|
logger.info("✓ luzia_claude_bridge_impl imported")
|
|
except ImportError as e:
|
|
logger.error(f"✗ Failed to import bridge: {e}")
|
|
return False
|
|
|
|
try:
|
|
from luzia_status_integration import (
|
|
LuziaStatusSystem,
|
|
LuziaStatusConfig,
|
|
get_status_system
|
|
)
|
|
logger.info("✓ luzia_status_integration imported")
|
|
except ImportError as e:
|
|
logger.error(f"✗ Failed to import integration: {e}")
|
|
return False
|
|
|
|
try:
|
|
from luzia_status_handler import get_status_handler
|
|
logger.info("✓ luzia_status_handler imported")
|
|
except ImportError as e:
|
|
logger.error(f"✗ Failed to import handler: {e}")
|
|
return False
|
|
|
|
try:
|
|
from luzia_enhanced_status_route import route_status_enhanced
|
|
logger.info("✓ luzia_enhanced_status_route imported")
|
|
except ImportError as e:
|
|
logger.error(f"✗ Failed to import enhanced route: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def test_configuration():
|
|
"""Test configuration loading"""
|
|
print("\n" + "=" * 60)
|
|
print("TEST 2: Configuration Loading")
|
|
print("=" * 60)
|
|
|
|
config_path = Path("/etc/luzia/status_config.toml")
|
|
if not config_path.exists():
|
|
logger.error(f"✗ Config file not found: {config_path}")
|
|
return False
|
|
logger.info(f"✓ Config file found: {config_path}")
|
|
|
|
try:
|
|
from luzia_status_integration import LuziaStatusConfig
|
|
config = LuziaStatusConfig()
|
|
logger.info(f"✓ Config loaded successfully")
|
|
|
|
verbosity = config.get("status_updates.verbosity")
|
|
logger.info(f" - Verbosity: {verbosity}")
|
|
|
|
show_started = config.get("status_updates.show_task_started")
|
|
logger.info(f" - Show task started: {show_started}")
|
|
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"✗ Failed to load config: {e}")
|
|
return False
|
|
|
|
|
|
def test_directories():
|
|
"""Test that required directories exist"""
|
|
print("\n" + "=" * 60)
|
|
print("TEST 3: Directory Structure")
|
|
print("=" * 60)
|
|
|
|
paths = [
|
|
Path("/etc/luzia"),
|
|
Path("/var/log/luzia"),
|
|
Path("/opt/server-agents/orchestrator/lib"),
|
|
]
|
|
|
|
all_exist = True
|
|
for path in paths:
|
|
if path.exists():
|
|
logger.info(f"✓ {path}")
|
|
else:
|
|
logger.error(f"✗ Missing: {path}")
|
|
all_exist = False
|
|
|
|
return all_exist
|
|
|
|
|
|
def test_files():
|
|
"""Test that all required files exist"""
|
|
print("\n" + "=" * 60)
|
|
print("TEST 4: Required Files")
|
|
print("=" * 60)
|
|
|
|
files = [
|
|
"/etc/luzia/status_config.toml",
|
|
"/opt/server-agents/orchestrator/lib/luzia_status_publisher_impl.py",
|
|
"/opt/server-agents/orchestrator/lib/luzia_claude_bridge_impl.py",
|
|
"/opt/server-agents/orchestrator/lib/luzia_status_integration.py",
|
|
"/opt/server-agents/orchestrator/lib/luzia_status_handler.py",
|
|
"/opt/server-agents/orchestrator/lib/luzia_enhanced_status_route.py",
|
|
]
|
|
|
|
all_exist = True
|
|
for file in files:
|
|
path = Path(file)
|
|
if path.exists():
|
|
size = path.stat().st_size
|
|
logger.info(f"✓ {file} ({size} bytes)")
|
|
else:
|
|
logger.error(f"✗ Missing: {file}")
|
|
all_exist = False
|
|
|
|
return all_exist
|
|
|
|
|
|
async def test_status_system():
|
|
"""Test status system initialization and basic operations"""
|
|
print("\n" + "=" * 60)
|
|
print("TEST 5: Status System Functionality")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
from luzia_status_integration import get_status_system
|
|
status_system = get_status_system()
|
|
|
|
if not status_system.is_enabled():
|
|
logger.error("✗ Status system not enabled")
|
|
return False
|
|
|
|
logger.info("✓ Status system initialized")
|
|
|
|
# Test publisher
|
|
if status_system.publisher:
|
|
logger.info("✓ Publisher available")
|
|
status_system.publisher.set_verbosity("verbose")
|
|
logger.info(" - Verbosity set to verbose")
|
|
else:
|
|
logger.error("✗ Publisher not available")
|
|
return False
|
|
|
|
# Test bridge
|
|
if status_system.bridge:
|
|
logger.info("✓ Bridge available")
|
|
else:
|
|
logger.error("✗ Bridge not available")
|
|
return False
|
|
|
|
# Test publishing a task
|
|
task_id = f"test-{datetime.now().strftime('%H%M%S')}"
|
|
logger.info(f" - Publishing test task: {task_id}")
|
|
|
|
await status_system.publish_task_started(
|
|
task_id=task_id,
|
|
project="test",
|
|
description="Integration test task",
|
|
estimated_duration_seconds=60
|
|
)
|
|
logger.info(" - Task started event published")
|
|
|
|
# Test progress
|
|
await status_system.publish_progress(
|
|
task_id=task_id,
|
|
progress_percent=50,
|
|
current_step=2,
|
|
total_steps=4,
|
|
current_step_name="Testing",
|
|
elapsed_seconds=30,
|
|
estimated_remaining_seconds=30
|
|
)
|
|
logger.info(" - Progress update published")
|
|
|
|
# Test completion
|
|
await status_system.publish_task_completed(
|
|
task_id=task_id,
|
|
elapsed_seconds=60,
|
|
findings_count=2,
|
|
status="APPROVED"
|
|
)
|
|
logger.info(" - Task completed event published")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"✗ Status system test failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
def test_cli_handler():
|
|
"""Test CLI handler"""
|
|
print("\n" + "=" * 60)
|
|
print("TEST 6: CLI Handler")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
from luzia_status_handler import get_status_handler
|
|
handler = get_status_handler()
|
|
|
|
if not handler.is_available():
|
|
logger.warning("⚠ Status handler not available (status system may not be fully initialized)")
|
|
return True # Not a failure, just not available
|
|
|
|
logger.info("✓ Status handler initialized")
|
|
|
|
# Test dashboard command
|
|
dashboard = handler.handle_command([])
|
|
if "LUZIA STATUS DASHBOARD" in dashboard or "No recent updates" in dashboard:
|
|
logger.info("✓ Dashboard command works")
|
|
else:
|
|
logger.warning("⚠ Dashboard output unexpected")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"✗ CLI handler test failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
def test_enhanced_route():
|
|
"""Test enhanced route function"""
|
|
print("\n" + "=" * 60)
|
|
print("TEST 7: Enhanced Route Function")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
from luzia_enhanced_status_route import route_status_enhanced
|
|
|
|
logger.info("✓ Enhanced route function imported")
|
|
|
|
# Test that it can be called (won't actually execute due to no config)
|
|
logger.info("✓ Enhanced route function ready for integration")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"✗ Enhanced route test failed: {e}")
|
|
return False
|
|
|
|
|
|
async def run_all_tests():
|
|
"""Run all tests"""
|
|
print("\n" + "=" * 60)
|
|
print("LUZIA STATUS INTEGRATION TEST SUITE")
|
|
print("=" * 60)
|
|
|
|
results = {
|
|
"Imports": test_imports(),
|
|
"Configuration": test_configuration(),
|
|
"Directories": test_directories(),
|
|
"Files": test_files(),
|
|
"Enhanced Route": test_enhanced_route(),
|
|
"CLI Handler": test_cli_handler(),
|
|
}
|
|
|
|
# Async tests
|
|
results["Status System"] = await test_status_system()
|
|
|
|
# Print summary
|
|
print("\n" + "=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
|
|
passed = sum(1 for v in results.values() if v)
|
|
total = len(results)
|
|
|
|
for test, result in results.items():
|
|
status = "PASS" if result else "FAIL"
|
|
symbol = "✓" if result else "✗"
|
|
print(f"{symbol} {test}: {status}")
|
|
|
|
print(f"\nTotal: {passed}/{total} tests passed")
|
|
print("=" * 60)
|
|
|
|
if passed == total:
|
|
print("\n✓ All tests passed! Status system is ready for production.")
|
|
return 0
|
|
else:
|
|
print(f"\n✗ {total - passed} test(s) failed. Please review the output above.")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit_code = asyncio.run(run_all_tests())
|
|
sys.exit(exit_code)
|