tony/WHOOSH
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix Docker Swarm discovery network name mismatch

Browse Source 
- 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>
This commit is contained in:
Claude Code
2025-10-10 10:35:25 +11:00
parent 2826b28645
commit 9aeaa433fc
36 changed files with 4721 additions and 2213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
tests/README.md Normal file
View File

@@ -0,0 +1,290 @@
# WHOOSH Council Artifact Tests
## Overview
This directory contains integration tests for verifying that WHOOSH councils are properly generating project artifacts through the CHORUS agent collaboration system.
## Test Coverage
The `test_council_artifacts.py` script performs end-to-end testing of:
1. **WHOOSH Health Check** - Verifies WHOOSH API is accessible
2. **Project Creation** - Creates a test project with council formation
3. **Council Formation** - Verifies council was created with correct structure
4. **Role Claiming** - Waits for CHORUS agents to claim council roles
5. **Artifact Fetching** - Retrieves artifacts produced by the council
6. **Content Validation** - Verifies artifact content is complete and valid
7. **Cleanup** - Removes test data (optional)
## Requirements
```bash
pip install requests
```
Or install from requirements file:
```bash
pip install -r requirements.txt
```
## Usage
### Basic Test Run
```bash
python test_council_artifacts.py
```
### With Verbose Output
```bash
python test_council_artifacts.py --verbose
```
### Custom WHOOSH URL
```bash
python test_council_artifacts.py --whoosh-url http://whoosh.example.com:8080
```
### Extended Wait Time for Role Claims
```bash
python test_council_artifacts.py --wait-time 60
```
### Skip Cleanup (Keep Test Project)
```bash
python test_council_artifacts.py --skip-cleanup
```
### Full Example
```bash
python test_council_artifacts.py \
--whoosh-url http://localhost:8800 \
--verbose \
--wait-time 45 \
--skip-cleanup
```
## Command-Line Options
| Option | Description | Default |
|--------|-------------|---------|
| `--whoosh-url URL` | WHOOSH base URL | `http://localhost:8800` |
| `--verbose`, `-v` | Enable detailed output | `False` |
| `--skip-cleanup` | Don't delete test project | `False` |
| `--wait-time SECONDS` | Max wait for role claims | `30` |
## Expected Output
### Successful Test Run
```
======================================================================
COUNCIL ARTIFACT GENERATION TEST SUITE
======================================================================
[14:23:45] HEADER: TEST 1: Checking WHOOSH health...
[14:23:45] SUCCESS: ✓ WHOOSH is healthy and accessible
[14:23:45] HEADER: TEST 2: Creating test project...
[14:23:46] SUCCESS: ✓ Project created successfully: abc-123-def
[14:23:46] INFO: Council ID: abc-123-def
[14:23:46] HEADER: TEST 3: Verifying council formation...
[14:23:46] SUCCESS: ✓ Council found: abc-123-def
[14:23:46] INFO: Status: forming
[14:23:46] HEADER: TEST 4: Waiting for agent role claims (max 30s)...
[14:24:15] SUCCESS: ✓ Council activated! All roles claimed
[14:24:15] HEADER: TEST 5: Fetching council artifacts...
[14:24:15] SUCCESS: ✓ Found 3 artifact(s)
Artifact 1:
ID: art-001
Type: architecture_document
Name: System Architecture Design
Status: approved
Produced by: chorus-agent-002
Produced at: 2025-10-06T14:24:10Z
[14:24:15] HEADER: TEST 6: Verifying artifact content...
[14:24:15] SUCCESS: ✓ All 3 artifact(s) are valid
[14:24:15] HEADER: TEST 7: Cleaning up test project...
[14:24:16] SUCCESS: ✓ Project deleted successfully: abc-123-def
======================================================================
TEST SUMMARY
======================================================================
Total Tests: 7
Passed: 7 ✓✓✓✓✓✓✓
Success Rate: 100.0%
```
### Test Failure Example
```
[14:23:46] HEADER: TEST 5: Fetching council artifacts...
[14:23:46] WARNING: ⚠ No artifacts found yet
[14:23:46] INFO: This is normal - councils need time to produce artifacts
======================================================================
TEST SUMMARY
======================================================================
Total Tests: 7
Passed: 6 ✓✓✓✓✓✓
Failed: 1 ✗
Success Rate: 85.7%
```
## Test Scenarios
### Scenario 1: Fresh Deployment Test
Tests a newly deployed WHOOSH/CHORUS system:
```bash
python test_council_artifacts.py --wait-time 60 --verbose
```
**Expected**: Role claiming may take longer on first run as agents initialize.
### Scenario 2: Production Readiness Test
Quick validation that production system is working:
```bash
python test_council_artifacts.py --whoosh-url https://whoosh.production.com
```
**Expected**: All tests should pass in < 1 minute.
### Scenario 3: Development/Debug Test
Keep test project for manual inspection:
```bash
python test_council_artifacts.py --skip-cleanup --verbose
```
**Expected**: Project remains in database for debugging.
## Troubleshooting
### Test 1 Fails: WHOOSH Not Accessible
**Problem**: Cannot connect to WHOOSH API
**Solutions**:
- Verify WHOOSH is running: `docker service ps CHORUS_whoosh`
- Check URL is correct: `--whoosh-url http://localhost:8800`
- Check firewall/network settings
### Test 4 Fails: Role Claims Timeout
**Problem**: CHORUS agents not claiming roles
**Solutions**:
- Increase wait time: `--wait-time 60`
- Check CHORUS agents are running: `docker service ps CHORUS_chorus`
- Check agent logs: `docker service logs CHORUS_chorus`
- Verify P2P discovery is working
### Test 5 Fails: No Artifacts Found
**Problem**: Council formed but no artifacts produced
**Solutions**:
- This is expected initially - councils need time to collaborate
- Check council status in UI or database
- Verify CHORUS agents have proper capabilities configured
- Check agent logs for artifact production errors
## Integration with CI/CD
### GitHub Actions Example
```yaml
name: Test Council Artifacts
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Start WHOOSH
run: docker-compose up -d
- name: Wait for services
run: sleep 30
- name: Run tests
run: |
cd tests
python test_council_artifacts.py --verbose
```
### Jenkins Example
```groovy
stage('Test Council Artifacts') {
steps {
sh '''
cd tests
python test_council_artifacts.py \
--whoosh-url http://whoosh-test:8080 \
--wait-time 60 \
--verbose
'''
}
}
```
## Test Data
The test creates a temporary project using:
- **Repository**: `https://gitea.chorus.services/tony/test-council-project`
- **Project Name**: Auto-generated from repository
- **Council**: Automatically formed with 8 core roles
All test data is cleaned up unless `--skip-cleanup` is specified.
## Exit Codes
- `0` - All tests passed
- `1` - One or more tests failed
- Non-zero - System error occurred
## Logging
Test logs include:
- Timestamp for each action
- Color-coded output (INFO/SUCCESS/WARNING/ERROR)
- Request/response details in verbose mode
- Complete artifact metadata
## Future Enhancements
- [ ] Test multiple concurrent project creations
- [ ] Verify artifact versioning
- [ ] Test artifact approval workflow
- [ ] Performance benchmarking
- [ ] Load testing with many councils
- [ ] WebSocket event stream validation
- [ ] Agent collaboration pattern verification
## Support
For issues or questions:
- Check logs: `docker service logs CHORUS_whoosh`
- Review integration status: `COUNCIL_AGENT_INTEGRATION_STATUS.md`
- Open issue on project repository

