#!/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()