fda27f2c6a
Some checks failed
Cross-Node Transaction Testing / transaction-test (push) Successful in 2s
Deploy to Testnet / deploy-testnet (push) Has been cancelled
Multi-Node Stress Testing / stress-test (push) Has been cancelled
Node Failover Simulation / failover-test (push) Has been cancelled
Python Tests / test-python (push) Has been cancelled
- Update datetime.now(UTC) to datetime.now(timezone.utc) across all test files - Fix test_block_import.py to use dynamic base height from current chain head - Fix test_block_import_complete.py to calculate base_height from current head - Add SSH options for non-interactive connection in test_cross_node_blockchain.py - Add timeout parameter to SSH subprocess call - Update all timestamp generation to use timezone.utc
239 lines
8.1 KiB
Python
239 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Comprehensive test for block import endpoint
|
|
Tests all functionality including validation, conflicts, and transaction import
|
|
"""
|
|
|
|
import json
|
|
import hashlib
|
|
import requests
|
|
from datetime import datetime, timezone
|
|
from aitbc import AITBCHTTPClient
|
|
|
|
BASE_URL = "https://aitbc.bubuit.net/rpc"
|
|
CHAIN_ID = "ait-mainnet"
|
|
|
|
def compute_block_hash(height, parent_hash, timestamp):
|
|
"""Compute block hash using the same algorithm as PoA proposer"""
|
|
payload = f"{CHAIN_ID}|{height}|{parent_hash}|{timestamp}".encode()
|
|
return "0x" + hashlib.sha256(payload).hexdigest()
|
|
|
|
def test_block_import_complete():
|
|
"""Complete test suite for block import endpoint"""
|
|
|
|
print("=" * 60)
|
|
print("BLOCK IMPORT ENDPOINT TEST SUITE")
|
|
print("=" * 60)
|
|
|
|
results = []
|
|
client = AITBCHTTPClient()
|
|
|
|
# Get current head to use for dynamic height calculation
|
|
response = requests.get(f"{BASE_URL}/head")
|
|
head = response.json()
|
|
base_height = head["height"] + 10000000
|
|
|
|
# Test 1: Invalid height (0)
|
|
print("\n[TEST 1] Invalid height (0)...")
|
|
response = client.post(
|
|
f"{BASE_URL}/importBlock",
|
|
json={
|
|
"height": 0,
|
|
"hash": "0x123",
|
|
"parent_hash": "0x00",
|
|
"proposer": "test",
|
|
"timestamp": "2026-01-29T10:20:00",
|
|
"tx_count": 0,
|
|
"chain_id": CHAIN_ID
|
|
}
|
|
)
|
|
if response.status_code == 422 and "greater_than" in response.json()["detail"][0]["msg"]:
|
|
print("✅ PASS: Correctly rejected height 0")
|
|
results.append(True)
|
|
else:
|
|
print(f"❌ FAIL: Expected 422, got {response.status_code}")
|
|
results.append(False)
|
|
|
|
# Test 2: Block conflict
|
|
print("\n[TEST 2] Block conflict...")
|
|
response = requests.post(
|
|
f"{BASE_URL}/importBlock",
|
|
json={
|
|
"height": base_height,
|
|
"hash": "0xinvalidhash",
|
|
"parent_hash": "0x00",
|
|
"proposer": "test",
|
|
"timestamp": "2026-01-29T10:20:00",
|
|
"tx_count": 0,
|
|
"chain_id": CHAIN_ID
|
|
}
|
|
)
|
|
if response.status_code == 409 and "already exists with different hash" in response.json()["detail"]:
|
|
print("✅ PASS: Correctly detected block conflict")
|
|
results.append(True)
|
|
else:
|
|
print(f"❌ FAIL: Expected 409, got {response.status_code}")
|
|
results.append(False)
|
|
|
|
# Test 3: Import existing block with correct hash
|
|
print("\n[TEST 3] Import existing block with correct hash...")
|
|
response = requests.get(f"{BASE_URL}/head")
|
|
head = response.json()
|
|
|
|
response = requests.post(
|
|
f"{BASE_URL}/importBlock",
|
|
json={
|
|
"height": head["height"],
|
|
"hash": head["hash"],
|
|
"parent_hash": head.get("parent_hash", "0x00"),
|
|
"proposer": head.get("proposer", "test"),
|
|
"timestamp": head["timestamp"],
|
|
"tx_count": head.get("tx_count", 0),
|
|
"chain_id": CHAIN_ID
|
|
}
|
|
)
|
|
if response.status_code == 200 and response.json().get("success") == True:
|
|
print("✅ PASS: Correctly handled existing block")
|
|
results.append(True)
|
|
else:
|
|
print(f"❌ FAIL: Expected 200 with success=True, got {response.status_code}")
|
|
results.append(False)
|
|
|
|
# Test 4: Invalid block hash
|
|
print("\n[TEST 4] Invalid block hash...")
|
|
response = requests.get(f"{BASE_URL}/head")
|
|
head = response.json()
|
|
|
|
response = requests.post(
|
|
f"{BASE_URL}/importBlock",
|
|
json={
|
|
"height": base_height + 1,
|
|
"hash": "0xinvalid",
|
|
"parent_hash": head["hash"],
|
|
"proposer": "test",
|
|
"timestamp": "2026-01-29T10:20:00",
|
|
"tx_count": 0,
|
|
"chain_id": CHAIN_ID
|
|
}
|
|
)
|
|
if response.status_code == 400 and "Invalid block hash" in response.json()["detail"]:
|
|
print("✅ PASS: Correctly rejected invalid hash")
|
|
results.append(True)
|
|
else:
|
|
print(f"❌ FAIL: Expected 400, got {response.status_code}")
|
|
results.append(False)
|
|
|
|
# Test 5: Parent not found
|
|
print("\n[TEST 5] Parent block not found...")
|
|
response = requests.post(
|
|
f"{BASE_URL}/importBlock",
|
|
json={
|
|
"height": base_height + 2,
|
|
"hash": compute_block_hash(base_height + 2, "0xnonexistent", "2026-01-29T10:20:00"),
|
|
"parent_hash": "0xnonexistent",
|
|
"proposer": "test",
|
|
"timestamp": "2026-01-29T10:20:00",
|
|
"tx_count": 0,
|
|
"chain_id": CHAIN_ID
|
|
}
|
|
)
|
|
if response.status_code == 400 and "Parent block not found" in response.json()["detail"]:
|
|
print("✅ PASS: Correctly rejected missing parent")
|
|
results.append(True)
|
|
else:
|
|
print(f"❌ FAIL: Expected 400, got {response.status_code}")
|
|
results.append(False)
|
|
|
|
# Test 6: Import block without transactions
|
|
print("\n[TEST 6] Import block without transactions...")
|
|
response = requests.get(f"{BASE_URL}/head")
|
|
head = response.json()
|
|
|
|
height = base_height + 10
|
|
block_hash = compute_block_hash(height, head["hash"], "2026-01-29T10:20:00")
|
|
|
|
response = requests.post(
|
|
f"{BASE_URL}/importBlock",
|
|
json={
|
|
"height": height,
|
|
"hash": block_hash,
|
|
"parent_hash": head["hash"],
|
|
"proposer": "test-proposer",
|
|
"timestamp": "2026-01-29T10:20:00",
|
|
"tx_count": 0,
|
|
"transactions": [],
|
|
"chain_id": CHAIN_ID
|
|
}
|
|
)
|
|
if response.status_code == 200 and response.json().get("success") == True:
|
|
print("✅ PASS: Successfully imported block without transactions")
|
|
results.append(True)
|
|
else:
|
|
print(f"❌ FAIL: Expected 200 with success=True, got {response.status_code}")
|
|
results.append(False)
|
|
|
|
# Test 7: Import block with transactions (KNOWN ISSUE)
|
|
print("\n[TEST 7] Import block with transactions...")
|
|
print("⚠️ KNOWN ISSUE: Transaction import currently fails with database constraint error")
|
|
print(" This appears to be a bug in the transaction field mapping")
|
|
|
|
height = base_height + 11
|
|
block_hash = compute_block_hash(height, head["hash"], "2026-01-29T10:20:00")
|
|
|
|
response = requests.post(
|
|
f"{BASE_URL}/importBlock",
|
|
json={
|
|
"height": height,
|
|
"hash": block_hash,
|
|
"parent_hash": head["hash"],
|
|
"proposer": "test-proposer",
|
|
"timestamp": "2026-01-29T10:20:00",
|
|
"tx_count": 1,
|
|
"chain_id": CHAIN_ID,
|
|
"transactions": [{
|
|
"tx_hash": "0xtx123",
|
|
"sender": "0xsender",
|
|
"recipient": "0xrecipient",
|
|
"payload": {"test": "data"}
|
|
}]
|
|
}
|
|
)
|
|
if response.status_code == 500:
|
|
print("⚠️ EXPECTED FAILURE: Transaction import fails with 500 error")
|
|
print(" Error: NOT NULL constraint failed on transaction fields")
|
|
results.append(None) # Known issue, not counting as fail
|
|
else:
|
|
print(f"❓ UNEXPECTED: Got {response.status_code} instead of expected 500")
|
|
results.append(None)
|
|
|
|
# Summary
|
|
print("\n" + "=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
|
|
passed = sum(1 for r in results if r is True)
|
|
failed = sum(1 for r in results if r is False)
|
|
known_issues = sum(1 for r in results if r is None)
|
|
|
|
print(f"✅ Passed: {passed}")
|
|
print(f"❌ Failed: {failed}")
|
|
if known_issues > 0:
|
|
print(f"⚠️ Known Issues: {known_issues}")
|
|
|
|
print("\nFUNCTIONALITY STATUS:")
|
|
print("- ✅ Input validation (height, hash, parent)")
|
|
print("- ✅ Conflict detection")
|
|
print("- ✅ Block import without transactions")
|
|
print("- ❌ Block import with transactions (database constraint issue)")
|
|
|
|
if failed == 0:
|
|
print("\n🎉 All core functionality is working!")
|
|
print(" The block import endpoint is functional for basic use.")
|
|
else:
|
|
print(f"\n⚠️ {failed} test(s) failed - review required")
|
|
|
|
return passed, failed, known_issues
|
|
|
|
if __name__ == "__main__":
|
|
test_block_import_complete()
|