eval_agent/design_draft/test_ev2_service.py

"""
Test script for EV2 Service

Simulates ShinkaEvolve sending notifications to the service
"""

import requests
import time
import json

# Service URL
SERVICE_URL = "http://localhost:8765"

def test_service_status():
    """Test if service is running"""
    print("🔍 Testing service status...")
    
    try:
        response = requests.get(f"{SERVICE_URL}/api/v1/status", timeout=5)
        if response.status_code == 200:
            data = response.json()
            print("✅ Service is running!")
            print(f"   Uptime: {data['uptime_seconds']:.1f}s")
            print(f"   Trigger mode: {data['config']['trigger_mode']}")
            print(f"   Trigger interval: {data['config']['trigger_interval']}")
            return True
        else:
            print(f"❌ Service returned status {response.status_code}")
            return False
    except requests.exceptions.ConnectionError:
        print("❌ Cannot connect to service. Is it running?")
        return False
    except Exception as e:
        print(f"❌ Error: {e}")
        return False


def send_generation_notification(generation: int, primary_score: float, results_dir: str):
    """Send a generation completion notification"""
    print(f"\n📤 Sending notification: gen={generation}, score={primary_score:.4f}")
    
    payload = {
        "generation": generation,
        "results_dir": results_dir,
        "primary_score": primary_score
    }
    
    try:
        response = requests.post(
            f"{SERVICE_URL}/api/v1/notify/generation_complete",
            json=payload,
            timeout=300  # Agent might take a while
        )
        
        if response.status_code == 200:
            data = response.json()
            print(f"   Status: {data['status']}")
            print(f"   Agent triggered: {data['agent_triggered']}")
            print(f"   Reason: {data['trigger_reason']}")
            print(f"   Processing time: {data['processing_time_ms']:.1f}ms")
            
            if data['agent_triggered'] and data.get('insights'):
                print(f"   Insights: {len(data['insights'])} found")
            
            return data
        else:
            print(f"❌ Request failed with status {response.status_code}")
            print(f"   Response: {response.text}")
            return None
            
    except Exception as e:
        print(f"❌ Error: {e}")
        return None


def simulate_evolution(num_generations: int = 25, results_dir: str = None):
    """
    Simulate an evolution run with changing scores
    
    This mimics what ShinkaEvolve would do
    """
    print(f"\n🧬 Simulating evolution with {num_generations} generations...")
    print(f"   Results dir: {results_dir}")
    print("=" * 70)
    
    # Simulate improving scores with some noise
    base_score = 2.40
    
    for gen in range(num_generations):
        # Simulate score improvement with noise
        improvement = gen * 0.005  # Gradual improvement
        noise = (hash(gen) % 100) / 10000  # Deterministic noise
        score = base_score + improvement + noise
        
        # Send notification
        response = send_generation_notification(gen, score, results_dir)
        
        # Small delay between generations (optional)
        time.sleep(0.1)
    
    print("\n" + "=" * 70)
    print("✅ Simulation complete!")


def test_manual_trigger(generation: int = 10):
    """Test manual trigger endpoint"""
    print(f"\n🔧 Testing manual trigger for generation {generation}...")
    
    try:
        response = requests.post(
            f"{SERVICE_URL}/api/v1/trigger/manual",
            params={"generation": generation},
            timeout=300
        )
        
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Manual trigger successful!")
            print(f"   Status: {data['status']}")
            print(f"   Message: {data['message']}")
            return data
        else:
            print(f"❌ Manual trigger failed: {response.status_code}")
            print(f"   Response: {response.text}")
            return None
            
    except Exception as e:
        print(f"❌ Error: {e}")
        return None


if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="Test EV2 Service")
    parser.add_argument("--results-dir", type=str, required=True,
                       help="Path to results directory")
    parser.add_argument("--num-gens", type=int, default=25,
                       help="Number of generations to simulate")
    parser.add_argument("--test-mode", type=str, default="simulate",
                       choices=["status", "simulate", "manual"],
                       help="Test mode")
    parser.add_argument("--generation", type=int, default=10,
                       help="Generation for manual trigger")
    
    args = parser.parse_args()
    
    # Check service status first
    if not test_service_status():
        print("\n💡 Tip: Start the service first:")
        print("   python eval_agent/ev2_service.py --config eval_agent/ev2_service_config.yaml")
        exit(1)
    
    if args.test_mode == "status":
        # Just check status
        pass
    
    elif args.test_mode == "simulate":
        # Simulate evolution
        simulate_evolution(args.num_gens, args.results_dir)
        
        # Show final status
        print("\n" + "=" * 70)
        test_service_status()
    
    elif args.test_mode == "manual":
        # Test manual trigger
        test_manual_trigger(args.generation)