253 lines
9.3 KiB
Python
253 lines
9.3 KiB
Python
"""
|
|
Integration tests for Dashboard with REST API
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.protocols.rest_api import RESTAPIServer
|
|
from src.core.setpoint_manager import SetpointManager
|
|
from src.core.emergency_stop import EmergencyStopManager
|
|
from src.core.security import SecurityManager
|
|
from src.monitoring.health_monitor import HealthMonitor
|
|
|
|
|
|
class TestDashboardIntegration:
|
|
"""Test dashboard integration with REST API"""
|
|
|
|
@pytest.fixture
|
|
def mock_managers(self):
|
|
"""Create mock managers for testing"""
|
|
setpoint_manager = Mock(spec=SetpointManager)
|
|
emergency_stop_manager = Mock(spec=EmergencyStopManager)
|
|
security_manager = Mock(spec=SecurityManager)
|
|
health_monitor = Mock(spec=HealthMonitor)
|
|
|
|
return {
|
|
'setpoint_manager': setpoint_manager,
|
|
'emergency_stop_manager': emergency_stop_manager,
|
|
'security_manager': security_manager,
|
|
'health_monitor': health_monitor
|
|
}
|
|
|
|
@pytest.fixture
|
|
def api_server(self, mock_managers):
|
|
"""Create REST API server with dashboard integration"""
|
|
server = RESTAPIServer(
|
|
setpoint_manager=mock_managers['setpoint_manager'],
|
|
emergency_stop_manager=mock_managers['emergency_stop_manager'],
|
|
health_monitor=mock_managers['health_monitor']
|
|
)
|
|
return server
|
|
|
|
@pytest.fixture
|
|
def client(self, api_server):
|
|
"""Create test client for REST API"""
|
|
return TestClient(api_server.app)
|
|
|
|
def test_dashboard_routes_available(self, client):
|
|
"""Test that dashboard routes are available through REST API"""
|
|
# Test dashboard root route
|
|
response = client.get("/dashboard")
|
|
assert response.status_code == 200
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
# Test dashboard API routes
|
|
response = client.get("/api/v1/dashboard/config")
|
|
assert response.status_code == 200
|
|
|
|
response = client.get("/api/v1/dashboard/status")
|
|
assert response.status_code == 200
|
|
|
|
response = client.get("/api/v1/dashboard/logs")
|
|
assert response.status_code == 200
|
|
|
|
def test_static_files_served(self, client):
|
|
"""Test that static files (JavaScript) are served"""
|
|
response = client.get("/static/dashboard.js")
|
|
|
|
# Should either return the file or 404 if not found
|
|
# But the route should be available
|
|
assert response.status_code in [200, 404]
|
|
|
|
# If file exists, check content type (both text/javascript and application/javascript are valid)
|
|
if response.status_code == 200:
|
|
content_type = response.headers.get("content-type", "")
|
|
assert "javascript" in content_type or "text/javascript" in content_type
|
|
|
|
def test_dashboard_configuration_flow(self, client):
|
|
"""Test complete configuration flow through dashboard API"""
|
|
# 1. Get current configuration
|
|
response = client.get("/api/v1/dashboard/config")
|
|
assert response.status_code == 200
|
|
current_config = response.json()
|
|
|
|
# Verify structure
|
|
assert "database" in current_config
|
|
assert "opcua" in current_config
|
|
assert "modbus" in current_config
|
|
assert "rest_api" in current_config
|
|
assert "monitoring" in current_config
|
|
assert "security" in current_config
|
|
|
|
# 2. Update configuration (mock validation)
|
|
with patch('src.dashboard.api.validate_configuration') as mock_validate:
|
|
mock_validate.return_value = Mock(valid=True, errors=[], warnings=[])
|
|
|
|
update_data = {
|
|
"database": {
|
|
"db_host": "new_host",
|
|
"db_port": 5433,
|
|
"db_name": "new_db",
|
|
"db_user": "new_user",
|
|
"db_password": "new_password"
|
|
},
|
|
"opcua": {
|
|
"enabled": True,
|
|
"host": "localhost",
|
|
"port": 4841
|
|
},
|
|
"modbus": {
|
|
"enabled": True,
|
|
"host": "localhost",
|
|
"port": 503,
|
|
"unit_id": 2
|
|
},
|
|
"rest_api": {
|
|
"enabled": True,
|
|
"host": "127.0.0.1",
|
|
"port": 8081,
|
|
"cors_enabled": False
|
|
},
|
|
"monitoring": {
|
|
"health_monitor_port": 9091,
|
|
"metrics_enabled": True
|
|
},
|
|
"security": {
|
|
"jwt_secret_key": "new_secret",
|
|
"api_key": "new_api_key"
|
|
}
|
|
}
|
|
|
|
response = client.post("/api/v1/dashboard/config", json=update_data)
|
|
assert response.status_code == 200
|
|
update_result = response.json()
|
|
assert update_result["valid"] == True
|
|
|
|
def test_dashboard_system_actions(self, client):
|
|
"""Test system actions through dashboard API"""
|
|
# Test restart
|
|
response = client.post("/api/v1/dashboard/restart")
|
|
assert response.status_code == 200
|
|
restart_result = response.json()
|
|
assert "message" in restart_result
|
|
assert "status" in restart_result
|
|
|
|
# Test backup
|
|
response = client.get("/api/v1/dashboard/backup")
|
|
assert response.status_code == 200
|
|
backup_result = response.json()
|
|
assert "message" in backup_result
|
|
assert "status" in backup_result
|
|
|
|
def test_dashboard_status_integration(self, client, mock_managers):
|
|
"""Test dashboard status integration with health monitor"""
|
|
# Mock health monitor to return specific status
|
|
mock_health_status = {
|
|
"status": "healthy",
|
|
"timestamp": "2024-01-01T10:00:00",
|
|
"components": {
|
|
"database": {
|
|
"status": "connected",
|
|
"message": "Database connection successful",
|
|
"last_check": "2024-01-01T10:00:00",
|
|
"response_time_ms": 5.2
|
|
},
|
|
"opcua_server": {
|
|
"status": "listening",
|
|
"message": "OPC UA server running",
|
|
"last_check": "2024-01-01T10:00:00",
|
|
"response_time_ms": 1.5
|
|
}
|
|
}
|
|
}
|
|
|
|
mock_managers['health_monitor'].get_health_status.return_value = {
|
|
"status": "healthy",
|
|
"timestamp": "2024-01-01T10:00:00"
|
|
}
|
|
|
|
# Test health endpoint
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
health_data = response.json()
|
|
assert "status" in health_data
|
|
|
|
# Test dashboard status endpoint
|
|
response = client.get("/api/v1/dashboard/status")
|
|
assert response.status_code == 200
|
|
status_data = response.json()
|
|
assert "application_status" in status_data
|
|
assert "database_status" in status_data
|
|
assert "opcua_status" in status_data
|
|
assert "modbus_status" in status_data
|
|
assert "rest_api_status" in status_data
|
|
assert "monitoring_status" in status_data
|
|
|
|
def test_dashboard_error_handling(self, client):
|
|
"""Test dashboard error handling"""
|
|
# Test with invalid configuration data
|
|
invalid_config = {
|
|
"database": {
|
|
"db_host": "", # Invalid - empty
|
|
"db_port": 70000, # Invalid port
|
|
"db_name": "", # Invalid - empty
|
|
"db_user": "", # Invalid - empty
|
|
"db_password": "password"
|
|
},
|
|
"opcua": {
|
|
"enabled": True,
|
|
"host": "localhost",
|
|
"port": 4840
|
|
},
|
|
"modbus": {
|
|
"enabled": True,
|
|
"host": "localhost",
|
|
"port": 502,
|
|
"unit_id": 1
|
|
},
|
|
"rest_api": {
|
|
"enabled": True,
|
|
"host": "0.0.0.0",
|
|
"port": 8080,
|
|
"cors_enabled": True
|
|
},
|
|
"monitoring": {
|
|
"health_monitor_port": 9090,
|
|
"metrics_enabled": True
|
|
},
|
|
"security": {
|
|
"jwt_secret_key": "secret",
|
|
"api_key": "api_key"
|
|
}
|
|
}
|
|
|
|
# Mock validation to return errors
|
|
with patch('src.dashboard.api.validate_configuration') as mock_validate:
|
|
mock_validate.return_value = Mock(
|
|
valid=False,
|
|
errors=[
|
|
"Database host is required",
|
|
"Database name is required",
|
|
"Database user is required",
|
|
"Database port must be between 1 and 65535"
|
|
],
|
|
warnings=[]
|
|
)
|
|
|
|
response = client.post("/api/v1/dashboard/config", json=invalid_config)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["valid"] == False
|
|
assert len(result["errors"]) == 4 |