diff --git a/src/ble_reticulum/BLEInterface.py b/src/ble_reticulum/BLEInterface.py index c36df90..e93bf78 100644 --- a/src/ble_reticulum/BLEInterface.py +++ b/src/ble_reticulum/BLEInterface.py @@ -2022,7 +2022,7 @@ class BLEInterface(Interface): try: # Store central's identity central_identity = bytes(data) - central_identity_hash = RNS.Identity.full_hash(central_identity)[:16].hex()[:16] + central_identity_hash = self._compute_identity_hash(central_identity) self.address_to_identity[sender_address] = central_identity self.identity_to_address[central_identity_hash] = sender_address diff --git a/tests/mock_ble_driver.py b/tests/mock_ble_driver.py index 994f967..111f965 100644 --- a/tests/mock_ble_driver.py +++ b/tests/mock_ble_driver.py @@ -80,7 +80,7 @@ class MockBLEDriver(BLEDriverInterface): # Callbacks (assigned by consumer) self.on_device_discovered: Optional[Callable[[BLEDevice], None]] = None - self.on_device_connected: Optional[Callable[[str], None]] = None + self.on_device_connected: Optional[Callable[[str, Optional[bytes]], None]] = None # address, peer_identity self.on_device_disconnected: Optional[Callable[[str], None]] = None self.on_data_received: Optional[Callable[[str, bytes], None]] = None self.on_mtu_negotiated: Optional[Callable[[str, int], None]] = None @@ -160,16 +160,21 @@ class MockBLEDriver(BLEDriverInterface): if address in self._connected_peers: return # Already connected + # Get peer identity if linked driver is set + peer_identity = None + if self._linked_driver and self._linked_driver.local_address == address: + peer_identity = self._linked_driver._identity + # Simulate connection with default MTU self._connected_peers[address] = { "role": "central", "mtu": 185, # Default MTU - "identity": None + "identity": peer_identity } - # Trigger callback + # Trigger callback with peer identity (central mode receives identity during connection) if self.on_device_connected: - self.on_device_connected(address) + self.on_device_connected(address, peer_identity) # Trigger MTU negotiation callback if self.on_mtu_negotiated: @@ -193,8 +198,9 @@ class MockBLEDriver(BLEDriverInterface): "identity": None } + # Peripheral role: identity is None because we receive it via handshake later if self.on_device_connected: - self.on_device_connected(address) + self.on_device_connected(address, None) if self.on_mtu_negotiated: self.on_mtu_negotiated(address, 185) diff --git a/tests/test_ble_peer_interface.py b/tests/test_ble_peer_interface.py index c3d0925..1473656 100644 --- a/tests/test_ble_peer_interface.py +++ b/tests/test_ble_peer_interface.py @@ -39,7 +39,8 @@ def create_mock_peer_interface(peer_address="AA:BB:CC:DD:EE:FF", peer_name="Test parent.reassemblers = {peer_address: BLEReassembler() if BLEReassembler else Mock()} parent.frag_lock = threading.RLock() # Use threading lock for mock parent.peer_lock = threading.RLock() # Use threading lock for mock - parent.loop = asyncio.get_event_loop() + # Create a new event loop for testing (Python 3.10+ requires this) + parent.loop = asyncio.new_event_loop() parent.gatt_server = Mock() parent.gatt_server.send_notification = AsyncMock(return_value=True) diff --git a/tests/test_peer_address_mac_rotation.py b/tests/test_peer_address_mac_rotation.py index eb9abd5..4f9614c 100644 --- a/tests/test_peer_address_mac_rotation.py +++ b/tests/test_peer_address_mac_rotation.py @@ -243,6 +243,9 @@ class TestPeerAddressMacRotation: When we receive an identity handshake from a new address but already have an interface for that identity, we must update peer_address. + + Note: This simulates MAC rotation where the old connection has dropped + but the peer interface is still alive (waiting for reconnection). """ old_addr = ble_interface._old_address new_addr = ble_interface._new_address @@ -253,8 +256,16 @@ class TestPeerAddressMacRotation: assert mock_peer_interface.peer_address == old_addr assert identity_hash in ble_interface.spawned_interfaces - # Setup: peer connected at new address - but NO identity mapping yet - # This simulates a central reconnecting at a new MAC address + # Setup for MAC rotation: old connection is gone, new connection arrives + # Remove old address from peers (simulates old connection dropped) + del ble_interface.peers[old_addr] + # Remove old address from address_to_identity (cleaned up after disconnect) + del ble_interface.address_to_identity[old_addr] + # Remove old address from identity_to_address + # (this gets cleared during disconnect cleanup in real code) + del ble_interface.identity_to_address[identity_hash] + + # New peer connects at new address ble_interface.peers[new_addr] = (Mock(is_connected=True), 0, 185) # NOTE: Do NOT add new_addr to address_to_identity - the handshake does that diff --git a/tests/test_v2_2_identity_handshake.py b/tests/test_v2_2_identity_handshake.py index 0f9d87a..56e9f76 100644 --- a/tests/test_v2_2_identity_handshake.py +++ b/tests/test_v2_2_identity_handshake.py @@ -53,54 +53,55 @@ if not hasattr(RNS, 'Identity'): RNS.Identity = MagicMock() RNS.Identity.full_hash = lambda x: (x * 2)[:16] # Simple mock -# Mock ble_reticulum.Interface (required by BLEInterface.py) -# First, ensure mock is in place BEFORE any imports that need it -rns_interfaces_mock = MagicMock() -_sys.modules['ble_reticulum'] = rns_interfaces_mock -_sys.modules['ble_reticulum.Interface'] = MagicMock() +# Mock ble_reticulum.Interface module (the base class module, not the whole namespace) +# We only mock the Interface.py module, allowing BLEInterface.py to be imported from src/ +if 'ble_reticulum.Interface' not in _sys.modules: + # Create mock Interface base class + class MockInterface: + MODE_FULL = 1 + def __init__(self): + self.IN = True + self.OUT = True + self.online = False -# Create mock Interface base class -class MockInterface: - MODE_FULL = 1 - def __init__(self): - self.IN = True - self.OUT = True - self.online = False + @staticmethod + def get_config_obj(configuration): + """Mock config object wrapper - just returns a dict-like object.""" + class ConfigObj: + def __init__(self, config): + self._config = config if config else {} - @staticmethod - def get_config_obj(configuration): - """Mock config object that returns dict values via attribute access.""" - class ConfigObj: - def __init__(self, config_dict): - self._config = config_dict if isinstance(config_dict, dict) else {} + def __getitem__(self, key): + return self._config.get(key) - def __getattr__(self, name): - return self._config.get(name) + def get(self, key, default=None): + return self._config.get(key, default) - def __contains__(self, key): - return key in self._config + def as_string(self, key, default=None): + val = self._config.get(key, default) + return str(val) if val is not None else default - def get(self, key, default=None): - return self._config.get(key, default) + def as_int(self, key, default=None): + val = self._config.get(key, default) + return int(val) if val is not None else default - def as_dict(self): - return self._config + def as_bool(self, key, default=False): + val = self._config.get(key, default) + if isinstance(val, bool): + return val + if isinstance(val, str): + return val.lower() in ('true', 'yes', '1', 'on') + return bool(val) if val is not None else default + return ConfigObj(configuration) - return ConfigObj(configuration) - -rns_interfaces_mock.Interface = MockInterface -_sys.modules['ble_reticulum.Interface'].Interface = MockInterface + # Create a mock module for ble_reticulum.Interface + interface_module = MagicMock() + interface_module.Interface = MockInterface + _sys.modules['ble_reticulum.Interface'] = interface_module from tests.mock_ble_driver import MockBLEDriver - -# Import BLEInterface directly using importlib to bypass RNS namespace conflicts -import importlib.util -_ble_interface_path = os.path.join(os.path.dirname(__file__), '..', 'src', 'RNS', 'Interfaces', 'BLEInterface.py') -_spec = importlib.util.spec_from_file_location("BLEInterface", _ble_interface_path) -_ble_module = importlib.util.module_from_spec(_spec) -_spec.loader.exec_module(_ble_module) -BLEInterface = _ble_module.BLEInterface -DiscoveredPeer = _ble_module.DiscoveredPeer +from ble_reticulum.BLEInterface import BLEInterface, DiscoveredPeer +from ble_reticulum.BLEFragmentation import BLEFragmenter, BLEReassembler import time @@ -172,8 +173,8 @@ class TestIdentityHandshakeBasics: # Create fragmenter and peer interface (simulating post-handshake state) frag_key = interface._get_fragmenter_key(existing_identity, central_address) - interface.fragmenters[frag_key] = interface._create_fragmenter(185) - interface.reassemblers[frag_key] = interface._create_reassembler() + interface.fragmenters[frag_key] = BLEFragmenter(mtu=185) + interface.reassemblers[frag_key] = BLEReassembler() # Receive 16-byte data packet (should be processed as data, not handshake) data_packet = b'\xaa\xbb\xcc\xdd\xee\xff\x11\x22\x33\x44\x55\x66\x77\x88\x99\x00' @@ -273,11 +274,7 @@ class TestIdentityHandshakeBidirectional: peripheral_driver = MockBLEDriver(local_address="BB:BB:BB:BB:BB:BB") MockBLEDriver.link_drivers(central_driver, peripheral_driver) - # Set peripheral identity - peripheral_identity = b'\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11' - peripheral_driver.set_identity(peripheral_identity) - - # Start both drivers + # Start both drivers first (sets up characteristic UUIDs) central_driver.start( service_uuid="test-uuid", rx_char_uuid="rx-uuid", @@ -291,6 +288,10 @@ class TestIdentityHandshakeBidirectional: identity_char_uuid="identity-uuid" ) + # Set peripheral identity (after start() so characteristic UUID is available) + peripheral_identity = b'\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11' + peripheral_driver.set_identity(peripheral_identity) + # Central connects to peripheral central_driver.connect(peripheral_driver.local_address) diff --git a/tests/test_v2_2_race_conditions.py b/tests/test_v2_2_race_conditions.py index afa13a2..460347b 100644 --- a/tests/test_v2_2_race_conditions.py +++ b/tests/test_v2_2_race_conditions.py @@ -64,11 +64,9 @@ if not hasattr(RNS, 'Identity'): RNS.Identity = MagicMock() RNS.Identity.full_hash = lambda x: (x * 2)[:16] -# Mock ble_reticulum.Interface (required by BLEInterface.py) -if 'ble_reticulum' not in _sys.modules: - rns_interfaces_mock = MagicMock() - _sys.modules['ble_reticulum'] = rns_interfaces_mock - +# Mock ble_reticulum.Interface module (the base class module, not the whole namespace) +# We only mock the Interface.py module, allowing BLEInterface.py to be imported from src/ +if 'ble_reticulum.Interface' not in _sys.modules: # Create mock Interface base class class MockInterface: MODE_FULL = 1 @@ -77,7 +75,40 @@ if 'ble_reticulum' not in _sys.modules: self.OUT = True self.online = False - rns_interfaces_mock.Interface = MockInterface + @staticmethod + def get_config_obj(configuration): + """Mock config object wrapper - just returns a dict-like object.""" + class ConfigObj: + def __init__(self, config): + self._config = config if config else {} + + def __getitem__(self, key): + return self._config.get(key) + + def get(self, key, default=None): + return self._config.get(key, default) + + def as_string(self, key, default=None): + val = self._config.get(key, default) + return str(val) if val is not None else default + + def as_int(self, key, default=None): + val = self._config.get(key, default) + return int(val) if val is not None else default + + def as_bool(self, key, default=False): + val = self._config.get(key, default) + if isinstance(val, bool): + return val + if isinstance(val, str): + return val.lower() in ('true', 'yes', '1', 'on') + return bool(val) if val is not None else default + return ConfigObj(configuration) + + # Create a mock module for ble_reticulum.Interface + interface_module = MagicMock() + interface_module.Interface = MockInterface + _sys.modules['ble_reticulum.Interface'] = interface_module from tests.mock_ble_driver import MockBLEDriver from ble_reticulum.BLEInterface import BLEInterface, DiscoveredPeer @@ -121,7 +152,8 @@ class TestRateLimiting: def test_connection_allowed_after_5_seconds(self): """Test that connection is allowed after 5-second cooldown.""" - driver = MockBLEDriver(local_address="AA:BB:CC:DD:EE:FF") + # Use local MAC lower than peer MAC so connection direction allows us to initiate + driver = MockBLEDriver(local_address="11:11:11:11:11:11") owner = MockOwner() config = {"name": "Test", "enable_central": True} @@ -129,7 +161,7 @@ class TestRateLimiting: interface.driver = driver interface.local_address = driver.local_address - peer_address = "11:22:33:44:55:66" + peer_address = "22:22:22:22:22:22" peer = DiscoveredPeer(peer_address, "TestPeer", -60) # Record connection attempt 6 seconds ago (past cooldown) @@ -146,7 +178,8 @@ class TestRateLimiting: def test_never_attempted_peer_allowed(self): """Test that peer with no prior attempts is allowed.""" - driver = MockBLEDriver(local_address="AA:BB:CC:DD:EE:FF") + # Use local MAC lower than peer MAC so connection direction allows us to initiate + driver = MockBLEDriver(local_address="11:11:11:11:11:11") owner = MockOwner() config = {"name": "Test", "enable_central": True} @@ -154,7 +187,7 @@ class TestRateLimiting: interface.driver = driver interface.local_address = driver.local_address - peer_address = "11:22:33:44:55:66" + peer_address = "22:22:22:22:22:22" peer = DiscoveredPeer(peer_address, "TestPeer", -60) # last_connection_attempt == 0 (never attempted) @@ -208,7 +241,8 @@ class TestDriverStateTracking: def test_multiple_rapid_discoveries_handled(self): """Test that rapid discovery callbacks don't cause duplicate connections.""" - driver = MockBLEDriver(local_address="AA:BB:CC:DD:EE:FF") + # Use local MAC lower than peer MAC so connection direction allows us to initiate + driver = MockBLEDriver(local_address="11:11:11:11:11:11") owner = MockOwner() config = {"name": "Test", "enable_central": True} @@ -216,21 +250,30 @@ class TestDriverStateTracking: interface.driver = driver interface.local_address = driver.local_address - peer_address = "11:22:33:44:55:66" + peer_address = "22:22:22:22:22:22" peer = DiscoveredPeer(peer_address, "TestPeer", -60) - # Simulate rapid discovery callbacks (5 times in quick succession) - for i in range(5): - interface.discovered_peers[peer_address] = peer - interface._select_peers_to_connect() + # Manually record the first connection attempt (simulating what _try_connect_to_peer does) + # This is needed because _select_peers_to_connect() only returns peers to connect, + # it doesn't actually initiate the connection or record the attempt + interface.discovered_peers[peer_address] = peer + first_selection = interface._select_peers_to_connect() - # After first selection, peer should have recorded attempt - # Subsequent selections should be rate-limited + # First selection should include the peer + assert len(first_selection) == 1 + assert first_selection[0].address == peer_address - # Check that last_connection_attempt was recorded + # Record the attempt (simulating what happens when connection is initiated) + peer.record_connection_attempt() + + # Subsequent rapid selections should be rate-limited + for i in range(4): + subsequent_selection = interface._select_peers_to_connect() + # Should be empty because peer was just attempted + assert len(subsequent_selection) == 0, f"Selection {i+2} should be empty due to rate limiting" + + # Verify timestamp was recorded assert peer.last_connection_attempt > 0 - - # Verify recent timestamp time_since = time.time() - peer.last_connection_attempt assert time_since < 1.0 # Should be very recent