188 lines
7.4 KiB
Python
188 lines
7.4 KiB
Python
"""
|
|
Tests for Protocol Mapping API endpoints
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch, AsyncMock
|
|
from fastapi.testclient import TestClient
|
|
from fastapi import FastAPI
|
|
|
|
from src.dashboard.api import dashboard_router
|
|
from src.dashboard.configuration_manager import ProtocolMapping, ProtocolType
|
|
|
|
|
|
class TestProtocolMappingAPIEndpoints:
|
|
"""Test protocol mapping API endpoints"""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
"""Create test client with dashboard router"""
|
|
app = FastAPI()
|
|
app.include_router(dashboard_router)
|
|
return TestClient(app)
|
|
|
|
def test_get_available_protocols(self, client):
|
|
"""Test GET /api/v1/dashboard/protocol-mappings/protocols endpoint"""
|
|
response = client.get("/api/v1/dashboard/protocol-mappings/protocols")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] == True
|
|
assert "protocols" in data
|
|
assert len(data["protocols"]) > 0
|
|
|
|
# Check that expected protocols are present
|
|
protocol_values = [p["value"] for p in data["protocols"]]
|
|
assert "modbus_tcp" in protocol_values
|
|
assert "opcua" in protocol_values
|
|
|
|
@patch('src.dashboard.api.configuration_manager')
|
|
def test_get_protocol_mappings(self, mock_config_manager, client):
|
|
"""Test GET /api/v1/dashboard/protocol-mappings endpoint"""
|
|
# Mock configuration manager
|
|
mock_mapping = ProtocolMapping(
|
|
id="test_mapping_001",
|
|
protocol_type=ProtocolType.MODBUS_TCP,
|
|
station_id="station_001",
|
|
pump_id="pump_001",
|
|
data_type="setpoint",
|
|
protocol_address="40001",
|
|
db_source="pump_data.setpoint"
|
|
)
|
|
mock_config_manager.get_protocol_mappings.return_value = [mock_mapping]
|
|
|
|
response = client.get("/api/v1/dashboard/protocol-mappings")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] == True
|
|
assert data["count"] == 1
|
|
assert len(data["mappings"]) == 1
|
|
assert data["mappings"][0]["id"] == "test_mapping_001"
|
|
|
|
@patch('src.dashboard.api.configuration_manager')
|
|
def test_get_protocol_mappings_with_filter(self, mock_config_manager, client):
|
|
"""Test GET /api/v1/dashboard/protocol-mappings with filtering"""
|
|
# Mock configuration manager
|
|
mock_mapping = ProtocolMapping(
|
|
id="test_mapping_001",
|
|
protocol_type=ProtocolType.MODBUS_TCP,
|
|
station_id="station_001",
|
|
pump_id="pump_001",
|
|
data_type="setpoint",
|
|
protocol_address="40001",
|
|
db_source="pump_data.setpoint"
|
|
)
|
|
mock_config_manager.get_protocol_mappings.return_value = [mock_mapping]
|
|
|
|
response = client.get("/api/v1/dashboard/protocol-mappings?protocol_type=modbus_tcp")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] == True
|
|
assert data["count"] == 1
|
|
|
|
@patch('src.dashboard.api.configuration_manager')
|
|
def test_create_protocol_mapping(self, mock_config_manager, client):
|
|
"""Test POST /api/v1/dashboard/protocol-mappings endpoint"""
|
|
# Mock configuration manager
|
|
mock_config_manager.add_protocol_mapping.return_value = True
|
|
|
|
mapping_data = {
|
|
"id": "test_mapping_001",
|
|
"protocol_type": "modbus_tcp",
|
|
"station_id": "station_001",
|
|
"pump_id": "pump_001",
|
|
"data_type": "setpoint",
|
|
"protocol_address": "40001",
|
|
"db_source": "pump_data.setpoint"
|
|
}
|
|
|
|
response = client.post("/api/v1/dashboard/protocol-mappings", json=mapping_data)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] == True
|
|
assert data["message"] == "Protocol mapping created successfully"
|
|
assert "mapping" in data
|
|
|
|
@patch('src.dashboard.api.configuration_manager')
|
|
def test_create_protocol_mapping_invalid_protocol(self, mock_config_manager, client):
|
|
"""Test POST /api/v1/dashboard/protocol-mappings with invalid protocol"""
|
|
mapping_data = {
|
|
"id": "test_mapping_001",
|
|
"protocol_type": "invalid_protocol",
|
|
"station_id": "station_001",
|
|
"pump_id": "pump_001",
|
|
"data_type": "setpoint",
|
|
"protocol_address": "40001",
|
|
"db_source": "pump_data.setpoint"
|
|
}
|
|
|
|
response = client.post("/api/v1/dashboard/protocol-mappings", json=mapping_data)
|
|
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert "Invalid protocol type" in data["detail"]
|
|
|
|
@patch('src.dashboard.api.configuration_manager')
|
|
def test_update_protocol_mapping(self, mock_config_manager, client):
|
|
"""Test PUT /api/v1/dashboard/protocol-mappings/{mapping_id} endpoint"""
|
|
# Mock configuration manager
|
|
mock_config_manager.update_protocol_mapping.return_value = True
|
|
|
|
mapping_data = {
|
|
"protocol_type": "modbus_tcp",
|
|
"station_id": "station_001",
|
|
"pump_id": "pump_001",
|
|
"data_type": "setpoint",
|
|
"protocol_address": "40002", # Updated address
|
|
"db_source": "pump_data.setpoint"
|
|
}
|
|
|
|
response = client.put("/api/v1/dashboard/protocol-mappings/test_mapping_001", json=mapping_data)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] == True
|
|
assert data["message"] == "Protocol mapping updated successfully"
|
|
|
|
@patch('src.dashboard.api.configuration_manager')
|
|
def test_delete_protocol_mapping(self, mock_config_manager, client):
|
|
"""Test DELETE /api/v1/dashboard/protocol-mappings/{mapping_id} endpoint"""
|
|
# Mock configuration manager
|
|
mock_config_manager.delete_protocol_mapping.return_value = True
|
|
|
|
response = client.delete("/api/v1/dashboard/protocol-mappings/test_mapping_001")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] == True
|
|
assert data["message"] == "Protocol mapping test_mapping_001 deleted successfully"
|
|
|
|
@patch('src.dashboard.api.configuration_manager')
|
|
def test_validate_protocol_mapping(self, mock_config_manager, client):
|
|
"""Test POST /api/v1/dashboard/protocol-mappings/{mapping_id}/validate endpoint"""
|
|
# Mock configuration manager
|
|
mock_config_manager.validate_protocol_mapping.return_value = {
|
|
"valid": True,
|
|
"errors": [],
|
|
"warnings": ["Database source should be in format 'table.column'"]
|
|
}
|
|
|
|
mapping_data = {
|
|
"protocol_type": "modbus_tcp",
|
|
"station_id": "station_001",
|
|
"pump_id": "pump_001",
|
|
"data_type": "setpoint",
|
|
"protocol_address": "40001",
|
|
"db_source": "pump_data" # Missing .column
|
|
}
|
|
|
|
response = client.post("/api/v1/dashboard/protocol-mappings/test_mapping_001/validate", json=mapping_data)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] == True
|
|
assert data["valid"] == True
|
|
assert len(data["warnings"]) > 0 |