calejocontrol
/
CalejoControl
1
0
Fork 0
Code Issues Pull Requests 4 Packages Projects Releases Wiki Activity
CalejoControl/tests/integration/test_dashboard_integration.py

253 lines
9.3 KiB
Python
Raw Normal View History

Add interactive dashboard with comprehensive testing - Implemented web-based dashboard with tab-based interface - Added configuration management with real-time validation - Created system status monitoring and log viewing - Implemented system actions (restart, backup, health checks) - Added comprehensive test suite with 35 tests (100% passing) - Integrated dashboard with existing REST API - Added responsive design for mobile and desktop - Implemented security validation and warnings - Created comprehensive documentation Features: - Status monitoring with color-coded indicators - Configuration management with web-based editor - Real-time log viewing with filtering - One-click system operations - Mobile-responsive design - Security validation for default credentials Testing: - 13 model tests for Pydantic data structures - 8 validation tests for configuration logic - 8 API endpoint tests for all dashboard routes - 6 integration tests with REST API - All 35 tests passing (100% success rate) Access: http://localhost:8080/dashboard
2025-10-30 07:22:00 +00:00
"""
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