245 lines
9.3 KiB
Python
245 lines
9.3 KiB
Python
|
"""
|
||
|
|
Integration tests for Protocol Server integration with ConfigurationManager
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import asyncio
|
||
|
|
from unittest.mock import Mock, AsyncMock
|
||
|
|
|
||
|
|
from src.dashboard.configuration_manager import (
|
||
|
|
ConfigurationManager,
|
||
|
|
ProtocolMapping,
|
||
|
|
ProtocolType
|
||
|
|
)
|
||
|
|
from src.database.flexible_client import FlexibleDatabaseClient
|
||
|
|
|
||
|
|
|
||
|
|
class TestProtocolServerIntegration:
|
||
|
|
"""Test integration between ConfigurationManager and Protocol Servers"""
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def sqlite_db_client(self):
|
||
|
|
"""Create SQLite database client for testing"""
|
||
|
|
db_client = FlexibleDatabaseClient('sqlite:///test_protocol_integration.db')
|
||
|
|
await db_client.connect()
|
||
|
|
db_client.create_tables()
|
||
|
|
yield db_client
|
||
|
|
await db_client.disconnect()
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def mock_setpoint_manager(self):
|
||
|
|
"""Create mock setpoint manager"""
|
||
|
|
mock = Mock()
|
||
|
|
mock.get_current_setpoint = Mock(return_value=50.0)
|
||
|
|
return mock
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def mock_security_manager(self):
|
||
|
|
"""Create mock security manager"""
|
||
|
|
mock = Mock()
|
||
|
|
mock.validate_access = Mock(return_value=True)
|
||
|
|
return mock
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def mock_audit_logger(self):
|
||
|
|
"""Create mock audit logger"""
|
||
|
|
mock = Mock()
|
||
|
|
mock.log_event = Mock()
|
||
|
|
return mock
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_modbus_server_integration(self, sqlite_db_client, mock_setpoint_manager, mock_security_manager, mock_audit_logger):
|
||
|
|
"""Test Modbus server integration with ConfigurationManager"""
|
||
|
|
# Import here to avoid circular imports
|
||
|
|
from src.protocols.modbus_server import ModbusServer
|
||
|
|
|
||
|
|
# Create ConfigurationManager with database
|
||
|
|
config_manager = ConfigurationManager(db_client=sqlite_db_client)
|
||
|
|
|
||
|
|
# Add Modbus mappings
|
||
|
|
modbus_mappings = [
|
||
|
|
ProtocolMapping(
|
||
|
|
id="modbus_setpoint_001",
|
||
|
|
station_id="station_001",
|
||
|
|
pump_id="pump_001",
|
||
|
|
protocol_type=ProtocolType.MODBUS_TCP,
|
||
|
|
protocol_address="100",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
),
|
||
|
|
ProtocolMapping(
|
||
|
|
id="modbus_setpoint_002",
|
||
|
|
station_id="station_001",
|
||
|
|
pump_id="pump_002",
|
||
|
|
protocol_type=ProtocolType.MODBUS_TCP,
|
||
|
|
protocol_address="101",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
)
|
||
|
|
]
|
||
|
|
|
||
|
|
for mapping in modbus_mappings:
|
||
|
|
config_manager.add_protocol_mapping(mapping)
|
||
|
|
|
||
|
|
# Create Modbus server with ConfigurationManager
|
||
|
|
modbus_server = ModbusServer(
|
||
|
|
configuration_manager=config_manager,
|
||
|
|
setpoint_manager=mock_setpoint_manager,
|
||
|
|
security_manager=mock_security_manager,
|
||
|
|
audit_logger=mock_audit_logger
|
||
|
|
)
|
||
|
|
|
||
|
|
# Initialize pump mapping
|
||
|
|
await modbus_server._initialize_pump_mapping()
|
||
|
|
|
||
|
|
# Verify that mappings were loaded
|
||
|
|
assert len(modbus_server.pump_addresses) == 2
|
||
|
|
assert "station_001/pump_001" in modbus_server.pump_addresses
|
||
|
|
assert "station_001/pump_002" in modbus_server.pump_addresses
|
||
|
|
assert modbus_server.pump_addresses["station_001/pump_001"] == 100
|
||
|
|
assert modbus_server.pump_addresses["station_001/pump_002"] == 101
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_opcua_server_integration(self, sqlite_db_client, mock_setpoint_manager, mock_security_manager, mock_audit_logger):
|
||
|
|
"""Test OPC UA server integration with ConfigurationManager"""
|
||
|
|
# Import here to avoid circular imports
|
||
|
|
from src.protocols.opcua_server import OPCUAServer
|
||
|
|
|
||
|
|
# Create ConfigurationManager with database
|
||
|
|
config_manager = ConfigurationManager(db_client=sqlite_db_client)
|
||
|
|
|
||
|
|
# Add OPC UA mappings
|
||
|
|
opcua_mappings = [
|
||
|
|
ProtocolMapping(
|
||
|
|
id="opcua_setpoint_001",
|
||
|
|
station_id="station_001",
|
||
|
|
pump_id="pump_001",
|
||
|
|
protocol_type=ProtocolType.OPC_UA,
|
||
|
|
protocol_address="ns=2;s=Station_001.Pump_001.Setpoint_Hz",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
),
|
||
|
|
ProtocolMapping(
|
||
|
|
id="opcua_setpoint_002",
|
||
|
|
station_id="station_001",
|
||
|
|
pump_id="pump_002",
|
||
|
|
protocol_type=ProtocolType.OPC_UA,
|
||
|
|
protocol_address="ns=2;s=Station_001.Pump_002.Setpoint_Hz",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
)
|
||
|
|
]
|
||
|
|
|
||
|
|
for mapping in opcua_mappings:
|
||
|
|
config_manager.add_protocol_mapping(mapping)
|
||
|
|
|
||
|
|
# Create OPC UA server with ConfigurationManager
|
||
|
|
opcua_server = OPCUAServer(
|
||
|
|
configuration_manager=config_manager,
|
||
|
|
setpoint_manager=mock_setpoint_manager,
|
||
|
|
security_manager=mock_security_manager,
|
||
|
|
audit_logger=mock_audit_logger
|
||
|
|
)
|
||
|
|
|
||
|
|
# Verify that mappings were loaded
|
||
|
|
opcua_mappings = opcua_server.configuration_manager.get_protocol_mappings(protocol_type=ProtocolType.OPC_UA)
|
||
|
|
assert len(opcua_mappings) == 2
|
||
|
|
|
||
|
|
# Verify specific mappings
|
||
|
|
mapping_ids = [m.id for m in opcua_mappings]
|
||
|
|
assert "opcua_setpoint_001" in mapping_ids
|
||
|
|
assert "opcua_setpoint_002" in mapping_ids
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_mixed_protocol_integration(self, sqlite_db_client, mock_setpoint_manager, mock_security_manager, mock_audit_logger):
|
||
|
|
"""Test integration with mixed protocol types"""
|
||
|
|
# Import here to avoid circular imports
|
||
|
|
from src.protocols.modbus_server import ModbusServer
|
||
|
|
from src.protocols.opcua_server import OPCUAServer
|
||
|
|
|
||
|
|
# Create ConfigurationManager with database
|
||
|
|
config_manager = ConfigurationManager(db_client=sqlite_db_client)
|
||
|
|
|
||
|
|
# Add mixed protocol mappings
|
||
|
|
mixed_mappings = [
|
||
|
|
ProtocolMapping(
|
||
|
|
id="modbus_mixed_001",
|
||
|
|
station_id="station_001",
|
||
|
|
pump_id="pump_001",
|
||
|
|
protocol_type=ProtocolType.MODBUS_TCP,
|
||
|
|
protocol_address="100",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
),
|
||
|
|
ProtocolMapping(
|
||
|
|
id="opcua_mixed_001",
|
||
|
|
station_id="station_001",
|
||
|
|
pump_id="pump_002",
|
||
|
|
protocol_type=ProtocolType.OPC_UA,
|
||
|
|
protocol_address="ns=2;s=Station_001.Pump_002.Setpoint_Hz",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
),
|
||
|
|
ProtocolMapping(
|
||
|
|
id="modbus_rtu_mixed_001",
|
||
|
|
station_id="station_002",
|
||
|
|
pump_id="pump_001",
|
||
|
|
protocol_type=ProtocolType.MODBUS_RTU,
|
||
|
|
protocol_address="40001",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
),
|
||
|
|
ProtocolMapping(
|
||
|
|
id="rest_mixed_001",
|
||
|
|
station_id="station_002",
|
||
|
|
pump_id="pump_002",
|
||
|
|
protocol_type=ProtocolType.REST_API,
|
||
|
|
protocol_address="https://api.example.com/v1/stations/002/pumps/002/setpoint",
|
||
|
|
data_type="setpoint",
|
||
|
|
db_source="frequency_hz"
|
||
|
|
)
|
||
|
|
]
|
||
|
|
|
||
|
|
for mapping in mixed_mappings:
|
||
|
|
config_manager.add_protocol_mapping(mapping)
|
||
|
|
|
||
|
|
# Create Modbus server
|
||
|
|
modbus_server = ModbusServer(
|
||
|
|
configuration_manager=config_manager,
|
||
|
|
setpoint_manager=mock_setpoint_manager,
|
||
|
|
security_manager=mock_security_manager,
|
||
|
|
audit_logger=mock_audit_logger
|
||
|
|
)
|
||
|
|
|
||
|
|
# Create OPC UA server
|
||
|
|
opcua_server = OPCUAServer(
|
||
|
|
configuration_manager=config_manager,
|
||
|
|
setpoint_manager=mock_setpoint_manager,
|
||
|
|
security_manager=mock_security_manager,
|
||
|
|
audit_logger=mock_audit_logger
|
||
|
|
)
|
||
|
|
|
||
|
|
# Initialize Modbus pump mapping
|
||
|
|
await modbus_server._initialize_pump_mapping()
|
||
|
|
|
||
|
|
# Verify Modbus mappings
|
||
|
|
assert len(modbus_server.pump_addresses) == 2 # Only Modbus TCP and RTU
|
||
|
|
|
||
|
|
# Verify OPC UA mappings
|
||
|
|
opcua_mappings = opcua_server.configuration_manager.get_protocol_mappings(protocol_type=ProtocolType.OPC_UA)
|
||
|
|
assert len(opcua_mappings) == 1
|
||
|
|
|
||
|
|
# Verify all mappings in ConfigurationManager
|
||
|
|
all_mappings = config_manager.get_protocol_mappings()
|
||
|
|
assert len(all_mappings) == 4
|
||
|
|
|
||
|
|
# Verify protocol type distribution
|
||
|
|
modbus_tcp_mappings = config_manager.get_protocol_mappings(protocol_type=ProtocolType.MODBUS_TCP)
|
||
|
|
modbus_rtu_mappings = config_manager.get_protocol_mappings(protocol_type=ProtocolType.MODBUS_RTU)
|
||
|
|
opcua_mappings = config_manager.get_protocol_mappings(protocol_type=ProtocolType.OPC_UA)
|
||
|
|
rest_mappings = config_manager.get_protocol_mappings(protocol_type=ProtocolType.REST_API)
|
||
|
|
|
||
|
|
assert len(modbus_tcp_mappings) == 1
|
||
|
|
assert len(modbus_rtu_mappings) == 1
|
||
|
|
assert len(opcua_mappings) == 1
|
||
|
|
assert len(rest_mappings) == 1
|