diff --git a/tests/mock_ble_driver.py b/tests/mock_ble_driver.py index 23d2ec7..994f967 100644 --- a/tests/mock_ble_driver.py +++ b/tests/mock_ble_driver.py @@ -41,7 +41,7 @@ if src_path not in sys.path: # Import directly using importlib to bypass RNS namespace conflicts # This avoids issues when a real RNS package is installed globally import importlib.util -bluetooth_driver_path = os.path.join(src_path, 'RNS', 'Interfaces', 'bluetooth_driver.py') +bluetooth_driver_path = os.path.join(src_path, 'ble_reticulum', 'bluetooth_driver.py') spec = importlib.util.spec_from_file_location("bluetooth_driver", bluetooth_driver_path) bluetooth_driver = importlib.util.module_from_spec(spec) spec.loader.exec_module(bluetooth_driver) diff --git a/tests/test_identity_cache.py b/tests/test_identity_cache.py index f723bdc..d52968d 100644 --- a/tests/test_identity_cache.py +++ b/tests/test_identity_cache.py @@ -85,8 +85,11 @@ def ble_interface(mock_rns, mock_driver): interface.spawned_interfaces = {} interface.address_to_identity = {} interface.identity_to_address = {} + interface.address_to_interface = {} # address -> BLEPeerInterface interface._identity_cache = {} interface._identity_cache_ttl = 60 + interface._pending_detach = {} # identity_hash -> timestamp + interface._pending_detach_grace_period = 2.0 # seconds # Fragmentation interface.fragmenters = {} @@ -143,12 +146,16 @@ class TestIdentityCacheOnDisconnect: assert cached_identity == identity assert time.time() - cached_time < 2 # Cached recently - # Assert: Active mappings should be cleaned up + # Assert: Address-specific mappings should be cleaned up immediately assert mac not in ble_interface.address_to_identity - assert identity_hash not in ble_interface.identity_to_address - # Assert: Peer interface was detached - mock_peer_if.detach.assert_called_once() + # Assert: identity_to_address and interface are NOT cleaned up immediately + # (grace period allows reconnection with same identity at new address) + assert identity_hash in ble_interface.identity_to_address + + # Assert: Detach is scheduled, not immediate + assert identity_hash in ble_interface._pending_detach + mock_peer_if.detach.assert_not_called() def test_disconnect_unknown_address_no_crash(self, ble_interface, mock_rns): """ diff --git a/tests/test_v2_2_mac_sorting.py b/tests/test_v2_2_mac_sorting.py index 81eee45..ecaf97d 100644 --- a/tests/test_v2_2_mac_sorting.py +++ b/tests/test_v2_2_mac_sorting.py @@ -360,7 +360,7 @@ class TestMACRotationBypassesSorting: list regardless of MAC sorting. Previously, the code fell through to the MAC sorting check which could skip the peer if local MAC > peer MAC. - Fix: After _cleanup_stale_interface(), immediately add peer and continue, + Fix: After _cleanup_stale_address(), immediately add peer and continue, bypassing the MAC sorting check. """ @@ -419,7 +419,7 @@ class TestMACRotationBypassesSorting: "MAC rotation should bypass MAC sorting and add peer" def test_mac_rotation_cleanup_is_called(self): - """Test that _cleanup_stale_interface is called during MAC rotation.""" + """Test that _cleanup_stale_address is called during MAC rotation.""" driver = MockBLEDriver(local_address="FF:FF:FF:FF:FF:FF") owner = MockOwner() @@ -430,13 +430,13 @@ class TestMACRotationBypassesSorting: # Track cleanup calls cleanup_calls = [] - original_cleanup = interface._cleanup_stale_interface + original_cleanup = interface._cleanup_stale_address def tracked_cleanup(identity_hash, old_address): cleanup_calls.append((identity_hash, old_address)) return original_cleanup(identity_hash, old_address) - interface._cleanup_stale_interface = tracked_cleanup + interface._cleanup_stale_address = tracked_cleanup # Set up MAC rotation scenario old_address = "AA:AA:AA:AA:AA:AA"