Added bootstrap module

This commit is contained in:
sumanjeet0012@gmail.com
2025-06-29 16:10:27 +05:30
parent 92c9ba7e46
commit 12ad2dcdf4
12 changed files with 419 additions and 4 deletions

View File

@ -0,0 +1 @@
"""Discovery tests for py-libp2p."""

View File

@ -0,0 +1 @@
"""Bootstrap discovery tests for py-libp2p."""

View File

@ -0,0 +1,70 @@
#!/usr/bin/env python3
"""
Simple test script to verify bootstrap functionality
"""
import os
import sys
# Add the parent directory to sys.path so we can import libp2p
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../../"))
from libp2p.discovery.bootstrap.utils import (
parse_bootstrap_peer_info,
validate_bootstrap_addresses,
)
def test_bootstrap_validation():
"""Test bootstrap address validation"""
print("🧪 Testing Bootstrap Address Validation")
# Test addresses
test_addresses = [
"/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SznbYGzPwp8qDrq",
"/ip4/127.0.0.1/tcp/8000/p2p/QmTest123", # This might be invalid peer ID format
"invalid-address",
"/ip4/192.168.1.1/tcp/4001", # Missing p2p component
"/ip6/2604:a880:1:20::203:d001/tcp/4001/p2p/QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM",
]
print(f"📋 Testing {len(test_addresses)} addresses:")
for addr in test_addresses:
print(f"{addr}")
# Validate addresses
valid_addresses = validate_bootstrap_addresses(test_addresses)
print(f"\n✅ Valid addresses ({len(valid_addresses)}):")
for addr in valid_addresses:
print(f"{addr}")
# Try to parse peer info
peer_info = parse_bootstrap_peer_info(addr)
if peer_info:
print(f" → Peer ID: {peer_info.peer_id}")
print(f" → Addresses: {[str(a) for a in peer_info.addrs]}")
else:
print(" → Failed to parse peer info")
return len(valid_addresses) > 0
if __name__ == "__main__":
print("=" * 60)
print("Bootstrap Module Test")
print("=" * 60)
try:
success = test_bootstrap_validation()
if success:
print("\n🎉 Bootstrap module test completed successfully!")
else:
print("\n❌ No valid bootstrap addresses found")
sys.exit(1)
except Exception as e:
print(f"\n💥 Test failed with error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

View File

@ -0,0 +1,78 @@
#!/usr/bin/env python3
"""
Test the full bootstrap discovery integration
"""
import logging
import secrets
import pytest
from libp2p import new_host
from libp2p.abc import PeerInfo
from libp2p.crypto.secp256k1 import create_new_key_pair
from libp2p.discovery.events.peerDiscovery import peerDiscovery
from libp2p.host.basic_host import BasicHost
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("bootstrap_test")
def on_peer_discovery(peer_info: PeerInfo) -> None:
"""Handler for peer discovery events."""
logger.info(f"🔍 Discovered peer: {peer_info.peer_id}")
logger.info(f" Addresses: {[str(addr) for addr in peer_info.addrs]}")
@pytest.mark.trio
async def test_bootstrap_integration():
"""Test bootstrap integration with new_host"""
print("🧪 Testing Bootstrap Integration")
# Test bootstrap addresses
bootstrap_addrs = [
"/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SznbYGzPwp8qDrq",
"/ip4/104.236.179.241/tcp/4001/p2p/QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM",
]
# Generate key pair
secret = secrets.token_bytes(32)
key_pair = create_new_key_pair(secret)
# Register peer discovery handler
peerDiscovery.register_peer_discovered_handler(on_peer_discovery)
print(f"🌐 Testing with {len(bootstrap_addrs)} bootstrap peers")
# Create host with bootstrap
host = new_host(key_pair=key_pair, bootstrap=bootstrap_addrs)
print("✅ Successfully created host with bootstrap")
print(f"📍 Host peer ID: {host.get_id()}")
print("🔗 Bootstrap discovery should process peers when host starts")
# Verify bootstrap discovery is set up (cast to BasicHost for type checking)
assert isinstance(host, BasicHost), "Host should be a BasicHost instance"
assert hasattr(host, "bootstrap"), "Host should have bootstrap attribute"
assert host.bootstrap is not None, "Bootstrap discovery should be initialized"
assert len(host.bootstrap.bootstrap_addrs) == len(bootstrap_addrs), (
"Bootstrap addresses should match"
)
print("🎉 Bootstrap integration test completed successfully!")
def test_bootstrap_no_addresses():
"""Test that bootstrap is not initialized when no addresses provided"""
secret = secrets.token_bytes(32)
key_pair = create_new_key_pair(secret)
# Create host without bootstrap
host = new_host(key_pair=key_pair)
# Verify bootstrap is not initialized
assert isinstance(host, BasicHost)
assert not hasattr(host, "bootstrap") or host.bootstrap is None, (
"Bootstrap should not be initialized when no addresses provided"
)