#!/usr/bin/env python3 """ Test script to verify elections blockchain integration. This script tests: 1. Blockchain recording when elections are created 2. Blockchain verification endpoints 3. Hash chain integrity 4. Tamper detection Run after backend is started: python3 test_blockchain_election.py """ import requests import json import time from datetime import datetime, timedelta BASE_URL = "http://localhost:8000" def test_blockchain_endpoint(): """Test GET /api/elections/blockchain endpoint""" print("\n" + "="*60) print("TEST 1: Get Elections Blockchain") print("="*60) try: response = requests.get(f"{BASE_URL}/api/elections/blockchain") data = response.json() if "blocks" in data: print(f"✓ Blockchain endpoint working") print(f" Total blocks: {data['verification']['total_blocks']}") print(f" Chain valid: {data['verification']['chain_valid']}") if data['blocks']: print(f"\nFirst block:") block = data['blocks'][0] print(f" Election ID: {block['election_id']}") print(f" Election name: {block['election_name']}") print(f" Candidates: {block['candidates_count']}") print(f" Block hash: {block['block_hash'][:32]}...") print(f" Signature: {block['signature'][:32]}...") return True else: print("⚠ No blocks in blockchain yet") print(" Elections may not have been initialized") return False else: print(f"✗ Unexpected response format: {data}") return False except requests.exceptions.ConnectionError: print("✗ Cannot connect to backend") print(f" Is it running on {BASE_URL}?") return False except Exception as e: print(f"✗ Error: {e}") return False def test_election_verification(election_id=1): """Test GET /api/elections/{election_id}/blockchain-verify endpoint""" print("\n" + "="*60) print(f"TEST 2: Verify Election {election_id} Blockchain Integrity") print("="*60) try: response = requests.get(f"{BASE_URL}/api/elections/{election_id}/blockchain-verify") data = response.json() if "verified" in data: print(f"✓ Verification endpoint working") print(f" Verified: {data['verified']}") print(f" Hash valid: {data['hash_valid']}") print(f" Chain valid: {data['chain_valid']}") print(f" Signature valid: {data['signature_valid']}") if data['verified']: print(f"\n✓ Election blockchain integrity VERIFIED") return True else: print(f"\n⚠ Election blockchain verification FAILED") if not data['hash_valid']: print(f" - Block hash mismatch (possible tampering)") if not data['chain_valid']: print(f" - Chain broken (previous block modified)") if not data['signature_valid']: print(f" - Invalid signature") return False else: print(f"✗ Unexpected response: {data}") return False except requests.exceptions.ConnectionError: print("✗ Cannot connect to backend") return False except Exception as e: print(f"✗ Error: {e}") return False def test_all_elections_active(): """Test GET /api/elections/active endpoint""" print("\n" + "="*60) print("TEST 3: Get Active Elections") print("="*60) try: response = requests.get(f"{BASE_URL}/api/elections/active") if response.status_code == 200: data = response.json() if isinstance(data, list): print(f"✓ Active elections endpoint working") print(f" Active elections: {len(data)}") if data: for election in data: print(f"\n Election {election['id']}: {election['name']}") print(f" Start: {election['start_date']}") print(f" End: {election['end_date']}") print(f" Active: {election['is_active']}") return True else: print(" No active elections") return True else: print(f"✗ Expected list, got: {type(data)}") return False else: print(f"✗ Status {response.status_code}: {response.text}") return False except Exception as e: print(f"✗ Error: {e}") return False def test_debug_all_elections(): """Test GET /api/elections/debug/all endpoint""" print("\n" + "="*60) print("TEST 4: Debug All Elections") print("="*60) try: response = requests.get(f"{BASE_URL}/api/elections/debug/all") data = response.json() if "elections" in data: print(f"✓ Debug endpoint working") print(f" Current server time: {data['current_time']}") print(f" Total elections: {len(data['elections'])}") active_count = sum(1 for e in data['elections'] if e['should_be_active']) print(f" Elections that should be active: {active_count}") if active_count > 0: print(f"\nActive elections:") for e in data['elections']: if e['should_be_active']: print(f" ✓ Election {e['id']}: {e['name']}") else: print(f"\n⚠ No elections are currently active") print(f" Check: start_date <= server_time <= end_date") return True else: print(f"✗ Unexpected response: {data}") return False except Exception as e: print(f"✗ Error: {e}") return False def test_backend_health(): """Test backend health check""" print("\n" + "="*60) print("TEST 0: Backend Health Check") print("="*60) try: # Try the root endpoint for health (Nginx intercepts /health with plain text) response = requests.get(f"{BASE_URL}/") data = response.json() if data.get("name"): print(f"✓ Backend is running") print(f" Name: {data['name']}") print(f" Version: {data.get('version', 'unknown')}") return True else: print(f"✗ Backend response unexpected: {data}") return False except requests.exceptions.ConnectionError: print("✗ Cannot connect to backend") print(f" Make sure backend is running on {BASE_URL}") print(" Run: docker compose -f docker-compose.multinode.yml up -d") return False except Exception as e: print(f"✗ Error: {e}") return False def main(): """Run all tests""" print("\n") print("╔" + "="*58 + "╗") print("║" + " "*58 + "║") print("║ Elections Blockchain Integration Tests" + " "*18 + "║") print("║" + " "*58 + "║") print("╚" + "="*58 + "╝") results = [] # Test backend first health_ok = test_backend_health() if not health_ok: print("\n✗ Backend is not running, cannot continue tests") return results.append(("Backend Health", test_backend_health())) results.append(("Blockchain Endpoint", test_blockchain_endpoint())) results.append(("Active Elections", test_all_elections_active())) results.append(("Debug Elections", test_debug_all_elections())) results.append(("Election Verification", test_election_verification())) # Summary print("\n" + "="*60) print("TEST SUMMARY") print("="*60) passed = sum(1 for _, result in results if result) total = len(results) for test_name, result in results: status = "✓ PASS" if result else "✗ FAIL" print(f"{status}: {test_name}") print(f"\nTotal: {passed}/{total} tests passed") if passed == total: print("\n✓ All tests passed! Elections blockchain integration working correctly.") else: print(f"\n⚠ {total - passed} test(s) failed. Check logs above for details.") if __name__ == "__main__": main()