9aeaa433fc
- Changed NetworkName from 'chorus_default' to 'chorus_net' - This matches the actual network 'CHORUS_chorus_net' (service prefix added automatically) - Fixes discovered_count:0 issue - now successfully discovering all 25 agents - Updated IMPLEMENTATION-SUMMARY with deployment status Result: All 25 CHORUS agents now discovered successfully via Docker Swarm API 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
441 lines
18 KiB
Python
Executable File
441 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Test Suite for Council-Generated Project Artifacts
|
|
|
|
This test verifies the complete flow:
|
|
1. Project creation triggers council formation
|
|
2. Council roles are claimed by CHORUS agents
|
|
3. Council produces artifacts
|
|
4. Artifacts are retrievable via API
|
|
|
|
Usage:
|
|
python test_council_artifacts.py
|
|
python test_council_artifacts.py --verbose
|
|
python test_council_artifacts.py --wait-time 60
|
|
"""
|
|
|
|
import requests
|
|
import time
|
|
import json
|
|
import sys
|
|
import argparse
|
|
from typing import Dict, List, Optional
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
|
|
|
|
class Color:
|
|
"""ANSI color codes for terminal output"""
|
|
HEADER = '\033[95m'
|
|
OKBLUE = '\033[94m'
|
|
OKCYAN = '\033[96m'
|
|
OKGREEN = '\033[92m'
|
|
WARNING = '\033[93m'
|
|
FAIL = '\033[91m'
|
|
ENDC = '\033[0m'
|
|
BOLD = '\033[1m'
|
|
UNDERLINE = '\033[4m'
|
|
|
|
|
|
class TestStatus(Enum):
|
|
"""Test execution status"""
|
|
PENDING = "pending"
|
|
RUNNING = "running"
|
|
PASSED = "passed"
|
|
FAILED = "failed"
|
|
SKIPPED = "skipped"
|
|
|
|
|
|
class CouncilArtifactTester:
|
|
"""Test harness for council artifact generation"""
|
|
|
|
def __init__(self, whoosh_url: str = "http://localhost:8800", verbose: bool = False):
|
|
self.whoosh_url = whoosh_url
|
|
self.verbose = verbose
|
|
self.auth_token = "dev-token"
|
|
self.test_results = []
|
|
self.created_project_id = None
|
|
|
|
def log(self, message: str, level: str = "INFO"):
|
|
"""Log a message with color coding"""
|
|
colors = {
|
|
"INFO": Color.OKBLUE,
|
|
"SUCCESS": Color.OKGREEN,
|
|
"WARNING": Color.WARNING,
|
|
"ERROR": Color.FAIL,
|
|
"HEADER": Color.HEADER
|
|
}
|
|
color = colors.get(level, "")
|
|
timestamp = datetime.now().strftime("%H:%M:%S")
|
|
print(f"{color}[{timestamp}] {level}: {message}{Color.ENDC}")
|
|
|
|
def verbose_log(self, message: str):
|
|
"""Log only if verbose mode is enabled"""
|
|
if self.verbose:
|
|
self.log(message, "INFO")
|
|
|
|
def record_test(self, name: str, status: TestStatus, details: str = ""):
|
|
"""Record test result"""
|
|
self.test_results.append({
|
|
"name": name,
|
|
"status": status.value,
|
|
"details": details,
|
|
"timestamp": datetime.now().isoformat()
|
|
})
|
|
|
|
def make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Optional[Dict]:
|
|
"""Make HTTP request to WHOOSH API"""
|
|
url = f"{self.whoosh_url}{endpoint}"
|
|
headers = {
|
|
"Authorization": f"Bearer {self.auth_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
try:
|
|
if method == "GET":
|
|
response = requests.get(url, headers=headers, timeout=30)
|
|
elif method == "POST":
|
|
response = requests.post(url, headers=headers, json=data, timeout=30)
|
|
elif method == "DELETE":
|
|
response = requests.delete(url, headers=headers, timeout=30)
|
|
else:
|
|
raise ValueError(f"Unsupported HTTP method: {method}")
|
|
|
|
self.verbose_log(f"{method} {endpoint} -> {response.status_code}")
|
|
|
|
if response.status_code in [200, 201, 202]:
|
|
return response.json()
|
|
else:
|
|
self.log(f"Request failed: {response.status_code} - {response.text}", "ERROR")
|
|
return None
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
self.log(f"Request exception: {e}", "ERROR")
|
|
return None
|
|
|
|
def test_1_whoosh_health(self) -> bool:
|
|
"""Test 1: Verify WHOOSH is accessible"""
|
|
self.log("TEST 1: Checking WHOOSH health...", "HEADER")
|
|
|
|
try:
|
|
# WHOOSH doesn't have a dedicated health endpoint, use projects list
|
|
headers = {"Authorization": f"Bearer {self.auth_token}"}
|
|
response = requests.get(f"{self.whoosh_url}/api/v1/projects", headers=headers, timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
project_count = len(data.get("projects", []))
|
|
self.log(f"✓ WHOOSH is healthy and accessible ({project_count} existing projects)", "SUCCESS")
|
|
self.record_test("WHOOSH Health Check", TestStatus.PASSED, f"{project_count} projects")
|
|
return True
|
|
else:
|
|
self.log(f"✗ WHOOSH health check failed: {response.status_code}", "ERROR")
|
|
self.record_test("WHOOSH Health Check", TestStatus.FAILED, f"Status: {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
self.log(f"✗ Cannot reach WHOOSH: {e}", "ERROR")
|
|
self.record_test("WHOOSH Health Check", TestStatus.FAILED, str(e))
|
|
return False
|
|
|
|
def test_2_create_project(self) -> bool:
|
|
"""Test 2: Create a test project"""
|
|
self.log("TEST 2: Creating test project...", "HEADER")
|
|
|
|
# Use an existing GITEA repository for testing
|
|
# Generate unique name by appending timestamp
|
|
import random
|
|
test_suffix = random.randint(1000, 9999)
|
|
test_repo = f"https://gitea.chorus.services/tony/TEST"
|
|
|
|
self.verbose_log(f"Using repository: {test_repo}")
|
|
|
|
project_data = {
|
|
"repository_url": test_repo
|
|
}
|
|
|
|
result = self.make_request("POST", "/api/v1/projects", project_data)
|
|
|
|
if result and "id" in result:
|
|
self.created_project_id = result["id"]
|
|
self.log(f"✓ Project created successfully: {self.created_project_id}", "SUCCESS")
|
|
self.log(f" Name: {result.get('name', 'N/A')}", "INFO")
|
|
self.log(f" Status: {result.get('status', 'unknown')}", "INFO")
|
|
self.verbose_log(f" Project details: {json.dumps(result, indent=2)}")
|
|
self.record_test("Create Project", TestStatus.PASSED, f"Project ID: {self.created_project_id}")
|
|
return True
|
|
else:
|
|
self.log("✗ Failed to create project", "ERROR")
|
|
self.record_test("Create Project", TestStatus.FAILED)
|
|
return False
|
|
|
|
def test_3_verify_council_formation(self) -> bool:
|
|
"""Test 3: Verify council was formed for the project"""
|
|
self.log("TEST 3: Verifying council formation...", "HEADER")
|
|
|
|
if not self.created_project_id:
|
|
self.log("✗ No project ID available", "ERROR")
|
|
self.record_test("Council Formation", TestStatus.SKIPPED, "No project created")
|
|
return False
|
|
|
|
result = self.make_request("GET", f"/api/v1/projects/{self.created_project_id}")
|
|
|
|
if result:
|
|
council_id = result.get("id") # Council ID is same as project ID
|
|
status = result.get("status", "unknown")
|
|
|
|
self.log(f"✓ Council found: {council_id}", "SUCCESS")
|
|
self.log(f" Status: {status}", "INFO")
|
|
self.log(f" Name: {result.get('name', 'N/A')}", "INFO")
|
|
|
|
self.record_test("Council Formation", TestStatus.PASSED, f"Council: {council_id}, Status: {status}")
|
|
return True
|
|
else:
|
|
self.log("✗ Council not found", "ERROR")
|
|
self.record_test("Council Formation", TestStatus.FAILED)
|
|
return False
|
|
|
|
def test_4_wait_for_role_claims(self, max_wait_seconds: int = 30) -> bool:
|
|
"""Test 4: Wait for CHORUS agents to claim roles"""
|
|
self.log(f"TEST 4: Waiting for agent role claims (max {max_wait_seconds}s)...", "HEADER")
|
|
|
|
if not self.created_project_id:
|
|
self.log("✗ No project ID available", "ERROR")
|
|
self.record_test("Role Claims", TestStatus.SKIPPED, "No project created")
|
|
return False
|
|
|
|
start_time = time.time()
|
|
claimed_roles = 0
|
|
|
|
while time.time() - start_time < max_wait_seconds:
|
|
# Check council status
|
|
result = self.make_request("GET", f"/api/v1/projects/{self.created_project_id}")
|
|
|
|
if result:
|
|
# TODO: Add endpoint to get council agents/claims
|
|
# For now, check if status changed to 'active'
|
|
status = result.get("status", "unknown")
|
|
|
|
if status == "active":
|
|
self.log(f"✓ Council activated! All roles claimed", "SUCCESS")
|
|
self.record_test("Role Claims", TestStatus.PASSED, "Council activated")
|
|
return True
|
|
|
|
self.verbose_log(f" Council status: {status}, waiting...")
|
|
|
|
time.sleep(2)
|
|
|
|
elapsed = time.time() - start_time
|
|
self.log(f"⚠ Timeout waiting for role claims ({elapsed:.1f}s)", "WARNING")
|
|
self.log(f" Council may still be forming - this is normal for new deployments", "INFO")
|
|
self.record_test("Role Claims", TestStatus.FAILED, f"Timeout after {elapsed:.1f}s")
|
|
return False
|
|
|
|
def test_5_fetch_artifacts(self) -> bool:
|
|
"""Test 5: Fetch artifacts produced by the council"""
|
|
self.log("TEST 5: Fetching council artifacts...", "HEADER")
|
|
|
|
if not self.created_project_id:
|
|
self.log("✗ No project ID available", "ERROR")
|
|
self.record_test("Fetch Artifacts", TestStatus.SKIPPED, "No project created")
|
|
return False
|
|
|
|
result = self.make_request("GET", f"/api/v1/councils/{self.created_project_id}/artifacts")
|
|
|
|
if result:
|
|
artifacts = result.get("artifacts") or [] # Handle null artifacts
|
|
|
|
if len(artifacts) > 0:
|
|
self.log(f"✓ Found {len(artifacts)} artifact(s)", "SUCCESS")
|
|
|
|
for i, artifact in enumerate(artifacts, 1):
|
|
self.log(f"\n Artifact {i}:", "INFO")
|
|
self.log(f" ID: {artifact.get('id')}", "INFO")
|
|
self.log(f" Type: {artifact.get('artifact_type')}", "INFO")
|
|
self.log(f" Name: {artifact.get('artifact_name')}", "INFO")
|
|
self.log(f" Status: {artifact.get('status')}", "INFO")
|
|
self.log(f" Produced by: {artifact.get('produced_by', 'N/A')}", "INFO")
|
|
self.log(f" Produced at: {artifact.get('produced_at')}", "INFO")
|
|
|
|
if self.verbose and artifact.get('content'):
|
|
content_preview = artifact['content'][:200]
|
|
self.verbose_log(f" Content preview: {content_preview}...")
|
|
|
|
self.record_test("Fetch Artifacts", TestStatus.PASSED, f"Found {len(artifacts)} artifacts")
|
|
return True
|
|
else:
|
|
self.log("⚠ No artifacts found yet", "WARNING")
|
|
self.log(" This is normal - councils need time to produce artifacts", "INFO")
|
|
self.record_test("Fetch Artifacts", TestStatus.FAILED, "No artifacts produced yet")
|
|
return False
|
|
else:
|
|
self.log("✗ Failed to fetch artifacts", "ERROR")
|
|
self.record_test("Fetch Artifacts", TestStatus.FAILED, "API request failed")
|
|
return False
|
|
|
|
def test_6_verify_artifact_content(self) -> bool:
|
|
"""Test 6: Verify artifact content is valid"""
|
|
self.log("TEST 6: Verifying artifact content...", "HEADER")
|
|
|
|
if not self.created_project_id:
|
|
self.log("✗ No project ID available", "ERROR")
|
|
self.record_test("Artifact Content Validation", TestStatus.SKIPPED, "No project created")
|
|
return False
|
|
|
|
result = self.make_request("GET", f"/api/v1/councils/{self.created_project_id}/artifacts")
|
|
|
|
if result:
|
|
artifacts = result.get("artifacts") or [] # Handle null artifacts
|
|
|
|
if len(artifacts) == 0:
|
|
self.log("⚠ No artifacts to validate", "WARNING")
|
|
self.record_test("Artifact Content Validation", TestStatus.SKIPPED, "No artifacts")
|
|
return False
|
|
|
|
valid_count = 0
|
|
for artifact in artifacts:
|
|
has_content = bool(artifact.get('content') or artifact.get('content_json'))
|
|
has_metadata = all([
|
|
artifact.get('artifact_type'),
|
|
artifact.get('artifact_name'),
|
|
artifact.get('status')
|
|
])
|
|
|
|
if has_content and has_metadata:
|
|
valid_count += 1
|
|
self.verbose_log(f" ✓ Artifact {artifact.get('id')} is valid")
|
|
else:
|
|
self.log(f" ✗ Artifact {artifact.get('id')} is incomplete", "WARNING")
|
|
|
|
if valid_count == len(artifacts):
|
|
self.log(f"✓ All {valid_count} artifact(s) are valid", "SUCCESS")
|
|
self.record_test("Artifact Content Validation", TestStatus.PASSED, f"{valid_count}/{len(artifacts)} valid")
|
|
return True
|
|
else:
|
|
self.log(f"⚠ Only {valid_count}/{len(artifacts)} artifact(s) are valid", "WARNING")
|
|
self.record_test("Artifact Content Validation", TestStatus.FAILED, f"{valid_count}/{len(artifacts)} valid")
|
|
return False
|
|
else:
|
|
self.log("✗ Failed to fetch artifacts for validation", "ERROR")
|
|
self.record_test("Artifact Content Validation", TestStatus.FAILED, "API request failed")
|
|
return False
|
|
|
|
def test_7_cleanup(self) -> bool:
|
|
"""Test 7: Cleanup - delete test project"""
|
|
self.log("TEST 7: Cleaning up test project...", "HEADER")
|
|
|
|
if not self.created_project_id:
|
|
self.log("⚠ No project to clean up", "WARNING")
|
|
self.record_test("Cleanup", TestStatus.SKIPPED, "No project created")
|
|
return True
|
|
|
|
result = self.make_request("DELETE", f"/api/v1/projects/{self.created_project_id}")
|
|
|
|
if result:
|
|
self.log(f"✓ Project deleted successfully: {self.created_project_id}", "SUCCESS")
|
|
self.record_test("Cleanup", TestStatus.PASSED)
|
|
return True
|
|
else:
|
|
self.log(f"⚠ Failed to delete project - manual cleanup may be needed", "WARNING")
|
|
self.record_test("Cleanup", TestStatus.FAILED)
|
|
return False
|
|
|
|
def run_all_tests(self, skip_cleanup: bool = False, wait_time: int = 30):
|
|
"""Run all tests in sequence"""
|
|
self.log("\n" + "="*70, "HEADER")
|
|
self.log("COUNCIL ARTIFACT GENERATION TEST SUITE", "HEADER")
|
|
self.log("="*70 + "\n", "HEADER")
|
|
|
|
tests = [
|
|
("WHOOSH Health Check", self.test_1_whoosh_health, []),
|
|
("Create Test Project", self.test_2_create_project, []),
|
|
("Verify Council Formation", self.test_3_verify_council_formation, []),
|
|
("Wait for Role Claims", self.test_4_wait_for_role_claims, [wait_time]),
|
|
("Fetch Artifacts", self.test_5_fetch_artifacts, []),
|
|
("Validate Artifact Content", self.test_6_verify_artifact_content, []),
|
|
]
|
|
|
|
if not skip_cleanup:
|
|
tests.append(("Cleanup Test Data", self.test_7_cleanup, []))
|
|
|
|
passed = 0
|
|
failed = 0
|
|
skipped = 0
|
|
|
|
for name, test_func, args in tests:
|
|
try:
|
|
result = test_func(*args)
|
|
if result:
|
|
passed += 1
|
|
else:
|
|
# Check if it was skipped
|
|
last_result = self.test_results[-1] if self.test_results else None
|
|
if last_result and last_result["status"] == "skipped":
|
|
skipped += 1
|
|
else:
|
|
failed += 1
|
|
except Exception as e:
|
|
self.log(f"✗ Test exception: {e}", "ERROR")
|
|
self.record_test(name, TestStatus.FAILED, str(e))
|
|
failed += 1
|
|
|
|
print() # Blank line between tests
|
|
|
|
# Print summary
|
|
self.print_summary(passed, failed, skipped)
|
|
|
|
def print_summary(self, passed: int, failed: int, skipped: int):
|
|
"""Print test summary"""
|
|
total = passed + failed + skipped
|
|
|
|
self.log("="*70, "HEADER")
|
|
self.log("TEST SUMMARY", "HEADER")
|
|
self.log("="*70, "HEADER")
|
|
|
|
self.log(f"\nTotal Tests: {total}", "INFO")
|
|
self.log(f" Passed: {passed} {Color.OKGREEN}{'✓' * passed}{Color.ENDC}", "SUCCESS")
|
|
if failed > 0:
|
|
self.log(f" Failed: {failed} {Color.FAIL}{'✗' * failed}{Color.ENDC}", "ERROR")
|
|
if skipped > 0:
|
|
self.log(f" Skipped: {skipped} {Color.WARNING}{'○' * skipped}{Color.ENDC}", "WARNING")
|
|
|
|
success_rate = (passed / total * 100) if total > 0 else 0
|
|
self.log(f"\nSuccess Rate: {success_rate:.1f}%", "INFO")
|
|
|
|
if self.created_project_id:
|
|
self.log(f"\nTest Project ID: {self.created_project_id}", "INFO")
|
|
|
|
# Detailed results
|
|
if self.verbose:
|
|
self.log("\nDetailed Results:", "HEADER")
|
|
for result in self.test_results:
|
|
status_color = {
|
|
"passed": Color.OKGREEN,
|
|
"failed": Color.FAIL,
|
|
"skipped": Color.WARNING
|
|
}.get(result["status"], "")
|
|
|
|
self.log(f" {result['name']}: {status_color}{result['status'].upper()}{Color.ENDC}", "INFO")
|
|
if result.get("details"):
|
|
self.log(f" {result['details']}", "INFO")
|
|
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
parser = argparse.ArgumentParser(description="Test council artifact generation")
|
|
parser.add_argument("--whoosh-url", default="http://localhost:8800",
|
|
help="WHOOSH base URL (default: http://localhost:8800)")
|
|
parser.add_argument("--verbose", "-v", action="store_true",
|
|
help="Enable verbose output")
|
|
parser.add_argument("--skip-cleanup", action="store_true",
|
|
help="Skip cleanup step (leave test project)")
|
|
parser.add_argument("--wait-time", type=int, default=30,
|
|
help="Seconds to wait for role claims (default: 30)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
tester = CouncilArtifactTester(whoosh_url=args.whoosh_url, verbose=args.verbose)
|
|
tester.run_all_tests(skip_cleanup=args.skip_cleanup, wait_time=args.wait_time)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|