From 921dce2dbaf2c5f98444f41b8639b797230f95c7 Mon Sep 17 00:00:00 2001 From: torlando-tech Date: Thu, 1 Jan 2026 19:32:17 -0500 Subject: [PATCH] test: add comprehensive tests for interface cleanup methods MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add test coverage for new BLEInterface cleanup functionality: - _cleanup_pending_identity_connections: timeout handling for non-Reticulum devices - _process_pending_detaches: delayed interface detachment with grace period - _validate_spawned_interfaces: orphaned interface cleanup - Multi-address disconnect handling (keeping interface alive for MAC rotation) - Thread-safe _spawn_peer_interface with locking - Central disconnect callback with address fallback These tests cover the 195 lines of new code in BLEInterface.py to improve patch coverage for PR #35. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- tests/test_interface_cleanup.py | 701 ++++++++++++++++++++++++++++++++ 1 file changed, 701 insertions(+) create mode 100644 tests/test_interface_cleanup.py diff --git a/tests/test_interface_cleanup.py b/tests/test_interface_cleanup.py new file mode 100644 index 0000000..dfb3a3b --- /dev/null +++ b/tests/test_interface_cleanup.py @@ -0,0 +1,701 @@ +""" +Tests for BLEInterface cleanup methods. + +These tests cover the new cleanup functionality added to handle: +1. Pending identity connection timeouts (non-Reticulum devices) +2. Delayed interface detachment with grace period (MAC rotation) +3. Orphaned interface validation (race condition protection) +4. Multi-address identity handling (same identity at multiple addresses) + +These tests exercise the REAL BLEInterface code with mocked external dependencies. +""" + +import pytest +import time +import threading +from unittest.mock import Mock, MagicMock, patch, PropertyMock +import sys +import os + +# Ensure src is in path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../src')) + + +@pytest.fixture +def mock_rns(): + """Mock RNS module for testing.""" + with patch.dict('sys.modules', {'RNS': MagicMock()}): + import sys + rns = sys.modules['RNS'] + rns.LOG_CRITICAL = 0 + rns.LOG_ERROR = 1 + rns.LOG_WARNING = 2 + rns.LOG_NOTICE = 3 + rns.LOG_INFO = 4 + rns.LOG_VERBOSE = 5 + rns.LOG_DEBUG = 6 + rns.LOG_EXTREME = 7 + rns.log = Mock() + rns.prettyhexrep = lambda x: x.hex() if isinstance(x, bytes) else str(x) + rns.Transport = Mock() + rns.Transport.interfaces = [] + yield rns + + +@pytest.fixture +def mock_driver(): + """Create a mock BLE driver.""" + driver = Mock() + driver.on_device_connected = None + driver.on_device_disconnected = None + driver.on_data_received = None + driver.on_identity_received = None + driver.on_error = None + driver.on_duplicate_identity_detected = None + driver.on_address_changed = None + driver.request_identity_resync = Mock(return_value=False) + driver.get_cached_identity = Mock(return_value=None) + driver.disconnect = Mock() + driver.connected_peers = [] + return driver + + +@pytest.fixture +def ble_interface(mock_rns, mock_driver): + """ + Create a BLEInterface instance with mocked dependencies. + + This uses the REAL BLEInterface class but mocks external dependencies + to allow testing the cleanup logic. + """ + try: + from ble_reticulum.BLEInterface import BLEInterface + + # Create interface without calling __init__ + interface = object.__new__(BLEInterface) + + # Initialize all required attributes manually + interface.name = "TestBLE" + interface.owner = Mock() + interface.online = True + interface.driver = mock_driver + + # Identity mappings + interface.peers = {} + interface.spawned_interfaces = {} + interface.address_to_identity = {} + interface.identity_to_address = {} + interface.address_to_interface = {} + + # Identity cache + interface._identity_cache = {} + interface._identity_cache_ttl = 60 + + # Pending identity connections + interface._pending_identity_connections = {} + interface._pending_identity_timeout = 30 + + # Pending detachments + interface._pending_detach = {} + interface._pending_detach_grace_period = 2.0 + + # Fragmentation + interface.fragmenters = {} + interface.reassemblers = {} + interface.pending_mtu = {} + + # Locks + interface.peer_lock = threading.RLock() + interface.frag_lock = threading.RLock() + + # Other attributes + interface.HW_MTU = 500 + interface.MIN_MTU = 20 + interface.bitrate = 700000 + interface.rxb = 0 + interface.txb = 0 + interface.OUT = True + interface.IN = True + + # Add required methods + interface._compute_identity_hash = lambda x: x[:8].hex() + interface._get_fragmenter_key = lambda ident, addr: f"{ident[:8].hex()}" + + return interface + + except ImportError as e: + pytest.skip(f"Cannot import BLEInterface (RNS not installed): {e}") + + +class TestCleanupPendingIdentityConnections: + """Test _cleanup_pending_identity_connections() method.""" + + def test_no_pending_connections(self, ble_interface, mock_rns): + """Should handle empty pending connections gracefully.""" + assert len(ble_interface._pending_identity_connections) == 0 + ble_interface._cleanup_pending_identity_connections() + # Should not crash, no disconnects called + ble_interface.driver.disconnect.assert_not_called() + + def test_connection_not_timed_out(self, ble_interface, mock_rns): + """Connections within timeout should not be disconnected.""" + mac = "AA:BB:CC:DD:EE:FF" + # Connection started 5 seconds ago (within 30s timeout) + ble_interface._pending_identity_connections[mac] = time.time() - 5 + + ble_interface._cleanup_pending_identity_connections() + + # Should not disconnect + ble_interface.driver.disconnect.assert_not_called() + # Should still be tracked + assert mac in ble_interface._pending_identity_connections + + def test_connection_timed_out(self, ble_interface, mock_rns): + """Connections past timeout should be disconnected.""" + mac = "AA:BB:CC:DD:EE:FF" + # Connection started 35 seconds ago (past 30s timeout) + ble_interface._pending_identity_connections[mac] = time.time() - 35 + + ble_interface._cleanup_pending_identity_connections() + + # Should disconnect + ble_interface.driver.disconnect.assert_called_once_with(mac) + # Should be removed from tracking + assert mac not in ble_interface._pending_identity_connections + + def test_multiple_connections_mixed_states(self, ble_interface, mock_rns): + """Should handle mix of timed-out and valid connections.""" + mac_expired1 = "AA:BB:CC:DD:EE:01" + mac_valid = "AA:BB:CC:DD:EE:02" + mac_expired2 = "AA:BB:CC:DD:EE:03" + + ble_interface._pending_identity_connections[mac_expired1] = time.time() - 40 + ble_interface._pending_identity_connections[mac_valid] = time.time() - 5 + ble_interface._pending_identity_connections[mac_expired2] = time.time() - 35 + + ble_interface._cleanup_pending_identity_connections() + + # Should disconnect both expired connections + assert ble_interface.driver.disconnect.call_count == 2 + # Valid connection should remain + assert mac_valid in ble_interface._pending_identity_connections + assert mac_expired1 not in ble_interface._pending_identity_connections + assert mac_expired2 not in ble_interface._pending_identity_connections + + def test_disconnect_error_handled(self, ble_interface, mock_rns): + """Errors during disconnect should be caught and logged.""" + mac = "AA:BB:CC:DD:EE:FF" + ble_interface._pending_identity_connections[mac] = time.time() - 35 + ble_interface.driver.disconnect.side_effect = Exception("BLE disconnect failed") + + # Should not raise + ble_interface._cleanup_pending_identity_connections() + + # Connection should still be removed from tracking + assert mac not in ble_interface._pending_identity_connections + + +class TestProcessPendingDetaches: + """Test _process_pending_detaches() method.""" + + def test_no_pending_detaches(self, ble_interface, mock_rns): + """Should handle empty pending detaches gracefully.""" + assert len(ble_interface._pending_detach) == 0 + ble_interface._process_pending_detaches() + # Should not crash + + def test_detach_within_grace_period(self, ble_interface, mock_rns): + """Detaches within grace period should not execute.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + + # Scheduled 0.5 seconds ago (within 2s grace period) + ble_interface._pending_detach[identity_hash] = time.time() - 0.5 + + # Create mock interface + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + + ble_interface._process_pending_detaches() + + # Should not detach yet + mock_peer_if.detach.assert_not_called() + assert identity_hash in ble_interface._pending_detach + assert identity_hash in ble_interface.spawned_interfaces + + def test_detach_after_grace_period_no_reconnect(self, ble_interface, mock_rns): + """Detaches past grace period should execute if no reconnection.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + + # Scheduled 3 seconds ago (past 2s grace period) + ble_interface._pending_detach[identity_hash] = time.time() - 3 + ble_interface.identity_to_address[identity_hash] = "AA:BB:CC:DD:EE:FF" + + # Create mock interface + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + + ble_interface._process_pending_detaches() + + # Should detach + mock_peer_if.detach.assert_called_once() + assert identity_hash not in ble_interface._pending_detach + assert identity_hash not in ble_interface.spawned_interfaces + assert identity_hash not in ble_interface.identity_to_address + + def test_detach_cancelled_if_reconnected(self, ble_interface, mock_rns): + """Detach should be cancelled if address reconnected during grace period.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + + # Scheduled 3 seconds ago (past 2s grace period) + ble_interface._pending_detach[identity_hash] = time.time() - 3 + + # But address reconnected (identity mapping exists) + ble_interface.address_to_identity[mac] = identity + + # Create mock interface + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + + ble_interface._process_pending_detaches() + + # Should NOT detach (reconnected during grace period) + mock_peer_if.detach.assert_not_called() + assert identity_hash not in ble_interface._pending_detach # Pending cleared + assert identity_hash in ble_interface.spawned_interfaces # Interface kept + + def test_detach_cleans_up_fragmenter(self, ble_interface, mock_rns): + """Detach should clean up fragmenter and reassembler.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + frag_key = identity_hash + + # Scheduled 3 seconds ago (past 2s grace period) + ble_interface._pending_detach[identity_hash] = time.time() - 3 + ble_interface.identity_to_address[identity_hash] = "AA:BB:CC:DD:EE:FF" + + # Create mock interface + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + + # Create mock fragmenter/reassembler + ble_interface.fragmenters[frag_key] = Mock() + ble_interface.reassemblers[frag_key] = Mock() + + ble_interface._process_pending_detaches() + + # Fragmenter/reassembler should be cleaned up + assert frag_key not in ble_interface.fragmenters + assert frag_key not in ble_interface.reassemblers + + def test_detach_interface_already_gone(self, ble_interface, mock_rns): + """Should handle case where interface was already removed.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + + # Scheduled 3 seconds ago (past 2s grace period) + ble_interface._pending_detach[identity_hash] = time.time() - 3 + + # No interface exists (already cleaned up elsewhere) + assert identity_hash not in ble_interface.spawned_interfaces + + # Should not crash + ble_interface._process_pending_detaches() + + # Pending entry should be cleared + assert identity_hash not in ble_interface._pending_detach + + +class TestValidateSpawnedInterfaces: + """Test _validate_spawned_interfaces() method.""" + + def test_no_orphaned_interfaces(self, ble_interface, mock_rns): + """Should handle case with no orphaned interfaces.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + + # Interface with connected address + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac] = mock_peer_if + ble_interface.address_to_identity[mac] = identity + ble_interface.identity_to_address[identity_hash] = mac + + # Driver shows address connected + ble_interface.driver.connected_peers = [mac] + + ble_interface._validate_spawned_interfaces() + + # Nothing should be detached + mock_peer_if.detach.assert_not_called() + assert identity_hash in ble_interface.spawned_interfaces + assert mac in ble_interface.address_to_interface + + def test_orphaned_address_mapping_cleaned(self, ble_interface, mock_rns): + """Should clean up address mappings for disconnected addresses.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac_disconnected = "AA:BB:CC:DD:EE:01" + mac_connected = "AA:BB:CC:DD:EE:02" + + # Interface with two address mappings + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac_disconnected] = mock_peer_if + ble_interface.address_to_interface[mac_connected] = mock_peer_if + ble_interface.address_to_identity[mac_disconnected] = identity + ble_interface.address_to_identity[mac_connected] = identity + ble_interface.identity_to_address[identity_hash] = mac_disconnected + + # Only one address still connected + ble_interface.driver.connected_peers = [mac_connected] + + ble_interface._validate_spawned_interfaces() + + # Disconnected address should be cleaned up + assert mac_disconnected not in ble_interface.address_to_interface + assert mac_disconnected not in ble_interface.address_to_identity + + # Connected address should remain + assert mac_connected in ble_interface.address_to_interface + + # Interface should NOT be detached (other address still connected) + mock_peer_if.detach.assert_not_called() + assert identity_hash in ble_interface.spawned_interfaces + + # identity_to_address should point to connected address + assert ble_interface.identity_to_address[identity_hash] == mac_connected + + def test_orphaned_interface_detached(self, ble_interface, mock_rns): + """Should detach interface when all addresses disconnected.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + frag_key = identity_hash + + # Interface with no connected addresses + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac] = mock_peer_if + ble_interface.address_to_identity[mac] = identity + ble_interface.identity_to_address[identity_hash] = mac + + # Create fragmenter/reassembler + ble_interface.fragmenters[frag_key] = Mock() + ble_interface.reassemblers[frag_key] = Mock() + + # No addresses connected + ble_interface.driver.connected_peers = [] + + ble_interface._validate_spawned_interfaces() + + # Interface should be detached + mock_peer_if.detach.assert_called_once() + assert identity_hash not in ble_interface.spawned_interfaces + assert identity_hash not in ble_interface.identity_to_address + assert mac not in ble_interface.address_to_interface + assert mac not in ble_interface.address_to_identity + + # Fragmenter/reassembler should be cleaned up + assert frag_key not in ble_interface.fragmenters + assert frag_key not in ble_interface.reassemblers + + def test_validation_handles_exception(self, ble_interface, mock_rns): + """Should catch and log exceptions during validation.""" + # Make connected_peers property raise an exception + type(ble_interface.driver).connected_peers = PropertyMock( + side_effect=Exception("BLE driver error") + ) + + # Should not raise + ble_interface._validate_spawned_interfaces() + + +class TestMultiAddressDisconnect: + """Test disconnect callback with multiple addresses per identity.""" + + def test_disconnect_preserves_interface_with_other_addresses(self, ble_interface, mock_rns): + """When one address disconnects, interface should stay if others remain.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac1 = "AA:BB:CC:DD:EE:01" + mac2 = "AA:BB:CC:DD:EE:02" + + # Setup: Two addresses mapped to same identity/interface + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + mock_peer_if.detach = Mock() + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac1] = mock_peer_if + ble_interface.address_to_interface[mac2] = mock_peer_if + ble_interface.address_to_identity[mac1] = identity + ble_interface.address_to_identity[mac2] = identity + ble_interface.identity_to_address[identity_hash] = mac1 + + # Disconnect mac1 + ble_interface._device_disconnected_callback(mac1) + + # Interface should NOT be detached (mac2 still connected) + mock_peer_if.detach.assert_not_called() + assert identity_hash in ble_interface.spawned_interfaces + + # mac1 mappings should be cleaned + assert mac1 not in ble_interface.address_to_interface + assert mac1 not in ble_interface.address_to_identity + + # mac2 should remain + assert mac2 in ble_interface.address_to_interface + assert mac2 in ble_interface.address_to_identity + + # identity_to_address should point to mac2 now + assert ble_interface.identity_to_address[identity_hash] == mac2 + + # No pending detach (other address still connected) + assert identity_hash not in ble_interface._pending_detach + + def test_disconnect_schedules_detach_when_last_address(self, ble_interface, mock_rns): + """When last address disconnects, detach should be scheduled.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + + # Setup: Single address mapped to identity + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + mock_peer_if.detach = Mock() + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac] = mock_peer_if + ble_interface.address_to_identity[mac] = identity + ble_interface.identity_to_address[identity_hash] = mac + + # Disconnect + ble_interface._device_disconnected_callback(mac) + + # Interface should NOT be immediately detached + mock_peer_if.detach.assert_not_called() + + # But detach should be scheduled + assert identity_hash in ble_interface._pending_detach + + # Address mappings should be cleaned immediately + assert mac not in ble_interface.address_to_interface + assert mac not in ble_interface.address_to_identity + + +class TestCentralDisconnected: + """Test handle_central_disconnected with multi-address handling.""" + + def test_central_disconnect_preserves_interface_with_other_addresses(self, ble_interface, mock_rns): + """When central disconnects, interface should stay if other addresses exist.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac_central = "AA:BB:CC:DD:EE:01" + mac_other = "AA:BB:CC:DD:EE:02" + + # Setup: Two addresses mapped to same identity + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + mock_peer_if.detach = Mock() + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac_central] = mock_peer_if + ble_interface.address_to_interface[mac_other] = mock_peer_if + ble_interface.address_to_identity[mac_central] = identity + ble_interface.address_to_identity[mac_other] = identity + ble_interface.identity_to_address[identity_hash] = mac_central + + # Central disconnects + ble_interface.handle_central_disconnected(mac_central) + + # Interface should NOT be detached + mock_peer_if.detach.assert_not_called() + assert identity_hash in ble_interface.spawned_interfaces + + # identity_to_address should point to remaining address + assert ble_interface.identity_to_address[identity_hash] == mac_other + + def test_central_disconnect_uses_address_fallback(self, ble_interface, mock_rns): + """When no identity mapping, should use address_to_interface fallback.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + + # Setup: Only address_to_interface exists (no identity mapping) + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + mock_peer_if.detach = Mock() + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac] = mock_peer_if + ble_interface.identity_to_address[identity_hash] = mac + # Note: address_to_identity is NOT set - testing fallback + + # Central disconnects + ble_interface.handle_central_disconnected(mac) + + # Should still schedule detach via fallback + assert identity_hash in ble_interface._pending_detach + + +class TestSpawnPeerInterfaceThreadSafety: + """Test thread-safety of _spawn_peer_interface.""" + + def test_spawn_interface_reuses_existing(self, ble_interface, mock_rns): + """Should reuse existing interface for same identity at new address.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac_old = "AA:BB:CC:DD:EE:01" + mac_new = "AA:BB:CC:DD:EE:02" + + # Create existing interface + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + mock_peer_if.online = True + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[mac_old] = mock_peer_if + + # Spawn interface for new address with same identity + result = ble_interface._spawn_peer_interface( + mac_new, "RNS-peer", identity, client=None, mtu=185 + ) + + # Should return existing interface + assert result == mock_peer_if + + # New address should be mapped + assert ble_interface.address_to_interface[mac_new] == mock_peer_if + assert ble_interface.address_to_identity[mac_new] == identity + assert ble_interface.identity_to_address[identity_hash] == mac_new + + def test_spawn_interface_cancels_pending_detach(self, ble_interface, mock_rns): + """Spawning interface should cancel any pending detach.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + + # Create existing interface with pending detach + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + mock_peer_if.online = False + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface._pending_detach[identity_hash] = time.time() - 1 + + # Spawn interface (reuse) + result = ble_interface._spawn_peer_interface( + mac, "RNS-peer", identity, client=None, mtu=185 + ) + + # Pending detach should be cancelled + assert identity_hash not in ble_interface._pending_detach + + # Interface should be marked online + assert result.online is True + + +class TestDeviceConnectedCallback: + """Test _device_connected_callback with pending identity handling.""" + + def test_connected_cancels_pending_detach(self, ble_interface, mock_rns): + """Connection with identity should cancel pending detach.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + + # Setup: pending detach exists + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface._pending_detach[identity_hash] = time.time() - 1 + + # Mock driver role + ble_interface.driver.get_peer_role = Mock(return_value="central") + ble_interface._check_duplicate_identity = Mock(return_value=False) + ble_interface._record_connection_success = Mock() + ble_interface._spawn_peer_interface = Mock(return_value=mock_peer_if) + ble_interface._ensure_fragmenter = Mock() + + # Device connects + ble_interface._device_connected_callback(mac, identity) + + # Pending detach should be cancelled + assert identity_hash not in ble_interface._pending_detach + + def test_connected_removes_from_pending_identity(self, ble_interface, mock_rns): + """Connection with identity should remove from pending identity tracking.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + mac = "AA:BB:CC:DD:EE:FF" + + # Setup: pending identity exists + ble_interface._pending_identity_connections[mac] = time.time() - 5 + + # Mock driver role + ble_interface.driver.get_peer_role = Mock(return_value="central") + ble_interface._check_duplicate_identity = Mock(return_value=False) + ble_interface._record_connection_success = Mock() + ble_interface._spawn_peer_interface = Mock(return_value=Mock()) + ble_interface._ensure_fragmenter = Mock() + + # Device connects + ble_interface._device_connected_callback(mac, identity) + + # Should be removed from pending tracking + assert mac not in ble_interface._pending_identity_connections + + def test_peripheral_connection_tracked_for_timeout(self, ble_interface, mock_rns): + """Peripheral connection without identity should be tracked for timeout.""" + mac = "AA:BB:CC:DD:EE:FF" + + # Mock driver role (peripheral = waiting for identity handshake) + ble_interface.driver.get_peer_role = Mock(return_value="peripheral") + + # Device connects as peripheral (no identity yet) + ble_interface._device_connected_callback(mac, None) + + # Should be tracked for identity timeout + assert mac in ble_interface._pending_identity_connections + + +class TestCleanupStaleAddress: + """Test _cleanup_stale_address (renamed from _cleanup_stale_interface).""" + + def test_cleanup_preserves_interface(self, ble_interface, mock_rns): + """Cleanup should preserve interface for reuse.""" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = identity[:8].hex() + old_mac = "AA:BB:CC:DD:EE:01" + + # Setup: interface and mappings + mock_peer_if = Mock() + mock_peer_if.peer_identity = identity + mock_peer_if.detach = Mock() + ble_interface.spawned_interfaces[identity_hash] = mock_peer_if + ble_interface.address_to_interface[old_mac] = mock_peer_if + ble_interface.address_to_identity[old_mac] = identity + ble_interface.identity_to_address[identity_hash] = old_mac + ble_interface.pending_mtu[old_mac] = 185 + + # Cleanup old address + ble_interface._cleanup_stale_address(identity_hash, old_mac) + + # Interface should NOT be detached (preserved for reuse) + mock_peer_if.detach.assert_not_called() + assert identity_hash in ble_interface.spawned_interfaces + + # Old address mappings should be cleaned + assert old_mac not in ble_interface.address_to_interface + assert old_mac not in ble_interface.address_to_identity + assert old_mac not in ble_interface.pending_mtu + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])