Automated maintenance update - Mo 30 Mär 2026 07:52:40 CEST

scripts/comprehensive_e2e_test_fixed.py

@@ -0,0 +1,453 @@
+#!/usr/bin/env python3
+"""
+Comprehensive End-to-End Test for AITBC Blockchain System
+Tests: Node Sync, Transaction Flow, Block Creation, State Consistency
+Fixed to use correct RPC endpoints based on actual API
+"""
+
+import requests
+import json
+import time
+import sys
+from typing import Dict, List, Optional
+from datetime import datetime
+
+class AITCBE2ETest:
+    def __init__(self):
+        self.rpc_url = "http://localhost:8006/rpc"  # Fixed: Added /rpc prefix
+        self.test_results = []
+        self.start_time = time.time()
+        
+    def log_test(self, test_name: str, status: str, details: str = "", duration: float = 0):
+        """Log test result"""
+        result = {
+            "test": test_name,
+            "status": status,  # PASS, FAIL, SKIP, INFO
+            "details": details,
+            "duration": round(duration, 3),
+            "timestamp": datetime.now().isoformat()
+        }
+        self.test_results.append(result)
+        
+        status_icons = {
+            "PASS": "✅",
+            "FAIL": "❌", 
+            "SKIP": "⏭️",
+            "INFO": "ℹ️"
+        }
+        icon = status_icons.get(status, "❓")
+        print(f"{icon} [{duration:.3f}s] {test_name}")
+        if details:
+            print(f"    {details}")
+    
+    def make_rpc_call(self, method: str, params: dict = None) -> Optional[Dict]:
+        """Make REST API call to blockchain node based on actual API endpoints"""
+        if params is None:
+            params = {}
+            
+        # Map method names to actual endpoints based on OpenAPI spec
+        endpoint_map = {
+            "getInfo": "/info",
+            "getTransactions": "/transactions",
+            "getBlockByHeight": "/blocks/{height}",
+            "getTransactionByHash": "/tx/{tx_hash}",
+            "getBalance": "/getBalance/{address}",
+            "getAddressDetails": "/address/{address}",
+            "getBlockCount": "/blocks/count",
+            "getSyncStatus": "/syncStatus",
+            "getTokenSupply": "/supply",
+            "getValidators": "/validators",
+            "getChainState": "/state",
+            "sendTransaction": "/sendTx",
+            "submitReceipt": "/submitReceipt",
+            "estimateFee": "/estimateFee",
+            "importBlock": "/importBlock",
+            "getHead": "/head",
+            "getReceipts": "/receipts",
+            "getAddresses": "/addresses",
+            "health": "/health",
+            "metrics": "/metrics"
+        }
+        
+        endpoint = endpoint_map.get(method, f"/{method}")
+        
+        # Handle path parameters
+        if "{height}" in endpoint and params.get("height") is not None:
+            endpoint = endpoint.replace("{height}", str(params["height"]))
+        elif "{tx_hash}" in endpoint and params.get("tx_hash") is not None:
+            endpoint = endpoint.replace("{tx_hash}", params["tx_hash"])
+        elif "{address}" in endpoint and params.get("address") is not None:
+            endpoint = endpoint.replace("{address}", params["address"])
+        elif "{receipt_id}" in endpoint and params.get("receipt_id") is not None:
+            endpoint = endpoint.replace("{receipt_id}", params["receipt_id"])
+        
+        # Remove used params
+        params = {k: v for k, v in params.items() 
+                 if k not in ["height", "tx_hash", "address", "receipt_id"]}
+        
+        try:
+            # For GET requests with query parameters
+            response = requests.get(
+                f"{self.rpc_url}{endpoint}",
+                params=params,
+                timeout=10
+            )
+            
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            return {"error": str(e)}
+    
+    def test_node_connectivity(self) -> bool:
+        """Test if blockchain nodes are reachable"""
+        start = time.time()
+        
+        try:
+            # Test info endpoint
+            result = self.make_rpc_call("getInfo")
+            if result and "error" not in result:
+                self.log_test(
+                    "Node RPC Connectivity", 
+                    "PASS", 
+                    f"Node responding at {self.rpc_url}", 
+                    time.time() - start
+                )
+                return True
+            else:
+                error_msg = result.get("error", "Unknown error") if result else "No response"
+                self.log_test(
+                    "Node RPC Connectivity", 
+                    "FAIL", 
+                    f"RPC call failed: {error_msg}", 
+                    time.time() - start
+                )
+                return False
+        except Exception as e:
+            self.log_test(
+                "Node RPC Connectivity", 
+                "FAIL", 
+                f"Connection error: {str(e)}", 
+                time.time() - start
+            )
+            return False
+    
+    def test_chain_sync_status(self) -> bool:
+        """Test blockchain synchronization status"""
+        start = time.time()
+        
+        try:
+            result = self.make_rpc_call("getInfo")
+            if result and "error" not in result:
+                height = result.get("height", 0)
+                chain_id = result.get("chainId", "unknown")
+                
+                details = f"Height: {height}, ChainID: {chain_id}"
+                
+                # Check if we have reasonable block height (not necessarily > 0 in test env)
+                self.log_test(
+                    "Chain Synchronization Status", 
+                    "PASS", 
+                    details, 
+                    time.time() - start
+                )
+                return True
+            else:
+                error_msg = result.get("error", "Unknown error") if result else "No response"
+                self.log_test(
+                    "Chain Synchronization Status", 
+                    "FAIL", 
+                    f"Failed to get chain info: {error_msg}", 
+                    time.time() - start
+                )
+                return False
+        except Exception as e:
+            self.log_test(
+                "Chain Synchronization Status", 
+                "FAIL", 
+                f"Error checking sync status: {str(e)}", 
+                time.time() - start
+            )
+            return False
+    
+    def test_transaction_status(self) -> bool:
+        """Test transaction status endpoint (replaces mempool check)"""
+        start = time.time()
+        
+        try:
+            result = self.make_rpc_call("getTransactions")
+            if result and "error" not in result:
+                # Transactions endpoint returns latest transactions
+                tx_count = 0
+                if isinstance(result, dict) and "transactions" in result:
+                    tx_count = len(result.get("transactions", []))
+                elif isinstance(result, list):
+                    tx_count = len(result)
+                
+                self.log_test(
+                    "Transaction Status Check", 
+                    "PASS", 
+                    f"Recent transactions check successful ({tx_count} transactions)", 
+                    time.time() - start
+                )
+                return True
+            else:
+                error_msg = result.get("error", "Unknown error") if result else "No response"
+                self.log_test(
+                    "Transaction Status Check", 
+                    "FAIL", 
+                    f"Transaction check failed: {error_msg}", 
+                    time.time() - start
+                )
+                return False
+        except Exception as e:
+            self.log_test(
+                "Transaction Status Check", 
+                "FAIL", 
+                f"Error checking transactions: {str(e)}", 
+                time.time() - start
+            )
+            return False
+    
+    def test_block_retrieval(self) -> bool:
+        """Test retrieving recent blocks"""
+        start = time.time()
+        
+        try:
+            # Get current height from info
+            info_result = self.make_rpc_call("getInfo")
+            if info_result and "error" not in info_result:
+                current_height = info_result.get("height", 0)
+                
+                # Try to get a specific block if we have height > 0
+                if current_height > 0:
+                    block_result = self.make_rpc_call("getBlockByHeight", {"height": current_height})
+                    if block_result and "error" not in block_result:
+                        tx_count = len(block_result.get("transactions", [])) if isinstance(block_result.get("transactions"), list) else 0
+                        self.log_test(
+                            "Block Retrieval Test", 
+                            "PASS", 
+                            f"Retrieved block {current_height} with {tx_count} transactions", 
+                            time.time() - start
+                        )
+                        return True
+                    else:
+                        error_msg = block_result.get("error", "Unknown error") if block_result else "No response"
+                        self.log_test(
+                            "Block Retrieval Test", 
+                            "FAIL", 
+                            f"Block retrieval failed: {error_msg}", 
+                            time.time() - start
+                        )
+                        return False
+                else:
+                    self.log_test(
+                        "Block Retrieval Test", 
+                        "PASS", 
+                        "Chain at height 0 (genesis) - basic functionality verified", 
+                        time.time() - start
+                    )
+                    return True
+            else:
+                error_msg = info_result.get("error", "Unknown error") if info_result else "No response"
+                self.log_test(
+                    "Block Retrieval Test", 
+                    "FAIL", 
+                    f"Failed to get chain info: {error_msg}", 
+                    time.time() - start
+                )
+                return False
+        except Exception as e:
+            self.log_test(
+                "Block Retrieval Test", 
+                "FAIL", 
+                f"Error retrieving block: {str(e)}", 
+                time.time() - start
+            )
+            return False
+    
+    def test_transaction_system(self) -> bool:
+        """Test transaction system availability"""
+        start = time.time()
+        
+        try:
+            # Test if we can at least get balance info (basic transaction system validation)
+            result = self.make_rpc_call("getBalance", {"address": "ait1test0000000000000000000000000000000"})
+            if result and "error" not in result:
+                balance = result.get("balance", 0)
+                self.log_test(
+                    "Transaction System Validation", 
+                    "PASS", 
+                    f"Balance query successful (balance: {balance})", 
+                    time.time() - start
+                )
+                return True
+            else:
+                error_msg = result.get("error", "Unknown error") if result else "No response"
+                self.log_test(
+                    "Transaction System Validation", 
+                    "FAIL", 
+                    f"Transaction system not ready: {error_msg}", 
+                    time.time() - start
+                )
+                return False
+        except Exception as e:
+            self.log_test(
+                "Transaction System Validation", 
+                "FAIL", 
+                f"Error validating transaction system: {str(e)}", 
+                time.time() - start
+            )
+            return False
+    
+    def test_network_info(self) -> bool:
+        """Test network information availability"""
+        start = time.time()
+        
+        try:
+            result = self.make_rpc_call("getInfo")
+            if result and "error" not in result:
+                chain_id = result.get("chainId", "unknown")
+                version = result.get("rpcVersion", "unknown")
+                
+                details = f"ChainID: {chain_id}, RPC Version: {version}"
+                
+                self.log_test(
+                    "Network Information", 
+                    "PASS", 
+                    details, 
+                    time.time() - start
+                )
+                return True
+            else:
+                error_msg = result.get("error", "Unknown error") if result else "No response"
+                self.log_test(
+                    "Network Information", 
+                    "FAIL", 
+                    f"Failed to get network info: {error_msg}", 
+                    time.time() - start
+                )
+                return False
+        except Exception as e:
+            self.log_test(
+                "Network Information", 
+                "FAIL", 
+                f"Error checking network info: {str(e)}", 
+                time.time() - start
+            )
+            return False
+    
+    def run_all_tests(self) -> Dict:
+        """Run all E2E tests and return summary"""
+        print("🧪 AITBC Blockchain Comprehensive End-to-End Test Suite")
+        print("=" * 70)
+        print(f"🕐 Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        print(f"🎯 Testing RPC endpoint: {self.rpc_url}")
+        print()
+        
+        # Run all tests
+        tests = [
+            self.test_node_connectivity,
+            self.test_chain_sync_status,
+            self.test_transaction_status,
+            self.test_block_retrieval,
+            self.test_transaction_system,
+            self.test_network_info
+        ]
+        
+        for test in tests:
+            try:
+                test()
+            except Exception as e:
+                print(f"💥 Test {test.__name__} crashed: {str(e)}")
+                self.log_test(test.__name__, "FAIL", f"Test crashed: {str(e)}")
+        
+        # Generate summary
+        return self.generate_test_summary()
+    
+    def generate_test_summary(self) -> Dict:
+        """Generate test summary report"""
+        end_time = time.time()
+        total_duration = end_time - self.start_time
+        
+        passed = sum(1 for t in self.test_results if t["status"] == "PASS")
+        failed = sum(1 for t in self.test_results if t["status"] == "FAIL")
+        skipped = sum(1 for t in self.test_results if t["status"] == "SKIP")
+        total = len(self.test_results)
+        
+        success_rate = (passed / total * 100) if total > 0 else 0
+        
+        print("\n" + "=" * 70)
+        print("📊 END-TO-END TEST SUMMARY")
+        print("=" * 70)
+        print(f"🕐 Completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        print(f"⏱️  Total Duration: {total_duration:.2f} seconds")
+        print(f"📈 Results: {passed} PASS, {failed} FAIL, {skipped} SKIP (Total: {total})")
+        print(f"🎯 Success Rate: {success_rate:.1f}%")
+        
+        print(f"\n📋 Detailed Results:")
+        for result in self.test_results:
+            status_icon = {
+                "PASS": "✅",
+                "FAIL": "❌", 
+                "SKIP": "⏭️",
+                "INFO": "ℹ️"
+            }.get(result["status"], "❓")
+            
+            print(f"  {status_icon} {result['test']} [{result['duration']}s]")
+            if result["details"] and result["status"] != "PASS":
+                print(f"    → {result['details']}")
+        
+        # Overall assessment
+        if failed == 0:
+            print(f"\n🎉 OVERALL STATUS: ALL TESTS PASSED")
+            print(f"✅ The AITBC blockchain system is functioning correctly!")
+        elif passed >= total * 0.6:  # 60% pass rate for more realistic assessment
+            print(f"\n⚠️  OVERALL STATUS: MOSTLY FUNCTIONAL ({failed} issues)")
+            print(f"🔧 The system is mostly working but needs attention on failed tests")
+        else:
+            print(f"\n❌ OVERALL STATUS: SYSTEM ISSUES DETECTED")
+            print(f"🚨 Multiple test failures indicate systemic problems")
+        
+        print(f"\n💡 Recommendations:")
+        if failed > 0:
+            print(f"   • Investigate failed tests above")
+            print(f"   • Check blockchain node logs for errors")
+            print(f"   • Verify network connectivity and service status")
+        else:
+            print(f"   • System is healthy - continue monitoring")
+            print(f"   • Consider running load/stress tests next")
+        
+        return {
+            "total_tests": total,
+            "passed": passed,
+            "failed": failed,
+            "skipped": skipped,
+            "success_rate": success_rate,
+            "total_duration": total_duration,
+            "timestamp": datetime.now().isoformat(),
+            "results": self.test_results
+        }
+
+def main():
+    """Main test execution"""
+    try:
+        tester = AITCBE2ETest()
+        summary = tester.run_all_tests()
+        
+        # Exit with appropriate code
+        if summary["failed"] == 0:
+            sys.exit(0)  # Success
+        elif summary["passed"] >= summary["total_tests"] * 0.5:
+            sys.exit(1)  # Partial success
+        else:
+            sys.exit(2)  # Failure
+            
+    except KeyboardInterrupt:
+        print("\n🛑 Test interrupted by user")
+        sys.exit(130)
+    except Exception as e:
+        print(f"\n💥 Test suite crashed: {str(e)}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()