mirror of
https://github.com/torlando-tech/columba
synced 2026-08-12 18:07:15 -04:00
0f29de2054
Bug 1 - MAC Rotation: - Bypass MAC sorting after MAC rotation cleanup to ensure reconnection - When a peer's MAC address rotates, we must reconnect regardless of which device has the higher MAC address - Updated ble-reticulum submodule with fix and tests Bug 2 - Dual Connection Deduplication: - Add onAddressChanged callback from Kotlin to Python - When Kotlin deduplicates a dual connection (central + peripheral), Python now updates its address mappings so messages continue to flow - Updates identity_to_address, address_to_identity, and fragmenter keys Files changed: - KotlinBLEBridge.kt: Add setOnAddressChanged callback - bluetooth_driver.py: Add on_address_changed to abstract interface - android_ble_driver.py: Register and handle address change events - BLEInterface.py: Add _address_changed_callback to update mappings - Added unit tests for both bugs 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
677 lines
26 KiB
Python
677 lines
26 KiB
Python
"""
|
|
Test suite for Android BLE Driver
|
|
|
|
Tests the race condition handling in AndroidBLEDriver where data may arrive
|
|
from an address with a pending identity but no completed connection.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import unittest
|
|
from unittest.mock import Mock, MagicMock, patch
|
|
from enum import Enum, auto
|
|
import threading
|
|
|
|
# Add parent directory to path to import modules
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Mock RNS before importing the driver
|
|
mock_rns = MagicMock()
|
|
mock_rns.LOG_DEBUG = 5
|
|
mock_rns.LOG_INFO = 4
|
|
mock_rns.LOG_WARNING = 3
|
|
mock_rns.LOG_ERROR = 2
|
|
mock_rns.LOG_EXTREME = 6
|
|
mock_rns.log = MagicMock()
|
|
sys.modules['RNS'] = mock_rns
|
|
|
|
|
|
# Create mock DriverState enum
|
|
class MockDriverState(Enum):
|
|
IDLE = auto()
|
|
SCANNING = auto()
|
|
ADVERTISING = auto()
|
|
CONNECTING = auto()
|
|
|
|
|
|
# Mock bluetooth_driver module
|
|
mock_bluetooth_driver = MagicMock()
|
|
mock_bluetooth_driver.DriverState = MockDriverState
|
|
mock_bluetooth_driver.BLEDriverInterface = MagicMock()
|
|
mock_bluetooth_driver.BLEDevice = MagicMock()
|
|
sys.modules['bluetooth_driver'] = mock_bluetooth_driver
|
|
|
|
|
|
class MockAndroidBLEDriver:
|
|
"""
|
|
Test harness that mimics AndroidBLEDriver structure.
|
|
|
|
This extracts the relevant logic from AndroidBLEDriver._handle_data_received
|
|
for testing the race condition fix.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._connected_peers = []
|
|
self._peer_roles = {}
|
|
self._peer_mtus = {}
|
|
self._pending_identities = {}
|
|
self._identity_lock = threading.Lock()
|
|
|
|
# Callbacks
|
|
self.on_device_connected = None
|
|
self.on_mtu_negotiated = None
|
|
self.on_data_received = None
|
|
|
|
def _handle_data_received(self, address: str, data: bytes):
|
|
"""Handle data received - mirrors the fix in android_ble_driver.py."""
|
|
try:
|
|
# Check if this address has a pending identity but never got onConnected
|
|
if address not in self._connected_peers:
|
|
with self._identity_lock:
|
|
pending_identity = self._pending_identities.get(address)
|
|
if pending_identity:
|
|
# Finalize connection with pending identity
|
|
self._connected_peers.append(address)
|
|
self._peer_roles[address] = "peripheral"
|
|
# Remove from pending before callback
|
|
del self._pending_identities[address]
|
|
|
|
# Call on_device_connected to create identity mappings
|
|
if self.on_device_connected:
|
|
self.on_device_connected(address, pending_identity)
|
|
|
|
# Also call on_mtu_negotiated to create reassembler
|
|
mtu = self._peer_mtus.get(address, 23)
|
|
if self.on_mtu_negotiated:
|
|
self.on_mtu_negotiated(address, mtu)
|
|
|
|
if self.on_data_received:
|
|
self.on_data_received(address, data)
|
|
|
|
except Exception as e:
|
|
pass # Error handling
|
|
|
|
|
|
class TestAndroidBLEDriverRaceCondition(unittest.TestCase):
|
|
"""Test race condition handling when data arrives before connection completes."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.driver = MockAndroidBLEDriver()
|
|
|
|
# Set up mock callbacks
|
|
self.driver.on_device_connected = Mock()
|
|
self.driver.on_mtu_negotiated = Mock()
|
|
self.driver.on_data_received = Mock()
|
|
|
|
def test_data_from_unknown_address_without_pending_identity(self):
|
|
"""
|
|
Test that data from unknown address without pending identity is passed through.
|
|
|
|
When data arrives from an address that:
|
|
- Is NOT in connected_peers
|
|
- Does NOT have a pending identity
|
|
|
|
The driver should NOT call on_device_connected, just pass data through.
|
|
"""
|
|
address = "AA:BB:CC:DD:EE:FF"
|
|
data = b"test data"
|
|
|
|
self.driver._handle_data_received(address, data)
|
|
|
|
# Should not trigger connection callbacks
|
|
self.driver.on_device_connected.assert_not_called()
|
|
self.driver.on_mtu_negotiated.assert_not_called()
|
|
|
|
# Should still pass data through
|
|
self.driver.on_data_received.assert_called_once_with(address, data)
|
|
|
|
def test_data_from_connected_peer(self):
|
|
"""
|
|
Test that data from connected peer is passed through normally.
|
|
|
|
When data arrives from an address that IS in connected_peers,
|
|
the driver should just pass data through without any connection logic.
|
|
"""
|
|
address = "AA:BB:CC:DD:EE:FF"
|
|
data = b"test data"
|
|
|
|
# Mark peer as connected
|
|
self.driver._connected_peers.append(address)
|
|
|
|
self.driver._handle_data_received(address, data)
|
|
|
|
# Should not trigger connection callbacks (already connected)
|
|
self.driver.on_device_connected.assert_not_called()
|
|
self.driver.on_mtu_negotiated.assert_not_called()
|
|
|
|
# Should pass data through
|
|
self.driver.on_data_received.assert_called_once_with(address, data)
|
|
|
|
def test_data_from_address_with_pending_identity(self):
|
|
"""
|
|
Test that data from address with pending identity triggers connection finalization.
|
|
|
|
This is the key race condition fix: When data arrives from an address that:
|
|
- Is NOT in connected_peers
|
|
- HAS a pending identity
|
|
|
|
The driver should:
|
|
1. Add to connected_peers
|
|
2. Call on_device_connected with the pending identity
|
|
3. Call on_mtu_negotiated to create reassembler
|
|
4. Pass data through to on_data_received
|
|
"""
|
|
address = "AA:BB:CC:DD:EE:FF"
|
|
identity = b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10"
|
|
data = b"test data"
|
|
|
|
# Set up pending identity
|
|
self.driver._pending_identities[address] = identity
|
|
|
|
self.driver._handle_data_received(address, data)
|
|
|
|
# Should trigger connection finalization
|
|
self.driver.on_device_connected.assert_called_once_with(address, identity)
|
|
self.driver.on_mtu_negotiated.assert_called_once()
|
|
|
|
# Check MTU callback was called with address and default MTU (23)
|
|
mtu_call_args = self.driver.on_mtu_negotiated.call_args
|
|
self.assertEqual(mtu_call_args[0][0], address)
|
|
self.assertEqual(mtu_call_args[0][1], 23) # Default BLE 4.0 MTU
|
|
|
|
# Should pass data through
|
|
self.driver.on_data_received.assert_called_once_with(address, data)
|
|
|
|
# Should be added to connected peers
|
|
self.assertIn(address, self.driver._connected_peers)
|
|
self.assertEqual(self.driver._peer_roles[address], "peripheral")
|
|
|
|
# Should be removed from pending identities
|
|
self.assertNotIn(address, self.driver._pending_identities)
|
|
|
|
def test_data_from_address_with_pending_identity_uses_cached_mtu(self):
|
|
"""
|
|
Test that MTU from cache is used when available.
|
|
|
|
If we have a cached MTU for this address (from earlier negotiation),
|
|
the driver should use that MTU value instead of the default.
|
|
"""
|
|
address = "AA:BB:CC:DD:EE:FF"
|
|
identity = b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10"
|
|
data = b"test data"
|
|
cached_mtu = 512
|
|
|
|
# Set up pending identity and cached MTU
|
|
self.driver._pending_identities[address] = identity
|
|
self.driver._peer_mtus[address] = cached_mtu
|
|
|
|
self.driver._handle_data_received(address, data)
|
|
|
|
# Check MTU callback was called with cached MTU
|
|
mtu_call_args = self.driver.on_mtu_negotiated.call_args
|
|
self.assertEqual(mtu_call_args[0][1], cached_mtu)
|
|
|
|
def test_pending_identity_consumed_only_once(self):
|
|
"""
|
|
Test that pending identity is removed after first use.
|
|
|
|
If data arrives twice from same address with pending identity,
|
|
only the first time should trigger connection finalization.
|
|
"""
|
|
address = "AA:BB:CC:DD:EE:FF"
|
|
identity = b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10"
|
|
data1 = b"first data"
|
|
data2 = b"second data"
|
|
|
|
# Set up pending identity
|
|
self.driver._pending_identities[address] = identity
|
|
|
|
# First data arrival - should trigger connection
|
|
self.driver._handle_data_received(address, data1)
|
|
|
|
# Reset mocks
|
|
self.driver.on_device_connected.reset_mock()
|
|
self.driver.on_mtu_negotiated.reset_mock()
|
|
|
|
# Second data arrival - should NOT trigger connection (already connected)
|
|
self.driver._handle_data_received(address, data2)
|
|
|
|
# Should not trigger connection callbacks second time
|
|
self.driver.on_device_connected.assert_not_called()
|
|
self.driver.on_mtu_negotiated.assert_not_called()
|
|
|
|
|
|
class TestAndroidBLEDriverCallbackOrdering(unittest.TestCase):
|
|
"""Test that callbacks are called in correct order."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.driver = MockAndroidBLEDriver()
|
|
|
|
# Track call order
|
|
self.call_order = []
|
|
|
|
def track_connected(addr, identity):
|
|
self.call_order.append(('connected', addr))
|
|
|
|
def track_mtu(addr, mtu):
|
|
self.call_order.append(('mtu', addr, mtu))
|
|
|
|
def track_data(addr, data):
|
|
self.call_order.append(('data', addr, len(data)))
|
|
|
|
self.driver.on_device_connected = track_connected
|
|
self.driver.on_mtu_negotiated = track_mtu
|
|
self.driver.on_data_received = track_data
|
|
|
|
def test_callback_order_for_pending_identity(self):
|
|
"""
|
|
Test that callbacks are called in correct order for pending identity case.
|
|
|
|
Order must be:
|
|
1. on_device_connected (creates identity mappings)
|
|
2. on_mtu_negotiated (creates reassembler)
|
|
3. on_data_received (processes data with reassembler)
|
|
"""
|
|
address = "AA:BB:CC:DD:EE:FF"
|
|
identity = b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10"
|
|
data = b"test data"
|
|
|
|
self.driver._pending_identities[address] = identity
|
|
|
|
self.driver._handle_data_received(address, data)
|
|
|
|
# Verify order
|
|
self.assertEqual(len(self.call_order), 3)
|
|
self.assertEqual(self.call_order[0][0], 'connected')
|
|
self.assertEqual(self.call_order[1][0], 'mtu')
|
|
self.assertEqual(self.call_order[2][0], 'data')
|
|
|
|
|
|
class MockBLEInterface:
|
|
"""
|
|
Test harness that mimics BLEInterface structure for testing address change callbacks.
|
|
|
|
This tests the fix for dual connection deduplication where Python wasn't notified
|
|
when Kotlin changed the address mapping during deduplication.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.identity_to_address = {}
|
|
self.address_to_identity = {}
|
|
self.fragmenters = {}
|
|
self.reassemblers = {}
|
|
self.frag_lock = threading.Lock()
|
|
|
|
def _get_fragmenter_key(self, peer_identity, address):
|
|
"""Generate key for fragmenter/reassembler lookup."""
|
|
if peer_identity:
|
|
return f"{peer_identity.hex()[:8]}_{address}"
|
|
return address
|
|
|
|
def _address_changed_callback(self, old_address: str, new_address: str, identity_hash: str):
|
|
"""
|
|
Handle address change during dual connection deduplication.
|
|
|
|
When Kotlin deduplicates a dual connection (same identity connected as both
|
|
central and peripheral), it closes one direction and notifies Python via
|
|
this callback so Python can update its address mappings.
|
|
|
|
Args:
|
|
old_address: The address that was closed/removed
|
|
new_address: The address that remains active
|
|
identity_hash: The 32-char hex identity hash for this peer
|
|
"""
|
|
# Update identity_to_address mapping
|
|
if identity_hash in self.identity_to_address:
|
|
self.identity_to_address[identity_hash] = new_address
|
|
|
|
# Update address_to_identity mapping
|
|
peer_identity = self.address_to_identity.get(old_address)
|
|
if peer_identity:
|
|
self.address_to_identity[new_address] = peer_identity
|
|
# Keep old mapping for fallback resolution during transition
|
|
|
|
# Update fragmenter/reassembler keys
|
|
if peer_identity:
|
|
old_key = self._get_fragmenter_key(peer_identity, old_address)
|
|
new_key = self._get_fragmenter_key(peer_identity, new_address)
|
|
with self.frag_lock:
|
|
if old_key in self.fragmenters:
|
|
self.fragmenters[new_key] = self.fragmenters.pop(old_key)
|
|
if old_key in self.reassemblers:
|
|
self.reassemblers[new_key] = self.reassemblers.pop(old_key)
|
|
|
|
|
|
class TestBLEInterfaceAddressChangedCallback(unittest.TestCase):
|
|
"""
|
|
Test _address_changed_callback which handles address changes during
|
|
dual connection deduplication.
|
|
|
|
When Kotlin deduplicates dual connections (same identity connected as both
|
|
central and peripheral), it closes one direction and notifies Python via
|
|
this callback so Python can update its address mappings.
|
|
"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.interface = MockBLEInterface()
|
|
|
|
def test_address_changed_updates_identity_to_address_mapping(self):
|
|
"""
|
|
Test that _address_changed_callback updates identity_to_address mapping.
|
|
|
|
When address changes from old to new, the identity_to_address map
|
|
should point to the new address.
|
|
"""
|
|
old_address = "11:22:33:44:55:66"
|
|
new_address = "AA:BB:CC:DD:EE:FF"
|
|
identity_hash = "ab5609dfffb33b21a102e1ff81196be5"
|
|
peer_identity = b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10'
|
|
|
|
# Set up initial state
|
|
self.interface.identity_to_address[identity_hash] = old_address
|
|
self.interface.address_to_identity[old_address] = peer_identity
|
|
|
|
# Call the callback (this should fail - method doesn't exist yet)
|
|
self.interface._address_changed_callback(old_address, new_address, identity_hash)
|
|
|
|
# Verify mapping updated
|
|
self.assertEqual(self.interface.identity_to_address[identity_hash], new_address)
|
|
|
|
def test_address_changed_updates_address_to_identity_mapping(self):
|
|
"""
|
|
Test that _address_changed_callback updates address_to_identity mapping.
|
|
|
|
The new address should map to the same peer identity.
|
|
"""
|
|
old_address = "11:22:33:44:55:66"
|
|
new_address = "AA:BB:CC:DD:EE:FF"
|
|
identity_hash = "ab5609dfffb33b21a102e1ff81196be5"
|
|
peer_identity = b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10'
|
|
|
|
# Set up initial state
|
|
self.interface.identity_to_address[identity_hash] = old_address
|
|
self.interface.address_to_identity[old_address] = peer_identity
|
|
|
|
# Call the callback
|
|
self.interface._address_changed_callback(old_address, new_address, identity_hash)
|
|
|
|
# Verify new address maps to peer identity
|
|
self.assertEqual(self.interface.address_to_identity[new_address], peer_identity)
|
|
|
|
def test_address_changed_updates_fragmenter_keys(self):
|
|
"""
|
|
Test that _address_changed_callback updates fragmenter keys.
|
|
|
|
Fragmenters are keyed by (identity_hash, address). When address changes,
|
|
the fragmenter should be accessible via the new key.
|
|
"""
|
|
old_address = "11:22:33:44:55:66"
|
|
new_address = "AA:BB:CC:DD:EE:FF"
|
|
identity_hash = "ab5609dfffb33b21a102e1ff81196be5"
|
|
peer_identity = b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10'
|
|
mock_fragmenter = Mock()
|
|
|
|
# Set up initial state
|
|
self.interface.identity_to_address[identity_hash] = old_address
|
|
self.interface.address_to_identity[old_address] = peer_identity
|
|
old_key = self.interface._get_fragmenter_key(peer_identity, old_address)
|
|
self.interface.fragmenters[old_key] = mock_fragmenter
|
|
|
|
# Call the callback
|
|
self.interface._address_changed_callback(old_address, new_address, identity_hash)
|
|
|
|
# Verify fragmenter is accessible via new key
|
|
new_key = self.interface._get_fragmenter_key(peer_identity, new_address)
|
|
self.assertIn(new_key, self.interface.fragmenters)
|
|
self.assertEqual(self.interface.fragmenters[new_key], mock_fragmenter)
|
|
|
|
def test_address_changed_updates_reassembler_keys(self):
|
|
"""
|
|
Test that _address_changed_callback updates reassembler keys.
|
|
|
|
Reassemblers are keyed by (identity_hash, address). When address changes,
|
|
the reassembler should be accessible via the new key.
|
|
"""
|
|
old_address = "11:22:33:44:55:66"
|
|
new_address = "AA:BB:CC:DD:EE:FF"
|
|
identity_hash = "ab5609dfffb33b21a102e1ff81196be5"
|
|
peer_identity = b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10'
|
|
mock_reassembler = Mock()
|
|
|
|
# Set up initial state
|
|
self.interface.identity_to_address[identity_hash] = old_address
|
|
self.interface.address_to_identity[old_address] = peer_identity
|
|
old_key = self.interface._get_fragmenter_key(peer_identity, old_address)
|
|
self.interface.reassemblers[old_key] = mock_reassembler
|
|
|
|
# Call the callback
|
|
self.interface._address_changed_callback(old_address, new_address, identity_hash)
|
|
|
|
# Verify reassembler is accessible via new key
|
|
new_key = self.interface._get_fragmenter_key(peer_identity, new_address)
|
|
self.assertIn(new_key, self.interface.reassemblers)
|
|
self.assertEqual(self.interface.reassemblers[new_key], mock_reassembler)
|
|
|
|
|
|
class MockDiscoveredPeer:
|
|
"""Mock discovered peer for testing."""
|
|
def __init__(self, address: str, name: str = "RNS-test", rssi: int = -50):
|
|
self.address = address
|
|
self.name = name
|
|
self.rssi = rssi
|
|
|
|
|
|
class MockBLEInterfaceMACRotation:
|
|
"""
|
|
Test harness for MAC rotation handling in _select_peers_to_connect.
|
|
|
|
Tests the fix where after MAC rotation cleanup, peer is immediately added
|
|
to scored_peers list bypassing MAC sorting.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.identity_to_address = {}
|
|
self.address_to_identity = {}
|
|
self.spawned_interfaces = {}
|
|
self.peers = {} # address -> active connection
|
|
self.connection_attempt_times = {}
|
|
self.local_mac = "AA:BB:CC:DD:EE:FF" # Higher MAC for sorting test
|
|
self.cleanup_called_with = [] # Track cleanup calls
|
|
|
|
def _compute_identity_hash(self, peer_identity):
|
|
"""Compute identity hash."""
|
|
if peer_identity:
|
|
return peer_identity.hex()[:16]
|
|
return None
|
|
|
|
def _score_peer(self, peer):
|
|
"""Score a peer for connection prioritization."""
|
|
return abs(peer.rssi) # Lower RSSI = higher priority
|
|
|
|
def _cleanup_stale_interface(self, identity_hash: str, old_address: str):
|
|
"""Mock cleanup - just track that it was called."""
|
|
self.cleanup_called_with.append((identity_hash, old_address))
|
|
# Clean up mappings like real method
|
|
if identity_hash in self.spawned_interfaces:
|
|
del self.spawned_interfaces[identity_hash]
|
|
if identity_hash in self.identity_to_address:
|
|
del self.identity_to_address[identity_hash]
|
|
|
|
def _select_peers_to_connect(self, discovered_peers):
|
|
"""
|
|
Simplified version of _select_peers_to_connect that includes the MAC rotation fix.
|
|
|
|
Returns list of (score, peer) tuples.
|
|
"""
|
|
scored_peers = []
|
|
|
|
for peer in discovered_peers:
|
|
address = peer.address
|
|
|
|
# Check for MAC rotation
|
|
peer_identity = self.address_to_identity.get(address)
|
|
if peer_identity:
|
|
identity_hash = self._compute_identity_hash(peer_identity)
|
|
if identity_hash in self.spawned_interfaces:
|
|
existing_address = self.identity_to_address.get(identity_hash)
|
|
if existing_address and existing_address != address:
|
|
# Same identity at different MAC = MAC rotation
|
|
if existing_address in self.peers:
|
|
# Old connection still active - skip
|
|
continue
|
|
else:
|
|
# Old connection dead - clean up and allow new connection
|
|
self._cleanup_stale_interface(identity_hash, existing_address)
|
|
# FIX: Bypass MAC sorting - we must reconnect after MAC rotation
|
|
score = self._score_peer(peer)
|
|
scored_peers.append((score, peer))
|
|
continue # Skip remaining checks, peer already added
|
|
|
|
# MAC sorting check (only reached if NOT MAC rotation)
|
|
if self.local_mac > address:
|
|
# We have higher MAC, we should initiate - add peer
|
|
score = self._score_peer(peer)
|
|
scored_peers.append((score, peer))
|
|
# else: peer has higher MAC, they should initiate - skip
|
|
|
|
return scored_peers
|
|
|
|
|
|
class TestMACRotationFix(unittest.TestCase):
|
|
"""
|
|
Test MAC rotation handling in _select_peers_to_connect.
|
|
|
|
Bug: After MAC rotation, peer interface wasn't recreated because MAC sorting
|
|
check skipped the peer. Fix: After cleanup, immediately add peer and continue.
|
|
"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.interface = MockBLEInterfaceMACRotation()
|
|
|
|
def test_mac_rotation_detected_and_cleanup_called(self):
|
|
"""
|
|
Test that MAC rotation is detected and cleanup is called.
|
|
|
|
When same identity appears at new MAC with stale old connection,
|
|
_cleanup_stale_interface should be called.
|
|
"""
|
|
old_address = "11:22:33:44:55:66"
|
|
new_address = "77:88:99:AA:BB:CC"
|
|
identity_hash = "ab5609dfffb33b21"
|
|
peer_identity = bytes.fromhex("ab5609dfffb33b21a102e1ff81196be5")
|
|
|
|
# Set up: identity exists at old address, but connection is stale (not in self.peers)
|
|
self.interface.identity_to_address[identity_hash] = old_address
|
|
self.interface.address_to_identity[new_address] = peer_identity
|
|
self.interface.spawned_interfaces[identity_hash] = Mock()
|
|
# Note: old_address NOT in self.peers (connection is dead)
|
|
|
|
# Create peer at new address
|
|
peer = MockDiscoveredPeer(new_address, "RNS-ab5609")
|
|
|
|
# Act
|
|
self.interface._select_peers_to_connect([peer])
|
|
|
|
# Assert: cleanup was called with correct arguments
|
|
self.assertEqual(len(self.interface.cleanup_called_with), 1)
|
|
self.assertEqual(self.interface.cleanup_called_with[0], (identity_hash, old_address))
|
|
|
|
def test_mac_rotation_bypasses_mac_sorting(self):
|
|
"""
|
|
Test that MAC rotation bypasses MAC sorting.
|
|
|
|
After MAC rotation cleanup, peer should be added to connection list
|
|
EVEN IF local MAC > peer MAC (which would normally skip the peer).
|
|
"""
|
|
old_address = "11:22:33:44:55:66"
|
|
new_address = "00:11:22:33:44:55" # Lower than local MAC (AA:BB:CC...)
|
|
identity_hash = "ab5609dfffb33b21"
|
|
peer_identity = bytes.fromhex("ab5609dfffb33b21a102e1ff81196be5")
|
|
|
|
# Set up: identity exists at old address, but connection is stale
|
|
self.interface.identity_to_address[identity_hash] = old_address
|
|
self.interface.address_to_identity[new_address] = peer_identity
|
|
self.interface.spawned_interfaces[identity_hash] = Mock()
|
|
# old_address NOT in self.peers (stale connection)
|
|
|
|
# Create peer at new address (lower MAC than local)
|
|
peer = MockDiscoveredPeer(new_address, "RNS-ab5609")
|
|
|
|
# Act
|
|
result = self.interface._select_peers_to_connect([peer])
|
|
|
|
# Assert: peer was added despite having lower MAC
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0][1].address, new_address)
|
|
|
|
def test_mac_sorting_applies_when_no_mac_rotation(self):
|
|
"""
|
|
Test that normal MAC sorting still works when there's no MAC rotation.
|
|
|
|
When peer MAC > local MAC and no rotation, peer should be skipped
|
|
(peer should initiate connection since they have higher MAC).
|
|
"""
|
|
new_address = "FF:FF:FF:FF:FF:FF" # Higher than local MAC (AA:BB:CC...)
|
|
|
|
# No existing identity mapping - this is a new peer
|
|
peer = MockDiscoveredPeer(new_address, "RNS-newpeer")
|
|
|
|
# Act
|
|
result = self.interface._select_peers_to_connect([peer])
|
|
|
|
# Assert: peer was NOT added (they have higher MAC, they should initiate)
|
|
self.assertEqual(len(result), 0)
|
|
|
|
def test_mac_sorting_adds_peer_when_local_mac_higher(self):
|
|
"""
|
|
Test that MAC sorting adds peer when local MAC is higher.
|
|
|
|
When local MAC > peer MAC and no rotation, we should initiate.
|
|
"""
|
|
new_address = "00:11:22:33:44:55" # Lower than local MAC (AA:BB:CC...)
|
|
|
|
# Set local MAC higher
|
|
self.interface.local_mac = "FF:FF:FF:FF:FF:FF"
|
|
|
|
peer = MockDiscoveredPeer(new_address, "RNS-newpeer")
|
|
|
|
# Act
|
|
result = self.interface._select_peers_to_connect([peer])
|
|
|
|
# Assert: peer was added (we have higher MAC, we initiate)
|
|
self.assertEqual(len(result), 1)
|
|
|
|
def test_active_connection_skips_rotation(self):
|
|
"""
|
|
Test that active connection prevents MAC rotation cleanup.
|
|
|
|
When old connection is still active (in self.peers), we should
|
|
NOT clean up or add the new address.
|
|
"""
|
|
old_address = "11:22:33:44:55:66"
|
|
new_address = "77:88:99:AA:BB:CC"
|
|
identity_hash = "ab5609dfffb33b21"
|
|
peer_identity = bytes.fromhex("ab5609dfffb33b21a102e1ff81196be5")
|
|
|
|
# Set up: identity exists at old address AND connection is active
|
|
self.interface.identity_to_address[identity_hash] = old_address
|
|
self.interface.address_to_identity[new_address] = peer_identity
|
|
self.interface.spawned_interfaces[identity_hash] = Mock()
|
|
self.interface.peers[old_address] = Mock() # Connection is ACTIVE
|
|
|
|
peer = MockDiscoveredPeer(new_address, "RNS-ab5609")
|
|
|
|
# Act
|
|
result = self.interface._select_peers_to_connect([peer])
|
|
|
|
# Assert: cleanup was NOT called, peer was NOT added
|
|
self.assertEqual(len(self.interface.cleanup_called_with), 0)
|
|
self.assertEqual(len(result), 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|