From d332f9a9bb0ecf5f608905019058e0787593c36d Mon Sep 17 00:00:00 2001 From: torlando-tech Date: Tue, 30 Dec 2025 14:14:50 -0500 Subject: [PATCH] test: add integration tests that exercise real BLEInterface code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add TestIdentityCacheIntegration class that imports and tests actual BLEInterface methods instead of just mocking the logic. This should provide codecov coverage on the changed lines. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- tests/test_identity_cache.py | 131 ++++++++++++++++++++++++++++++++++- 1 file changed, 129 insertions(+), 2 deletions(-) diff --git a/tests/test_identity_cache.py b/tests/test_identity_cache.py index 521df0e..2490207 100644 --- a/tests/test_identity_cache.py +++ b/tests/test_identity_cache.py @@ -13,15 +13,142 @@ This prevents data loss when: import pytest import time -from unittest.mock import Mock, MagicMock, patch +import threading +from unittest.mock import Mock, MagicMock, patch, AsyncMock import sys import os +import asyncio # Add src to path - conftest.py handles RNS mocking +def create_minimal_ble_interface(): + """ + Create a minimal BLEInterface instance for testing identity cache. + + Mocks external dependencies (driver, RNS) while keeping the real + identity cache logic intact. + """ + try: + from ble_reticulum.BLEInterface import BLEInterface + except ImportError: + pytest.skip("BLEInterface not available") + + with patch('ble_reticulum.BLEInterface.Interface.__init__'): + # Create interface with mocked owner + interface = object.__new__(BLEInterface) + + # Initialize required attributes + interface.name = "TestBLE" + interface.owner = Mock() + interface.online = False + interface.driver = Mock() + interface.driver.on_device_connected = None + interface.driver.on_device_disconnected = None + interface.driver.on_data_received = None + interface.driver.on_identity_received = None + interface.driver.on_error = None + interface.driver.on_duplicate_identity_detected = None + interface.driver.on_address_changed = None + interface.driver.request_identity_resync = Mock(return_value=False) + + # Initialize state dictionaries + interface.peers = {} + interface.spawned_interfaces = {} + interface.address_to_identity = {} + interface.identity_to_address = {} + interface._identity_cache = {} + interface._identity_cache_ttl = 60 + interface.fragmenters = {} + interface.reassemblers = {} + interface.pending_mtu = {} + + # Locks + interface.peer_lock = threading.RLock() + interface.frag_lock = threading.RLock() + + # Other required attributes + interface.HW_MTU = 500 + interface.MIN_MTU = 20 + interface.bitrate = 700000 + interface.rxb = 0 + interface.txb = 0 + + return interface + + return None + + +class TestIdentityCacheIntegration: + """Test identity cache with real BLEInterface methods.""" + + def test_identity_cache_initialized(self): + """BLEInterface should have identity cache attributes.""" + interface = create_minimal_ble_interface() + if interface is None: + pytest.skip("Could not create interface") + + assert hasattr(interface, '_identity_cache') + assert hasattr(interface, '_identity_cache_ttl') + assert interface._identity_cache_ttl == 60 + assert isinstance(interface._identity_cache, dict) + + def test_disconnect_callback_caches_identity(self): + """_device_disconnected_callback should cache identity before cleanup.""" + interface = create_minimal_ble_interface() + if interface is None: + pytest.skip("Could not create interface") + + # Setup: Simulate connected peer + mac = "54:8F:DF:44:79:2B" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = interface._compute_identity_hash(identity) if hasattr(interface, '_compute_identity_hash') else "456c6978" + + interface.address_to_identity[mac] = identity + interface.identity_to_address[identity_hash] = mac + + # Create mock peer interface + mock_peer_if = Mock() + mock_peer_if.detach = Mock() + interface.spawned_interfaces[identity_hash] = mock_peer_if + + # Call the disconnect callback + if hasattr(interface, '_device_disconnected_callback'): + interface._device_disconnected_callback(mac) + + # Assert: Identity should be cached + assert mac in interface._identity_cache + cached_identity, cached_time = interface._identity_cache[mac] + assert cached_identity == identity + assert time.time() - cached_time < 2 + + def test_address_changed_callback_migrates_mappings(self): + """_address_changed_callback should migrate identity to new address.""" + interface = create_minimal_ble_interface() + if interface is None: + pytest.skip("Could not create interface") + + old_mac = "54:8F:DF:44:79:2B" + new_mac = "6B:2B:EE:1A:5B:94" + identity = bytes.fromhex("456c6978823c85e228a545259c241e0e") + identity_hash = "456c6978" + + # Setup initial mapping + interface.address_to_identity[old_mac] = identity + interface.identity_to_address[identity_hash] = old_mac + + # Call address changed callback + if hasattr(interface, '_address_changed_callback'): + interface._address_changed_callback(old_mac, new_mac, identity_hash) + + # Assert: Mapping migrated to new address + assert old_mac not in interface.address_to_identity + assert new_mac in interface.address_to_identity + assert interface.identity_to_address[identity_hash] == new_mac + + class TestIdentityCache: - """Test identity caching for reconnection recovery.""" + """Test identity caching logic (unit tests with mocks).""" def test_identity_cached_on_disconnect(self): """