updates
This commit is contained in:
rUv
663
tests/integration/test_hardware_integration.py
Normal file
663
tests/integration/test_hardware_integration.py
Normal file
@@ -0,0 +1,663 @@
|
||||
"""
|
||||
Integration tests for hardware integration and router communication.
|
||||
|
||||
Tests WiFi router communication, CSI data collection, and hardware management.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
import numpy as np
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, Any, List, Optional
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
import json
|
||||
import socket
|
||||
|
||||
|
||||
class MockRouterInterface:
|
||||
"""Mock WiFi router interface for testing."""
|
||||
|
||||
def __init__(self, router_id: str, ip_address: str = "192.168.1.1"):
|
||||
self.router_id = router_id
|
||||
self.ip_address = ip_address
|
||||
self.is_connected = False
|
||||
self.is_authenticated = False
|
||||
self.csi_streaming = False
|
||||
self.connection_attempts = 0
|
||||
self.last_heartbeat = None
|
||||
self.firmware_version = "1.2.3"
|
||||
self.capabilities = ["csi", "beamforming", "mimo"]
|
||||
|
||||
async def connect(self) -> bool:
|
||||
"""Connect to the router."""
|
||||
self.connection_attempts += 1
|
||||
|
||||
# Simulate connection failure for testing
|
||||
if self.connection_attempts == 1:
|
||||
return False
|
||||
|
||||
await asyncio.sleep(0.1) # Simulate connection time
|
||||
self.is_connected = True
|
||||
return True
|
||||
|
||||
async def authenticate(self, username: str, password: str) -> bool:
|
||||
"""Authenticate with the router."""
|
||||
if not self.is_connected:
|
||||
return False
|
||||
|
||||
# Simulate authentication
|
||||
if username == "admin" and password == "correct_password":
|
||||
self.is_authenticated = True
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
async def start_csi_streaming(self, config: Dict[str, Any]) -> bool:
|
||||
"""Start CSI data streaming."""
|
||||
if not self.is_authenticated:
|
||||
return False
|
||||
|
||||
# This should fail initially to test proper error handling
|
||||
return False
|
||||
|
||||
async def stop_csi_streaming(self) -> bool:
|
||||
"""Stop CSI data streaming."""
|
||||
if self.csi_streaming:
|
||||
self.csi_streaming = False
|
||||
return True
|
||||
return False
|
||||
|
||||
async def get_status(self) -> Dict[str, Any]:
|
||||
"""Get router status."""
|
||||
return {
|
||||
"router_id": self.router_id,
|
||||
"ip_address": self.ip_address,
|
||||
"is_connected": self.is_connected,
|
||||
"is_authenticated": self.is_authenticated,
|
||||
"csi_streaming": self.csi_streaming,
|
||||
"firmware_version": self.firmware_version,
|
||||
"uptime_seconds": 3600,
|
||||
"signal_strength": -45.2,
|
||||
"temperature": 42.5,
|
||||
"cpu_usage": 15.3
|
||||
}
|
||||
|
||||
async def send_heartbeat(self) -> bool:
|
||||
"""Send heartbeat to router."""
|
||||
if not self.is_connected:
|
||||
return False
|
||||
|
||||
self.last_heartbeat = datetime.utcnow()
|
||||
return True
|
||||
|
||||
|
||||
class TestRouterConnection:
|
||||
"""Test router connection functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def router_interface(self):
|
||||
"""Create router interface for testing."""
|
||||
return MockRouterInterface("router_001", "192.168.1.100")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_connection_should_fail_initially(self, router_interface):
|
||||
"""Test router connection - should fail initially."""
|
||||
# First connection attempt should fail
|
||||
result = await router_interface.connect()
|
||||
|
||||
# This will fail initially because we designed the mock to fail first attempt
|
||||
assert result is False
|
||||
assert router_interface.is_connected is False
|
||||
assert router_interface.connection_attempts == 1
|
||||
|
||||
# Second attempt should succeed
|
||||
result = await router_interface.connect()
|
||||
assert result is True
|
||||
assert router_interface.is_connected is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_authentication_should_fail_initially(self, router_interface):
|
||||
"""Test router authentication - should fail initially."""
|
||||
# Connect first
|
||||
await router_interface.connect()
|
||||
await router_interface.connect() # Second attempt succeeds
|
||||
|
||||
# Test wrong credentials
|
||||
result = await router_interface.authenticate("admin", "wrong_password")
|
||||
|
||||
# This will fail initially
|
||||
assert result is False
|
||||
assert router_interface.is_authenticated is False
|
||||
|
||||
# Test correct credentials
|
||||
result = await router_interface.authenticate("admin", "correct_password")
|
||||
assert result is True
|
||||
assert router_interface.is_authenticated is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_csi_streaming_start_should_fail_initially(self, router_interface):
|
||||
"""Test CSI streaming start - should fail initially."""
|
||||
# Setup connection and authentication
|
||||
await router_interface.connect()
|
||||
await router_interface.connect() # Second attempt succeeds
|
||||
await router_interface.authenticate("admin", "correct_password")
|
||||
|
||||
# Try to start CSI streaming
|
||||
config = {
|
||||
"frequency": 5.8e9,
|
||||
"bandwidth": 80e6,
|
||||
"sample_rate": 1000,
|
||||
"antenna_config": "4x4_mimo"
|
||||
}
|
||||
|
||||
result = await router_interface.start_csi_streaming(config)
|
||||
|
||||
# This will fail initially because the mock is designed to return False
|
||||
assert result is False
|
||||
assert router_interface.csi_streaming is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_status_retrieval_should_fail_initially(self, router_interface):
|
||||
"""Test router status retrieval - should fail initially."""
|
||||
status = await router_interface.get_status()
|
||||
|
||||
# This will fail initially
|
||||
assert isinstance(status, dict)
|
||||
assert status["router_id"] == "router_001"
|
||||
assert status["ip_address"] == "192.168.1.100"
|
||||
assert "firmware_version" in status
|
||||
assert "uptime_seconds" in status
|
||||
assert "signal_strength" in status
|
||||
assert "temperature" in status
|
||||
assert "cpu_usage" in status
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_heartbeat_mechanism_should_fail_initially(self, router_interface):
|
||||
"""Test heartbeat mechanism - should fail initially."""
|
||||
# Heartbeat without connection should fail
|
||||
result = await router_interface.send_heartbeat()
|
||||
|
||||
# This will fail initially
|
||||
assert result is False
|
||||
assert router_interface.last_heartbeat is None
|
||||
|
||||
# Connect and try heartbeat
|
||||
await router_interface.connect()
|
||||
await router_interface.connect() # Second attempt succeeds
|
||||
|
||||
result = await router_interface.send_heartbeat()
|
||||
assert result is True
|
||||
assert router_interface.last_heartbeat is not None
|
||||
|
||||
|
||||
class TestMultiRouterManagement:
|
||||
"""Test management of multiple routers."""
|
||||
|
||||
@pytest.fixture
|
||||
def router_manager(self):
|
||||
"""Create router manager for testing."""
|
||||
class RouterManager:
|
||||
def __init__(self):
|
||||
self.routers = {}
|
||||
self.active_connections = 0
|
||||
|
||||
async def add_router(self, router_id: str, ip_address: str) -> bool:
|
||||
"""Add a router to management."""
|
||||
if router_id in self.routers:
|
||||
return False
|
||||
|
||||
router = MockRouterInterface(router_id, ip_address)
|
||||
self.routers[router_id] = router
|
||||
return True
|
||||
|
||||
async def connect_router(self, router_id: str) -> bool:
|
||||
"""Connect to a specific router."""
|
||||
if router_id not in self.routers:
|
||||
return False
|
||||
|
||||
router = self.routers[router_id]
|
||||
|
||||
# Try connecting twice (mock fails first time)
|
||||
success = await router.connect()
|
||||
if not success:
|
||||
success = await router.connect()
|
||||
|
||||
if success:
|
||||
self.active_connections += 1
|
||||
|
||||
return success
|
||||
|
||||
async def authenticate_router(self, router_id: str, username: str, password: str) -> bool:
|
||||
"""Authenticate with a router."""
|
||||
if router_id not in self.routers:
|
||||
return False
|
||||
|
||||
router = self.routers[router_id]
|
||||
return await router.authenticate(username, password)
|
||||
|
||||
async def get_all_status(self) -> Dict[str, Dict[str, Any]]:
|
||||
"""Get status of all routers."""
|
||||
status = {}
|
||||
for router_id, router in self.routers.items():
|
||||
status[router_id] = await router.get_status()
|
||||
return status
|
||||
|
||||
async def start_all_csi_streaming(self, config: Dict[str, Any]) -> Dict[str, bool]:
|
||||
"""Start CSI streaming on all authenticated routers."""
|
||||
results = {}
|
||||
for router_id, router in self.routers.items():
|
||||
if router.is_authenticated:
|
||||
results[router_id] = await router.start_csi_streaming(config)
|
||||
else:
|
||||
results[router_id] = False
|
||||
return results
|
||||
|
||||
return RouterManager()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_router_addition_should_fail_initially(self, router_manager):
|
||||
"""Test adding multiple routers - should fail initially."""
|
||||
# Add first router
|
||||
result1 = await router_manager.add_router("router_001", "192.168.1.100")
|
||||
|
||||
# This will fail initially
|
||||
assert result1 is True
|
||||
assert "router_001" in router_manager.routers
|
||||
|
||||
# Add second router
|
||||
result2 = await router_manager.add_router("router_002", "192.168.1.101")
|
||||
assert result2 is True
|
||||
assert "router_002" in router_manager.routers
|
||||
|
||||
# Try to add duplicate router
|
||||
result3 = await router_manager.add_router("router_001", "192.168.1.102")
|
||||
assert result3 is False
|
||||
assert len(router_manager.routers) == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_router_connections_should_fail_initially(self, router_manager):
|
||||
"""Test concurrent router connections - should fail initially."""
|
||||
# Add multiple routers
|
||||
await router_manager.add_router("router_001", "192.168.1.100")
|
||||
await router_manager.add_router("router_002", "192.168.1.101")
|
||||
await router_manager.add_router("router_003", "192.168.1.102")
|
||||
|
||||
# Connect to all routers concurrently
|
||||
connection_tasks = [
|
||||
router_manager.connect_router("router_001"),
|
||||
router_manager.connect_router("router_002"),
|
||||
router_manager.connect_router("router_003")
|
||||
]
|
||||
|
||||
results = await asyncio.gather(*connection_tasks)
|
||||
|
||||
# This will fail initially
|
||||
assert len(results) == 3
|
||||
assert all(results) # All connections should succeed
|
||||
assert router_manager.active_connections == 3
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_status_aggregation_should_fail_initially(self, router_manager):
|
||||
"""Test router status aggregation - should fail initially."""
|
||||
# Add and connect routers
|
||||
await router_manager.add_router("router_001", "192.168.1.100")
|
||||
await router_manager.add_router("router_002", "192.168.1.101")
|
||||
|
||||
await router_manager.connect_router("router_001")
|
||||
await router_manager.connect_router("router_002")
|
||||
|
||||
# Get all status
|
||||
all_status = await router_manager.get_all_status()
|
||||
|
||||
# This will fail initially
|
||||
assert isinstance(all_status, dict)
|
||||
assert len(all_status) == 2
|
||||
assert "router_001" in all_status
|
||||
assert "router_002" in all_status
|
||||
|
||||
# Verify status structure
|
||||
for router_id, status in all_status.items():
|
||||
assert "router_id" in status
|
||||
assert "ip_address" in status
|
||||
assert "is_connected" in status
|
||||
assert status["is_connected"] is True
|
||||
|
||||
|
||||
class TestCSIDataCollection:
|
||||
"""Test CSI data collection from routers."""
|
||||
|
||||
@pytest.fixture
|
||||
def csi_collector(self):
|
||||
"""Create CSI data collector."""
|
||||
class CSICollector:
|
||||
def __init__(self):
|
||||
self.collected_data = []
|
||||
self.is_collecting = False
|
||||
self.collection_rate = 0
|
||||
|
||||
async def start_collection(self, router_interfaces: List[MockRouterInterface]) -> bool:
|
||||
"""Start CSI data collection."""
|
||||
# This should fail initially
|
||||
return False
|
||||
|
||||
async def stop_collection(self) -> bool:
|
||||
"""Stop CSI data collection."""
|
||||
if self.is_collecting:
|
||||
self.is_collecting = False
|
||||
return True
|
||||
return False
|
||||
|
||||
async def collect_frame(self, router_interface: MockRouterInterface) -> Optional[Dict[str, Any]]:
|
||||
"""Collect a single CSI frame."""
|
||||
if not router_interface.csi_streaming:
|
||||
return None
|
||||
|
||||
# Simulate CSI data
|
||||
return {
|
||||
"timestamp": datetime.utcnow().isoformat(),
|
||||
"router_id": router_interface.router_id,
|
||||
"amplitude": np.random.rand(64, 32).tolist(),
|
||||
"phase": np.random.rand(64, 32).tolist(),
|
||||
"frequency": 5.8e9,
|
||||
"bandwidth": 80e6,
|
||||
"antenna_count": 4,
|
||||
"subcarrier_count": 64,
|
||||
"signal_quality": np.random.uniform(0.7, 0.95)
|
||||
}
|
||||
|
||||
def get_collection_stats(self) -> Dict[str, Any]:
|
||||
"""Get collection statistics."""
|
||||
return {
|
||||
"total_frames": len(self.collected_data),
|
||||
"collection_rate": self.collection_rate,
|
||||
"is_collecting": self.is_collecting,
|
||||
"last_collection": self.collected_data[-1]["timestamp"] if self.collected_data else None
|
||||
}
|
||||
|
||||
return CSICollector()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_csi_collection_start_should_fail_initially(self, csi_collector):
|
||||
"""Test CSI collection start - should fail initially."""
|
||||
router_interfaces = [
|
||||
MockRouterInterface("router_001", "192.168.1.100"),
|
||||
MockRouterInterface("router_002", "192.168.1.101")
|
||||
]
|
||||
|
||||
result = await csi_collector.start_collection(router_interfaces)
|
||||
|
||||
# This will fail initially because the collector is designed to return False
|
||||
assert result is False
|
||||
assert csi_collector.is_collecting is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_single_frame_collection_should_fail_initially(self, csi_collector):
|
||||
"""Test single frame collection - should fail initially."""
|
||||
router = MockRouterInterface("router_001", "192.168.1.100")
|
||||
|
||||
# Without CSI streaming enabled
|
||||
frame = await csi_collector.collect_frame(router)
|
||||
|
||||
# This will fail initially
|
||||
assert frame is None
|
||||
|
||||
# Enable CSI streaming (manually for testing)
|
||||
router.csi_streaming = True
|
||||
frame = await csi_collector.collect_frame(router)
|
||||
|
||||
assert frame is not None
|
||||
assert "timestamp" in frame
|
||||
assert "router_id" in frame
|
||||
assert "amplitude" in frame
|
||||
assert "phase" in frame
|
||||
assert frame["router_id"] == "router_001"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_collection_statistics_should_fail_initially(self, csi_collector):
|
||||
"""Test collection statistics - should fail initially."""
|
||||
stats = csi_collector.get_collection_stats()
|
||||
|
||||
# This will fail initially
|
||||
assert isinstance(stats, dict)
|
||||
assert "total_frames" in stats
|
||||
assert "collection_rate" in stats
|
||||
assert "is_collecting" in stats
|
||||
assert "last_collection" in stats
|
||||
|
||||
assert stats["total_frames"] == 0
|
||||
assert stats["is_collecting"] is False
|
||||
assert stats["last_collection"] is None
|
||||
|
||||
|
||||
class TestHardwareErrorHandling:
|
||||
"""Test hardware error handling scenarios."""
|
||||
|
||||
@pytest.fixture
|
||||
def unreliable_router(self):
|
||||
"""Create unreliable router for error testing."""
|
||||
class UnreliableRouter(MockRouterInterface):
|
||||
def __init__(self, router_id: str, ip_address: str = "192.168.1.1"):
|
||||
super().__init__(router_id, ip_address)
|
||||
self.failure_rate = 0.3 # 30% failure rate
|
||||
self.connection_drops = 0
|
||||
|
||||
async def connect(self) -> bool:
|
||||
"""Unreliable connection."""
|
||||
if np.random.random() < self.failure_rate:
|
||||
return False
|
||||
return await super().connect()
|
||||
|
||||
async def send_heartbeat(self) -> bool:
|
||||
"""Unreliable heartbeat."""
|
||||
if np.random.random() < self.failure_rate:
|
||||
self.is_connected = False
|
||||
self.connection_drops += 1
|
||||
return False
|
||||
return await super().send_heartbeat()
|
||||
|
||||
async def start_csi_streaming(self, config: Dict[str, Any]) -> bool:
|
||||
"""Unreliable CSI streaming."""
|
||||
if np.random.random() < self.failure_rate:
|
||||
return False
|
||||
|
||||
# Still return False for initial test failure
|
||||
return False
|
||||
|
||||
return UnreliableRouter("unreliable_router", "192.168.1.200")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connection_retry_mechanism_should_fail_initially(self, unreliable_router):
|
||||
"""Test connection retry mechanism - should fail initially."""
|
||||
max_retries = 5
|
||||
success = False
|
||||
|
||||
for attempt in range(max_retries):
|
||||
result = await unreliable_router.connect()
|
||||
if result:
|
||||
success = True
|
||||
break
|
||||
|
||||
# Wait before retry
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# This will fail initially due to randomness, but should eventually pass
|
||||
# The test demonstrates the need for retry logic
|
||||
assert success or unreliable_router.connection_attempts >= max_retries
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connection_drop_detection_should_fail_initially(self, unreliable_router):
|
||||
"""Test connection drop detection - should fail initially."""
|
||||
# Establish connection
|
||||
await unreliable_router.connect()
|
||||
await unreliable_router.connect() # Ensure connection
|
||||
|
||||
initial_drops = unreliable_router.connection_drops
|
||||
|
||||
# Send multiple heartbeats to trigger potential drops
|
||||
for _ in range(10):
|
||||
await unreliable_router.send_heartbeat()
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
# This will fail initially
|
||||
# Should detect connection drops
|
||||
final_drops = unreliable_router.connection_drops
|
||||
assert final_drops >= initial_drops # May have detected drops
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_hardware_timeout_handling_should_fail_initially(self):
|
||||
"""Test hardware timeout handling - should fail initially."""
|
||||
async def slow_operation():
|
||||
"""Simulate slow hardware operation."""
|
||||
await asyncio.sleep(2.0) # 2 second delay
|
||||
return "success"
|
||||
|
||||
# Test with timeout
|
||||
try:
|
||||
result = await asyncio.wait_for(slow_operation(), timeout=1.0)
|
||||
# This should not be reached
|
||||
assert False, "Operation should have timed out"
|
||||
except asyncio.TimeoutError:
|
||||
# This will fail initially because we expect timeout handling
|
||||
assert True # Timeout was properly handled
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_network_error_simulation_should_fail_initially(self):
|
||||
"""Test network error simulation - should fail initially."""
|
||||
class NetworkErrorRouter(MockRouterInterface):
|
||||
async def connect(self) -> bool:
|
||||
"""Simulate network error."""
|
||||
raise ConnectionError("Network unreachable")
|
||||
|
||||
router = NetworkErrorRouter("error_router", "192.168.1.999")
|
||||
|
||||
# This will fail initially
|
||||
with pytest.raises(ConnectionError, match="Network unreachable"):
|
||||
await router.connect()
|
||||
|
||||
|
||||
class TestHardwareConfiguration:
|
||||
"""Test hardware configuration management."""
|
||||
|
||||
@pytest.fixture
|
||||
def config_manager(self):
|
||||
"""Create configuration manager."""
|
||||
class ConfigManager:
|
||||
def __init__(self):
|
||||
self.default_config = {
|
||||
"frequency": 5.8e9,
|
||||
"bandwidth": 80e6,
|
||||
"sample_rate": 1000,
|
||||
"antenna_config": "4x4_mimo",
|
||||
"power_level": 20,
|
||||
"channel": 36
|
||||
}
|
||||
self.router_configs = {}
|
||||
|
||||
def get_router_config(self, router_id: str) -> Dict[str, Any]:
|
||||
"""Get configuration for a specific router."""
|
||||
return self.router_configs.get(router_id, self.default_config.copy())
|
||||
|
||||
def set_router_config(self, router_id: str, config: Dict[str, Any]) -> bool:
|
||||
"""Set configuration for a specific router."""
|
||||
# Validate configuration
|
||||
required_fields = ["frequency", "bandwidth", "sample_rate"]
|
||||
if not all(field in config for field in required_fields):
|
||||
return False
|
||||
|
||||
self.router_configs[router_id] = config
|
||||
return True
|
||||
|
||||
def validate_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Validate router configuration."""
|
||||
errors = []
|
||||
|
||||
# Frequency validation
|
||||
if "frequency" in config:
|
||||
freq = config["frequency"]
|
||||
if not (2.4e9 <= freq <= 6e9):
|
||||
errors.append("Frequency must be between 2.4GHz and 6GHz")
|
||||
|
||||
# Bandwidth validation
|
||||
if "bandwidth" in config:
|
||||
bw = config["bandwidth"]
|
||||
if bw not in [20e6, 40e6, 80e6, 160e6]:
|
||||
errors.append("Bandwidth must be 20, 40, 80, or 160 MHz")
|
||||
|
||||
# Sample rate validation
|
||||
if "sample_rate" in config:
|
||||
sr = config["sample_rate"]
|
||||
if not (100 <= sr <= 10000):
|
||||
errors.append("Sample rate must be between 100 and 10000 Hz")
|
||||
|
||||
return {
|
||||
"valid": len(errors) == 0,
|
||||
"errors": errors
|
||||
}
|
||||
|
||||
return ConfigManager()
|
||||
|
||||
def test_default_configuration_should_fail_initially(self, config_manager):
|
||||
"""Test default configuration retrieval - should fail initially."""
|
||||
config = config_manager.get_router_config("new_router")
|
||||
|
||||
# This will fail initially
|
||||
assert isinstance(config, dict)
|
||||
assert "frequency" in config
|
||||
assert "bandwidth" in config
|
||||
assert "sample_rate" in config
|
||||
assert "antenna_config" in config
|
||||
assert config["frequency"] == 5.8e9
|
||||
assert config["bandwidth"] == 80e6
|
||||
|
||||
def test_configuration_validation_should_fail_initially(self, config_manager):
|
||||
"""Test configuration validation - should fail initially."""
|
||||
# Valid configuration
|
||||
valid_config = {
|
||||
"frequency": 5.8e9,
|
||||
"bandwidth": 80e6,
|
||||
"sample_rate": 1000
|
||||
}
|
||||
|
||||
result = config_manager.validate_config(valid_config)
|
||||
|
||||
# This will fail initially
|
||||
assert result["valid"] is True
|
||||
assert len(result["errors"]) == 0
|
||||
|
||||
# Invalid configuration
|
||||
invalid_config = {
|
||||
"frequency": 10e9, # Too high
|
||||
"bandwidth": 100e6, # Invalid
|
||||
"sample_rate": 50 # Too low
|
||||
}
|
||||
|
||||
result = config_manager.validate_config(invalid_config)
|
||||
assert result["valid"] is False
|
||||
assert len(result["errors"]) == 3
|
||||
|
||||
def test_router_specific_configuration_should_fail_initially(self, config_manager):
|
||||
"""Test router-specific configuration - should fail initially."""
|
||||
router_id = "router_001"
|
||||
custom_config = {
|
||||
"frequency": 2.4e9,
|
||||
"bandwidth": 40e6,
|
||||
"sample_rate": 500,
|
||||
"antenna_config": "2x2_mimo"
|
||||
}
|
||||
|
||||
# Set custom configuration
|
||||
result = config_manager.set_router_config(router_id, custom_config)
|
||||
|
||||
# This will fail initially
|
||||
assert result is True
|
||||
|
||||
# Retrieve custom configuration
|
||||
retrieved_config = config_manager.get_router_config(router_id)
|
||||
assert retrieved_config["frequency"] == 2.4e9
|
||||
assert retrieved_config["bandwidth"] == 40e6
|
||||
assert retrieved_config["antenna_config"] == "2x2_mimo"
|
||||
|
||||
# Test invalid configuration
|
||||
invalid_config = {"frequency": 5.8e9} # Missing required fields
|
||||
result = config_manager.set_router_config(router_id, invalid_config)
|
||||
assert result is False
|
||||
Reference in New Issue
Block a user