test: add integration tests that exercise real BLEInterface code
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 <noreply@anthropic.com>
This commit is contained in:
parent
b1206b3c6e
commit
d332f9a9bb
1 changed files with 129 additions and 2 deletions
|
|
@ -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):
|
||||
"""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue