#!/usr/bin/env python3
"""
Test script for anti-idle system functionality.
"""
import json
import time
from pathlib import Path
from src.amicus.core import get_state_file, read_with_lock, write_with_lock
from src.amicus.config import ConfigManager
def test_cluster_metadata_initialization():
"""Test that cluster_metadata is initialized correctly."""
print("Test 1: Cluster Metadata Initialization")
config = ConfigManager()
cluster_settings = config.get("cluster_settings", {})
assert cluster_settings.get("max_agents") == 4, "Max agents should be 4"
assert cluster_settings.get("idle_timeout_seconds") == 30, "Idle timeout should be 30s"
assert cluster_settings.get("grace_period_seconds") == 30, "Grace period should be 30s"
print("✓ Cluster settings loaded correctly")
print(f" Max Agents: {cluster_settings['max_agents']}")
print(f" Idle Timeout: {cluster_settings['idle_timeout_seconds']}s")
print(f" Grace Period: {cluster_settings['grace_period_seconds']}s")
print()
def test_state_schema():
"""Test that the state schema supports new fields."""
print("Test 2: State Schema Enhancement")
state_file = get_state_file()
# Create a test state with enhanced schema
test_state = {
"summary": "Test state",
"next_steps": [
{"task": "Task 1", "status": "pending", "priority": "high", "created_at": time.time()},
{"task": "Task 2", "status": "in_progress", "priority": "medium", "created_at": time.time()},
],
"active_files": [],
"cluster_nodes": {
"Node-TEST1": {
"role": "bootstrap_manager",
"model": {"name": "test-model", "strength": "high"},
"last_heartbeat": time.time(),
"status": "working",
"current_task_id": None,
"idle_since": None,
"last_activity": time.time()
},
"Node-TEST2": {
"role": "developer",
"model": {"name": "test-model", "strength": "medium"},
"last_heartbeat": time.time(),
"status": "idle",
"current_task_id": None,
"idle_since": time.time() - 40, # Idle for 40 seconds
"last_activity": time.time() - 40
}
},
"cluster_metadata": {
"max_agents": 4,
"active_agent_count": 2,
"idle_agent_count": 1,
"pending_task_count": 1,
"manager_id": "Node-TEST1"
},
"work_distribution": {
"last_assessment": time.time(),
"workload_status": "underutilized",
"spawn_recommendation": "terminate_idle"
},
"timestamp": time.time()
}
write_with_lock(state_file, test_state)
# Read back and verify
loaded_state = read_with_lock(state_file)
assert "cluster_nodes" in loaded_state, "cluster_nodes missing"
assert "cluster_metadata" in loaded_state, "cluster_metadata missing"
assert "work_distribution" in loaded_state, "work_distribution missing"
# Verify node schema
test_node = loaded_state["cluster_nodes"]["Node-TEST2"]
assert test_node["status"] == "idle", "Node status incorrect"
assert test_node["idle_since"] is not None, "idle_since should be set"
# Verify metadata
metadata = loaded_state["cluster_metadata"]
assert metadata["max_agents"] == 4, "max_agents incorrect"
assert metadata["manager_id"] == "Node-TEST1", "manager_id incorrect"
# Verify work distribution
work_dist = loaded_state["work_distribution"]
assert work_dist["workload_status"] == "underutilized", "workload_status incorrect"
assert work_dist["spawn_recommendation"] == "terminate_idle", "spawn_recommendation incorrect"
print("✓ State schema validated successfully")
print(f" Active Nodes: {metadata['active_agent_count']}/{metadata['max_agents']}")
print(f" Idle Nodes: {metadata['idle_agent_count']}")
print(f" Workload Status: {work_dist['workload_status']}")
print()
def test_idle_detection_logic():
"""Test idle detection logic."""
print("Test 3: Idle Detection Logic")
state_file = get_state_file()
state = read_with_lock(state_file)
cluster_nodes = state.get("cluster_nodes", {})
for node_id, node_info in cluster_nodes.items():
status = node_info.get("status")
idle_since = node_info.get("idle_since")
role = node_info.get("role")
if status == "idle" and idle_since:
idle_duration = time.time() - idle_since
print(f" Node {node_id} ({role}): idle for {idle_duration:.1f}s")
if idle_duration > 30:
print(f" → Should be marked for termination")
else:
print(f" → Within grace period")
print("✓ Idle detection logic working")
print()
def cleanup():
"""Clean up test state."""
print("Cleaning up test state...")
state_file = get_state_file()
if state_file.exists():
state_file.unlink()
print("✓ Cleanup complete")
if __name__ == "__main__":
print("=" * 60)
print("Anti-Idle System Test Suite")
print("=" * 60)
print()
try:
test_cluster_metadata_initialization()
test_state_schema()
test_idle_detection_logic()
print("=" * 60)
print("All tests passed! ✓")
print("=" * 60)
except AssertionError as e:
print(f"\n✗ Test failed: {e}")
except Exception as e:
print(f"\n✗ Error: {e}")
import traceback
traceback.print_exc()
finally:
cleanup()
