File size: 5,595 Bytes
3f6526a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """
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)
|