#!/usr/bin/env python3 """ Test Composer Client for Debian Forge This script tests the composer client functionality for build submission, status monitoring, and build management. """ import json import os import sys from pathlib import Path # Add current directory to Python path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) def test_composer_client_import(): """Test importing the composer client""" print("Testing composer client import...") try: # Import from current directory from composer_client import ComposerClient, BuildRequest, BuildStatus, DebianAtomicBuilder print(" ✅ Composer client imported successfully") return True except ImportError as e: print(f" ❌ Failed to import composer client: {e}") return False def test_build_request_dataclass(): """Test BuildRequest dataclass""" print("\nTesting BuildRequest dataclass...") try: from composer_client import BuildRequest # Test basic creation request = BuildRequest( blueprint="debian-atomic-base", target="qcow2", architecture="amd64" ) if request.blueprint != "debian-atomic-base": print(" ❌ Blueprint field not set correctly") return False if request.target != "qcow2": print(" ❌ Target field not set correctly") return False if request.architecture != "amd64": print(" ❌ Architecture field not set correctly") return False # Test default values if request.compose_type != "debian-atomic": print(" ❌ Default compose_type not set correctly") return False if request.priority != "normal": print(" ❌ Default priority not set correctly") return False print(" ✅ BuildRequest dataclass works correctly") return True except Exception as e: print(f" ❌ BuildRequest test failed: {e}") return False def test_build_status_dataclass(): """Test BuildStatus dataclass""" print("\nTesting BuildStatus dataclass...") try: from composer_client import BuildStatus # Test basic creation status = BuildStatus( build_id="test-123", status="RUNNING", created_at="2024-12-19T10:00:00Z", blueprint="debian-atomic-base", target="qcow2", architecture="amd64" ) if status.build_id != "test-123": print(" ❌ Build ID field not set correctly") return False if status.status != "RUNNING": print(" ❌ Status field not set correctly") return False print(" ✅ BuildStatus dataclass works correctly") return True except Exception as e: print(f" ❌ BuildStatus test failed: {e}") return False def test_composer_client_initialization(): """Test ComposerClient initialization""" print("\nTesting ComposerClient initialization...") try: from composer_client import ComposerClient # Test default initialization client = ComposerClient() if client.base_url != "http://localhost:8700": print(" ❌ Default base_url not set correctly") return False if client.api_version != "v1": print(" ❌ Default api_version not set correctly") return False # Test custom initialization client = ComposerClient("http://example.com:9000", "v2") if client.base_url != "http://example.com:9000": print(" ❌ Custom base_url not set correctly") return False if client.api_version != "v2": print(" ❌ Custom api_version not set correctly") return False print(" ✅ ComposerClient initialization works correctly") return True except Exception as e: print(f" ❌ ComposerClient initialization test failed: {e}") return False def test_debian_atomic_builder(): """Test DebianAtomicBuilder class""" print("\nTesting DebianAtomicBuilder...") try: from composer_client import ComposerClient, DebianAtomicBuilder # Create a mock client (we won't actually connect) client = ComposerClient() builder = DebianAtomicBuilder(client) # Test builder creation if not hasattr(builder, 'client'): print(" ❌ Builder missing client attribute") return False # Test method availability required_methods = ['build_base_image', 'build_workstation_image', 'build_server_image'] for method in required_methods: if not hasattr(builder, method): print(f" ❌ Builder missing method: {method}") return False print(" ✅ DebianAtomicBuilder works correctly") return True except Exception as e: print(f" ❌ DebianAtomicBuilder test failed: {e}") return False def test_blueprint_validation(): """Test blueprint validation logic""" print("\nTesting blueprint validation...") # Check if blueprint files exist blueprint_dir = Path("blueprints") if not blueprint_dir.exists(): print(" ❌ Blueprint directory not found") return False blueprints = ["debian-atomic-base.json", "debian-atomic-workstation.json", "debian-atomic-server.json"] for blueprint_file in blueprints: blueprint_path = blueprint_dir / blueprint_file if not blueprint_path.exists(): print(f" ❌ Blueprint file not found: {blueprint_file}") return False try: with open(blueprint_path, 'r') as f: blueprint = json.load(f) # Validate blueprint structure required_fields = ["name", "description", "version", "packages"] for field in required_fields: if field not in blueprint: print(f" ❌ {blueprint_file} missing field: {field}") return False # Validate packages if not isinstance(blueprint["packages"], list): print(f" ❌ {blueprint_file} packages must be a list") return False for package in blueprint["packages"]: if "name" not in package: print(f" ❌ {blueprint_file} package missing name") return False print(f" ✅ {blueprint_file} validation passed") except json.JSONDecodeError as e: print(f" ❌ {blueprint_file} invalid JSON: {e}") return False except Exception as e: print(f" ❌ {blueprint_file} validation error: {e}") return False return True def test_api_endpoint_structure(): """Test API endpoint structure""" print("\nTesting API endpoint structure...") try: from composer_client import ComposerClient client = ComposerClient() # Test endpoint construction test_endpoints = [ ("blueprints/new", "POST"), ("blueprints/info/test", "GET"), ("blueprints/list", "GET"), ("compose", "POST"), ("compose/status/test", "GET"), ("compose/list", "GET"), ("compose/cancel/test", "DELETE"), ("compose/logs/test", "GET"), ("compose/image/test", "GET") ] for endpoint, method in test_endpoints: # This tests that the endpoint structure is valid # We can't actually make requests without a running composer if not endpoint.startswith(('blueprints/', 'compose/')): print(f" ❌ Invalid endpoint structure: {endpoint}") return False print(" ✅ API endpoint structure is valid") return True except Exception as e: print(f" ❌ API endpoint structure test failed: {e}") return False def test_error_handling(): """Test error handling in composer client""" print("\nTesting error handling...") try: from composer_client import ComposerClient client = ComposerClient() # Test invalid HTTP method try: client._make_request("INVALID", "test") print(" ❌ Should have raised error for invalid HTTP method") return False except ValueError: # Expected error pass print(" ✅ Error handling works correctly") return True except Exception as e: print(f" ❌ Error handling test failed: {e}") return False def main(): """Main test function""" print("Composer Client Test for Debian Forge") print("=" * 50) tests = [ ("Composer Client Import", test_composer_client_import), ("BuildRequest Dataclass", test_build_request_dataclass), ("BuildStatus Dataclass", test_build_status_dataclass), ("ComposerClient Initialization", test_composer_client_initialization), ("DebianAtomicBuilder", test_debian_atomic_builder), ("Blueprint Validation", test_blueprint_validation), ("API Endpoint Structure", test_api_endpoint_structure), ("Error Handling", test_error_handling) ] results = [] for test_name, test_func in tests: try: result = test_func() results.append((test_name, result)) except Exception as e: print(f" ❌ {test_name} test failed with exception: {e}") results.append((test_name, False)) # Summary print("\n" + "=" * 50) print("TEST SUMMARY") print("=" * 50) passed = 0 total = len(results) for test_name, result in results: status = "✅ PASS" if result else "❌ FAIL" print(f"{test_name}: {status}") if result: passed += 1 print(f"\nOverall: {passed}/{total} tests passed") if passed == total: print("🎉 All tests passed! Composer client is ready for use.") return 0 else: print("⚠️ Some tests failed. Please review the issues above.") return 1 if __name__ == '__main__': sys.exit(main())