492 lines
15 KiB
Python
492 lines
15 KiB
Python
|
"""
|
||
|
|
Automated Multi-Device Simulation Tests
|
||
|
|
|
||
|
|
Tests the BLE multi-device simulation framework to ensure:
|
||
|
|
- Mock BLE components work correctly
|
||
|
|
- Two nodes can discover and connect
|
||
|
|
- Data transfer works bidirectionally
|
||
|
|
- Fragmentation works with large packets
|
||
|
|
- Multiple transfer scenarios work
|
||
|
|
|
||
|
|
These tests use the simulation framework (no real BLE hardware required).
|
||
|
|
"""
|
||
|
|
|
||
|
|
import sys
|
||
|
|
import os
|
||
|
|
import pytest
|
||
|
|
import asyncio
|
||
|
|
from unittest.mock import Mock, patch
|
||
|
|
|
||
|
|
# Add project paths
|
||
|
|
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
sys.path.insert(0, os.path.join(project_root, 'src'))
|
||
|
|
sys.path.insert(0, os.path.join(project_root, 'examples'))
|
||
|
|
|
||
|
|
from two_device_simulator import (
|
||
|
|
MockBLEConnection,
|
||
|
|
MockBLEDevice,
|
||
|
|
SimulatedBLENode,
|
||
|
|
TwoDeviceSimulator
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Mock BLE Component Tests
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
class TestMockBLEComponents:
|
||
|
|
"""Test individual mock BLE components."""
|
||
|
|
|
||
|
|
def test_mock_device_creation(self):
|
||
|
|
"""Test MockBLEDevice can be created with correct attributes."""
|
||
|
|
device = MockBLEDevice(
|
||
|
|
address="AA:BB:CC:DD:EE:01",
|
||
|
|
name="Test-Device",
|
||
|
|
rssi=-65
|
||
|
|
)
|
||
|
|
|
||
|
|
assert device.address == "AA:BB:CC:DD:EE:01"
|
||
|
|
assert device.name == "Test-Device"
|
||
|
|
assert device.rssi == -65
|
||
|
|
assert "00000001-5824-4f48-9e1a-3b3e8f0c1234" in device.metadata["uuids"]
|
||
|
|
assert device.metadata["rssi"] == -65
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_mock_connection_lifecycle(self):
|
||
|
|
"""Test MockBLEConnection connect/disconnect."""
|
||
|
|
conn = MockBLEConnection("Node-A", "Node-B", mtu=185)
|
||
|
|
|
||
|
|
# Initially not connected
|
||
|
|
assert not conn.connected
|
||
|
|
|
||
|
|
# Connect
|
||
|
|
await conn.connect()
|
||
|
|
assert conn.connected
|
||
|
|
|
||
|
|
# Disconnect
|
||
|
|
await conn.disconnect()
|
||
|
|
assert not conn.connected
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_mock_connection_data_transfer(self):
|
||
|
|
"""Test data transfer between two MockBLEConnections."""
|
||
|
|
conn_a = MockBLEConnection("Node-A", "Node-B", mtu=185)
|
||
|
|
conn_b = MockBLEConnection("Node-B", "Node-A", mtu=185)
|
||
|
|
|
||
|
|
# Link them together
|
||
|
|
conn_a.set_peer(conn_b)
|
||
|
|
conn_b.set_peer(conn_a)
|
||
|
|
|
||
|
|
# Connect both
|
||
|
|
await conn_a.connect()
|
||
|
|
await conn_b.connect()
|
||
|
|
|
||
|
|
# Setup receiver
|
||
|
|
received = []
|
||
|
|
async def rx_callback(data):
|
||
|
|
received.append(data)
|
||
|
|
|
||
|
|
conn_b.set_rx_callback(rx_callback)
|
||
|
|
|
||
|
|
# Send data A → B
|
||
|
|
test_data = b"Hello from A!"
|
||
|
|
await conn_a.write(test_data)
|
||
|
|
await asyncio.sleep(0.01) # Allow delivery
|
||
|
|
|
||
|
|
assert len(received) == 1
|
||
|
|
assert received[0] == test_data
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_mock_connection_rejects_oversized_data(self):
|
||
|
|
"""Test that data exceeding MTU is rejected."""
|
||
|
|
conn = MockBLEConnection("Node-A", "Node-B", mtu=185)
|
||
|
|
await conn.connect()
|
||
|
|
|
||
|
|
oversized_data = b"X" * 200 # Exceeds MTU of 185
|
||
|
|
|
||
|
|
with pytest.raises(ValueError, match="exceeds MTU"):
|
||
|
|
await conn.write(oversized_data)
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_mock_connection_rejects_write_when_disconnected(self):
|
||
|
|
"""Test that writing to disconnected connection fails."""
|
||
|
|
conn = MockBLEConnection("Node-A", "Node-B", mtu=185)
|
||
|
|
|
||
|
|
# Not connected
|
||
|
|
with pytest.raises(RuntimeError, match="not connected"):
|
||
|
|
await conn.write(b"Test")
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bidirectional_data_transfer(self):
|
||
|
|
"""Test data can flow in both directions."""
|
||
|
|
conn_a = MockBLEConnection("Node-A", "Node-B", mtu=185)
|
||
|
|
conn_b = MockBLEConnection("Node-B", "Node-A", mtu=185)
|
||
|
|
|
||
|
|
conn_a.set_peer(conn_b)
|
||
|
|
conn_b.set_peer(conn_a)
|
||
|
|
|
||
|
|
await conn_a.connect()
|
||
|
|
await conn_b.connect()
|
||
|
|
|
||
|
|
# Setup receivers
|
||
|
|
received_a = []
|
||
|
|
received_b = []
|
||
|
|
|
||
|
|
async def rx_callback_a(data):
|
||
|
|
received_a.append(data)
|
||
|
|
|
||
|
|
async def rx_callback_b(data):
|
||
|
|
received_b.append(data)
|
||
|
|
|
||
|
|
conn_a.set_rx_callback(rx_callback_a)
|
||
|
|
conn_b.set_rx_callback(rx_callback_b)
|
||
|
|
|
||
|
|
# A → B
|
||
|
|
await conn_a.write(b"A to B")
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
|
||
|
|
# B → A
|
||
|
|
await conn_b.write(b"B to A")
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
|
||
|
|
assert len(received_b) == 1
|
||
|
|
assert received_b[0] == b"A to B"
|
||
|
|
assert len(received_a) == 1
|
||
|
|
assert received_a[0] == b"B to A"
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Simulated Node Tests
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
class TestSimulatedBLENode:
|
||
|
|
"""Test SimulatedBLENode functionality."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_node_discovery(self):
|
||
|
|
"""Test that a node can discover its peer."""
|
||
|
|
node = SimulatedBLENode(
|
||
|
|
name="Node-A",
|
||
|
|
address="AA:BB:CC:DD:EE:01",
|
||
|
|
peer_address="AA:BB:CC:DD:EE:02",
|
||
|
|
peer_name="Node-B"
|
||
|
|
)
|
||
|
|
|
||
|
|
devices = await node.discover_peers()
|
||
|
|
|
||
|
|
assert len(devices) == 1
|
||
|
|
assert devices[0].address == "AA:BB:CC:DD:EE:02"
|
||
|
|
assert devices[0].name == "Node-B"
|
||
|
|
|
||
|
|
def test_node_connection_creation(self):
|
||
|
|
"""Test that a node can create a connection."""
|
||
|
|
node = SimulatedBLENode(
|
||
|
|
name="Node-A",
|
||
|
|
address="AA:BB:CC:DD:EE:01",
|
||
|
|
peer_address="AA:BB:CC:DD:EE:02",
|
||
|
|
peer_name="Node-B"
|
||
|
|
)
|
||
|
|
|
||
|
|
conn = node.create_connection(mtu=247)
|
||
|
|
|
||
|
|
assert conn is not None
|
||
|
|
assert conn.name == "Node-A"
|
||
|
|
assert conn.peer_name == "Node-B"
|
||
|
|
assert conn.mtu == 247
|
||
|
|
|
||
|
|
def test_node_connection_singleton(self):
|
||
|
|
"""Test that creating connection twice returns same instance."""
|
||
|
|
node = SimulatedBLENode(
|
||
|
|
name="Node-A",
|
||
|
|
address="AA:BB:CC:DD:EE:01",
|
||
|
|
peer_address="AA:BB:CC:DD:EE:02",
|
||
|
|
peer_name="Node-B"
|
||
|
|
)
|
||
|
|
|
||
|
|
conn1 = node.create_connection()
|
||
|
|
conn2 = node.create_connection()
|
||
|
|
|
||
|
|
assert conn1 is conn2
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Two-Device Simulator Tests
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
class TestTwoDeviceSimulator:
|
||
|
|
"""Test the complete two-device simulator."""
|
||
|
|
|
||
|
|
def test_simulator_initialization(self):
|
||
|
|
"""Test that simulator creates two nodes correctly."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
|
||
|
|
assert sim.node_a is not None
|
||
|
|
assert sim.node_b is not None
|
||
|
|
assert sim.node_a.address == "AA:BB:CC:DD:EE:01"
|
||
|
|
assert sim.node_b.address == "AA:BB:CC:DD:EE:02"
|
||
|
|
assert sim.node_a.peer_address == sim.node_b.address
|
||
|
|
assert sim.node_b.peer_address == sim.node_a.address
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_simulator_discovery(self):
|
||
|
|
"""Test discovery test scenario."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
success = await sim.run_discovery_test()
|
||
|
|
# run_discovery_test uses assertions internally, if it returns it passed
|
||
|
|
assert success is None # Function doesn't return, just completes
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_simulator_connection(self):
|
||
|
|
"""Test connection establishment."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
assert sim.node_a.connection.connected
|
||
|
|
assert sim.node_b.connection.connected
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_simulator_data_transfer(self):
|
||
|
|
"""Test data transfer between nodes."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
# Setup receiver
|
||
|
|
received = []
|
||
|
|
async def rx_callback(data):
|
||
|
|
received.append(data)
|
||
|
|
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback)
|
||
|
|
|
||
|
|
# Send data
|
||
|
|
test_data = b"Test packet"
|
||
|
|
await sim.node_a.connection.write(test_data)
|
||
|
|
await asyncio.sleep(0.1)
|
||
|
|
|
||
|
|
assert len(received) == 1
|
||
|
|
assert received[0] == test_data
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_simulator_fragmentation(self):
|
||
|
|
"""Test fragmentation of large packets."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
# Large packet that requires fragmentation
|
||
|
|
large_data = b"X" * 500
|
||
|
|
mtu = sim.node_a.connection.mtu
|
||
|
|
expected_fragments = (len(large_data) + mtu - 1) // mtu
|
||
|
|
|
||
|
|
received_fragments = []
|
||
|
|
async def rx_callback(data):
|
||
|
|
received_fragments.append(data)
|
||
|
|
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback)
|
||
|
|
|
||
|
|
# Send in fragments
|
||
|
|
for i in range(expected_fragments):
|
||
|
|
start = i * mtu
|
||
|
|
end = min(start + mtu, len(large_data))
|
||
|
|
fragment = large_data[start:end]
|
||
|
|
await sim.node_a.connection.write(fragment)
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
|
||
|
|
# Verify all fragments received
|
||
|
|
assert len(received_fragments) == expected_fragments
|
||
|
|
|
||
|
|
# Verify reconstruction works
|
||
|
|
reconstructed = b''.join(received_fragments)
|
||
|
|
assert reconstructed == large_data
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_simulator_all_tests(self):
|
||
|
|
"""Test that all simulator tests pass."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
success = await sim.run_all_tests()
|
||
|
|
assert success is True
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Integration Scenarios
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
class TestIntegrationScenarios:
|
||
|
|
"""Test various integration scenarios."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_rapid_transfers(self):
|
||
|
|
"""Test rapid back-and-forth transfers."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
received_a = []
|
||
|
|
received_b = []
|
||
|
|
|
||
|
|
async def rx_callback_a(data):
|
||
|
|
received_a.append(data)
|
||
|
|
|
||
|
|
async def rx_callback_b(data):
|
||
|
|
received_b.append(data)
|
||
|
|
|
||
|
|
sim.node_a.connection.set_rx_callback(rx_callback_a)
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback_b)
|
||
|
|
|
||
|
|
# Send 10 packets each direction
|
||
|
|
for i in range(10):
|
||
|
|
await sim.node_a.connection.write(f"A→B {i}".encode())
|
||
|
|
await sim.node_b.connection.write(f"B→A {i}".encode())
|
||
|
|
await asyncio.sleep(0.001)
|
||
|
|
|
||
|
|
await asyncio.sleep(0.1) # Allow all deliveries
|
||
|
|
|
||
|
|
assert len(received_b) == 10
|
||
|
|
assert len(received_a) == 10
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_various_packet_sizes(self):
|
||
|
|
"""Test various packet sizes."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
test_sizes = [1, 10, 50, 100, 185] # Up to MTU
|
||
|
|
received = []
|
||
|
|
|
||
|
|
async def rx_callback(data):
|
||
|
|
received.append(len(data))
|
||
|
|
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback)
|
||
|
|
|
||
|
|
for size in test_sizes:
|
||
|
|
data = b"X" * size
|
||
|
|
await sim.node_a.connection.write(data)
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
|
||
|
|
assert received == test_sizes
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_connection_disconnect_reconnect(self):
|
||
|
|
"""Test disconnection and reconnection."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
# Verify connected
|
||
|
|
assert sim.node_a.connection.connected
|
||
|
|
|
||
|
|
# Disconnect
|
||
|
|
await sim.node_a.connection.disconnect()
|
||
|
|
assert not sim.node_a.connection.connected
|
||
|
|
|
||
|
|
# Reconnect
|
||
|
|
await sim.node_a.connection.connect()
|
||
|
|
assert sim.node_a.connection.connected
|
||
|
|
|
||
|
|
# Data transfer should work again
|
||
|
|
received = []
|
||
|
|
async def rx_callback(data):
|
||
|
|
received.append(data)
|
||
|
|
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback)
|
||
|
|
await sim.node_a.connection.write(b"After reconnect")
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
|
||
|
|
assert len(received) == 1
|
||
|
|
assert received[0] == b"After reconnect"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_empty_data_transfer(self):
|
||
|
|
"""Test that empty data can be sent (edge case)."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
received = []
|
||
|
|
async def rx_callback(data):
|
||
|
|
received.append(data)
|
||
|
|
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback)
|
||
|
|
|
||
|
|
# Send empty data
|
||
|
|
await sim.node_a.connection.write(b"")
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
|
||
|
|
assert len(received) == 1
|
||
|
|
assert received[0] == b""
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Performance Tests
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
class TestPerformance:
|
||
|
|
"""Test performance characteristics of simulation."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_throughput_simulation(self):
|
||
|
|
"""Test sustained throughput in simulation."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
packet_count = 100
|
||
|
|
packet_size = 100
|
||
|
|
received_count = 0
|
||
|
|
|
||
|
|
async def rx_callback(data):
|
||
|
|
nonlocal received_count
|
||
|
|
received_count += 1
|
||
|
|
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback)
|
||
|
|
|
||
|
|
# Send many packets
|
||
|
|
start = asyncio.get_event_loop().time()
|
||
|
|
for i in range(packet_count):
|
||
|
|
data = b"X" * packet_size
|
||
|
|
await sim.node_a.connection.write(data)
|
||
|
|
|
||
|
|
await asyncio.sleep(0.5) # Allow delivery
|
||
|
|
end = asyncio.get_event_loop().time()
|
||
|
|
|
||
|
|
duration = end - start
|
||
|
|
assert received_count == packet_count
|
||
|
|
assert duration < 2.0 # Should be fast in simulation
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_large_packet_fragmentation_performance(self):
|
||
|
|
"""Test performance with large packets requiring fragmentation."""
|
||
|
|
sim = TwoDeviceSimulator()
|
||
|
|
await sim.setup_connections()
|
||
|
|
|
||
|
|
# Very large packet (2KB)
|
||
|
|
large_data = b"X" * 2000
|
||
|
|
mtu = sim.node_a.connection.mtu
|
||
|
|
fragments_needed = (len(large_data) + mtu - 1) // mtu
|
||
|
|
|
||
|
|
received_fragments = []
|
||
|
|
async def rx_callback(data):
|
||
|
|
received_fragments.append(data)
|
||
|
|
|
||
|
|
sim.node_b.connection.set_rx_callback(rx_callback)
|
||
|
|
|
||
|
|
# Send fragments
|
||
|
|
start = asyncio.get_event_loop().time()
|
||
|
|
for i in range(fragments_needed):
|
||
|
|
start_pos = i * mtu
|
||
|
|
end_pos = min(start_pos + mtu, len(large_data))
|
||
|
|
fragment = large_data[start_pos:end_pos]
|
||
|
|
await sim.node_a.connection.write(fragment)
|
||
|
|
|
||
|
|
await asyncio.sleep(0.5) # Allow delivery
|
||
|
|
end = asyncio.get_event_loop().time()
|
||
|
|
|
||
|
|
duration = end - start
|
||
|
|
assert len(received_fragments) == fragments_needed
|
||
|
|
assert duration < 2.0 # Should be fast
|
||
|
|
|
||
|
|
# Verify reconstruction
|
||
|
|
reconstructed = b''.join(received_fragments)
|
||
|
|
assert reconstructed == large_data
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Run Tests
|
||
|
|
# ============================================================================
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
pytest.main([__file__, "-v", "--tb=short"])
|