BIN
tests/__pycache__/test_council_artifacts.cpython-312.pyc Normal file
View File

Binary file not shown.

144
tests/quick_health_check.py Executable file
View File

@@ -0,0 +1,144 @@
#!/usr/bin/env python3
"""
Quick Health Check for WHOOSH Council System
Performs rapid health checks on WHOOSH and CHORUS services.
Useful for monitoring and CI/CD pipelines.
Usage:
python quick_health_check.py
python quick_health_check.py --json # JSON output for monitoring tools
"""
import requests
import sys
import argparse
import json
from datetime import datetime
def check_whoosh(url: str = "http://localhost:8800") -> dict:
"""Check WHOOSH API health"""
try:
response = requests.get(f"{url}/api/health", timeout=5)
return {
"service": "WHOOSH",
"status": "healthy" if response.status_code == 200 else "unhealthy",
"status_code": response.status_code,
"url": url,
"error": None
}
except Exception as e:
return {
"service": "WHOOSH",
"status": "unreachable",
"status_code": None,
"url": url,
"error": str(e)
}
def check_project_count(url: str = "http://localhost:8800") -> dict:
"""Check how many projects exist"""
try:
headers = {"Authorization": "Bearer dev-token"}
response = requests.get(f"{url}/api/v1/projects", headers=headers, timeout=5)
if response.status_code == 200:
data = response.json()
projects = data.get("projects", [])
return {
"metric": "projects",
"count": len(projects),
"status": "ok",
"error": None
}
else:
return {
"metric": "projects",
"count": 0,
"status": "error",
"error": f"HTTP {response.status_code}"
}
except Exception as e:
return {
"metric": "projects",
"count": 0,
"status": "error",
"error": str(e)
}
def check_p2p_discovery(url: str = "http://localhost:8800") -> dict:
"""Check P2P discovery is finding agents"""
# Note: This would require a dedicated endpoint
# For now, we'll return a placeholder
return {
"metric": "p2p_discovery",
"status": "not_implemented",
"note": "Add /api/v1/p2p/agents endpoint to WHOOSH"
}
def main():
parser = argparse.ArgumentParser(description="Quick health check for WHOOSH")
parser.add_argument("--whoosh-url", default="http://localhost:8800",
help="WHOOSH base URL")
parser.add_argument("--json", action="store_true",
help="Output JSON for monitoring tools")
args = parser.parse_args()
# Perform checks
results = {
"timestamp": datetime.now().isoformat(),
"checks": {
"whoosh": check_whoosh(args.whoosh_url),
"projects": check_project_count(args.whoosh_url),
"p2p": check_p2p_discovery(args.whoosh_url)
}
}
# Calculate overall health
whoosh_healthy = results["checks"]["whoosh"]["status"] == "healthy"
projects_ok = results["checks"]["projects"]["status"] == "ok"
results["overall_status"] = "healthy" if whoosh_healthy and projects_ok else "degraded"
if args.json:
# JSON output for monitoring
print(json.dumps(results, indent=2))
sys.exit(0 if results["overall_status"] == "healthy" else 1)
else:
# Human-readable output
print("="*60)
print("WHOOSH SYSTEM HEALTH CHECK")
print("="*60)
print(f"Timestamp: {results['timestamp']}\n")
# WHOOSH Service
whoosh = results["checks"]["whoosh"]
status_symbol = "✓" if whoosh["status"] == "healthy" else "✗"
print(f"{status_symbol} WHOOSH API: {whoosh['status']}")
if whoosh["error"]:
print(f" Error: {whoosh['error']}")
print(f" URL: {whoosh['url']}\n")
# Projects
projects = results["checks"]["projects"]
print(f"📊 Projects: {projects['count']}")
if projects["error"]:
print(f" Error: {projects['error']}")
print()
# Overall
print("="*60)
overall = results["overall_status"]
print(f"Overall Status: {overall.upper()}")
print("="*60)
sys.exit(0 if overall == "healthy" else 1)
if __name__ == "__main__":
main()

2
tests/requirements.txt Normal file
View File

@@ -0,0 +1,2 @@
# Python dependencies for WHOOSH integration tests
requests>=2.31.0

440
tests/test_council_artifacts.py Executable file
View File

@@ -0,0 +1,440 @@
#!/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()