mirror of
https://github.com/luxfi/consensus.git
synced 2026-09-24 00:26:58 +00:00
zeekay
4bbeffe36e
- Fixed cmd/checker and cmd/consensus to use new Engine interface - Fixed integration tests to use GetConfig instead of Config() - Updated Python SDK tests for new Chain API - Added basic Rust tests that pass - Fixed import paths from luxfi/node to local packages - Moved test files to proper directories
149 lines
4.0 KiB
Python
149 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
|
|
# See the file LICENSE for licensing terms.
|
|
|
|
import sys
|
|
import time
|
|
from lux_consensus import (
|
|
Chain, Config, Block, Vote,
|
|
Status, Decision, VoteType,
|
|
ConsensusError, default_config, new_chain, new_block, new_vote
|
|
)
|
|
|
|
def test_initialization():
|
|
"""Test basic initialization"""
|
|
print("=== Test: Initialization ===")
|
|
|
|
config = default_config()
|
|
chain = new_chain(config)
|
|
print("✅ Chain created successfully")
|
|
|
|
# Chain will be automatically cleaned up when it goes out of scope
|
|
return True
|
|
|
|
def test_block_operations():
|
|
"""Test block operations"""
|
|
print("=== Test: Block Operations ===")
|
|
|
|
config = default_config()
|
|
chain = new_chain(config)
|
|
chain.start() # Start the engine first
|
|
|
|
# Create and add genesis block (already added by start)
|
|
# Create child block
|
|
genesis_id = chain.last_accepted
|
|
block1 = new_block(b"block1", genesis_id, 1, b"Block 1")
|
|
chain.add(block1)
|
|
print(f"✅ Added block 1: {block1.id}")
|
|
|
|
# Check status
|
|
status = chain.get_status(block1.id)
|
|
print(f"✅ Block 1 status: {status}")
|
|
|
|
# Test voting
|
|
vote = new_vote(block1.id, VoteType.COMMIT, b"validator1")
|
|
chain.record_vote(vote)
|
|
print(f"✅ Recorded vote for block 1")
|
|
|
|
chain.stop()
|
|
|
|
return True
|
|
|
|
def test_consensus_flow():
|
|
"""Test consensus flow"""
|
|
print("=== Test: Consensus Flow ===")
|
|
|
|
config = default_config()
|
|
config.k = 3 # Small network
|
|
config.alpha = 2 # Need 2 votes for acceptance
|
|
|
|
chain = new_chain(config)
|
|
chain.start()
|
|
|
|
# Create block (genesis already created by start)
|
|
genesis_id = chain.last_accepted
|
|
block = new_block(b"block1", genesis_id, 1, b"Test Block")
|
|
chain.add(block)
|
|
|
|
# Simulate voting from multiple validators
|
|
validators = [b"validator1", b"validator2", b"validator3"]
|
|
for validator in validators:
|
|
vote = new_vote(block.id, VoteType.COMMIT, validator)
|
|
chain.record_vote(vote)
|
|
print(f"✅ Validator {validator.decode()} voted")
|
|
|
|
# Check if block is accepted
|
|
if chain.is_accepted(block.id):
|
|
print("✅ Block achieved consensus and was accepted")
|
|
else:
|
|
print("⚠️ Block not yet accepted (may need more votes)")
|
|
|
|
chain.stop()
|
|
|
|
return True
|
|
|
|
def test_error_handling():
|
|
"""Test error handling"""
|
|
print("=== Test: Error Handling ===")
|
|
|
|
config = default_config()
|
|
chain = new_chain(config)
|
|
chain.start()
|
|
|
|
# Get status of non-existent block (should return UNKNOWN, not raise error)
|
|
status = chain.get_status(b"non_existent")
|
|
if status == Status.UNKNOWN:
|
|
print("✅ Non-existent block correctly returns UNKNOWN status")
|
|
else:
|
|
print("❌ Non-existent block should return UNKNOWN status")
|
|
return False
|
|
|
|
# Try to record vote for non-existent block (should raise error)
|
|
try:
|
|
vote = new_vote(b"non_existent", VoteType.COMMIT, b"validator1")
|
|
chain.record_vote(vote)
|
|
print("❌ Should have raised error for vote on non-existent block")
|
|
return False
|
|
except ConsensusError:
|
|
print("✅ Correctly raised error for vote on non-existent block")
|
|
|
|
chain.stop()
|
|
|
|
return True
|
|
|
|
def main():
|
|
"""Run all tests"""
|
|
print("Starting Lux Consensus Python SDK Tests\n")
|
|
|
|
tests = [
|
|
test_initialization,
|
|
test_block_operations,
|
|
test_consensus_flow,
|
|
test_error_handling
|
|
]
|
|
|
|
passed = 0
|
|
failed = 0
|
|
|
|
for test in tests:
|
|
try:
|
|
if test():
|
|
passed += 1
|
|
else:
|
|
failed += 1
|
|
print(f"❌ Test {test.__name__} failed")
|
|
except Exception as e:
|
|
failed += 1
|
|
print(f"❌ Test {test.__name__} raised exception: {e}")
|
|
print()
|
|
|
|
print(f"Results: {passed} passed, {failed} failed")
|
|
|
|
if failed > 0:
|
|
sys.exit(1)
|
|
|
|
print("\n✅ All tests passed!")
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |