fix(tests): update tests for driver callback signature and Python 3.14 compatibility
- Fix BLEInterface.handle_peripheral_data to use _compute_identity_hash instead of RNS.Identity.full_hash for consistent identity hash computation - Update MockBLEDriver.on_device_connected callback to match the (address, peer_identity) signature in bluetooth_driver.py - Fix test_v2_2_identity_handshake.py and test_v2_2_race_conditions.py to properly mock ble_reticulum.Interface without breaking the namespace - Use BLEFragmenter/BLEReassembler directly in tests instead of non-existent _create_fragmenter/_create_reassembler methods - Fix asyncio.get_event_loop() deprecation in test_ble_peer_interface.py for Python 3.10+ compatibility - Update MAC address test fixtures to account for v2.2 MAC sorting logic - Fix test_peer_address_mac_rotation to properly simulate MAC rotation where old connection is dropped before new one arrives Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
torlando-tech
parent
1e1023f914
commit
799b91122f
6 changed files with 138 additions and 76 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue