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

256 lines
8.7 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
"""
Tests for Dashboard 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,
validate_configuration,
ValidationResult
)
class TestDashboardAPIEndpoints:
"""Test dashboard API endpoints"""
@pytest.fixture
def client(self):
"""Create test client with dashboard router"""
app = FastAPI()
app.include_router(dashboard_router)
return TestClient(app)
@patch('src.dashboard.api.settings')
def test_get_configuration(self, mock_settings, client):
"""Test GET /api/v1/dashboard/config endpoint"""
# Mock settings
mock_settings.db_host = "test_host"
mock_settings.db_port = 5432
mock_settings.db_name = "test_db"
mock_settings.db_user = "test_user"
mock_settings.opcua_enabled = True
mock_settings.opcua_host = "localhost"
mock_settings.opcua_port = 4840
mock_settings.modbus_enabled = True
mock_settings.modbus_host = "localhost"
mock_settings.modbus_port = 502
mock_settings.modbus_unit_id = 1
mock_settings.rest_api_enabled = True
mock_settings.rest_api_host = "0.0.0.0"
mock_settings.rest_api_port = 8080
mock_settings.rest_api_cors_enabled = True
mock_settings.health_monitor_port = 9090
response = client.get("/api/v1/dashboard/config")
assert response.status_code == 200
data = response.json()
# Check database configuration
assert data["database"]["db_host"] == "test_host"
assert data["database"]["db_port"] == 5432
assert data["database"]["db_name"] == "test_db"
assert data["database"]["db_user"] == "test_user"
assert data["database"]["db_password"] == "********" # Masked
# Check protocol configuration
assert data["opcua"]["enabled"] == True
assert data["opcua"]["port"] == 4840
assert data["modbus"]["enabled"] == True
assert data["modbus"]["port"] == 502
# Check REST API configuration
assert data["rest_api"]["port"] == 8080
assert data["rest_api"]["cors_enabled"] == True
# Check monitoring configuration
assert data["monitoring"]["health_monitor_port"] == 9090
# Check security configuration (masked)
assert data["security"]["jwt_secret_key"] == "********"
assert data["security"]["api_key"] == "********"
@patch('src.dashboard.api.validate_configuration')
@patch('src.dashboard.api.save_configuration')
def test_update_configuration_valid(self, mock_save, mock_validate, client):
"""Test POST /api/v1/dashboard/config with valid configuration"""
# Mock validation to return valid result
mock_validate.return_value = ValidationResult(
valid=True,
errors=[],
warnings=["Test warning"]
)
config_data = {
"database": {
"db_host": "localhost",
"db_port": 5432,
"db_name": "calejo",
"db_user": "calejo",
"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"
}
}
response = client.post("/api/v1/dashboard/config", json=config_data)
assert response.status_code == 200
data = response.json()
assert data["valid"] == True
assert data["warnings"] == ["Test warning"]
assert len(data["errors"]) == 0
# Verify save_configuration was called
mock_save.assert_called_once()
@patch('src.dashboard.api.validate_configuration')
def test_update_configuration_invalid(self, mock_validate, client):
"""Test POST /api/v1/dashboard/config with invalid configuration"""
# Mock validation to return invalid result
mock_validate.return_value = ValidationResult(
valid=False,
errors=["Database host is required", "Invalid port"],
warnings=[]
)
config_data = {
"database": {
"db_host": "", # Invalid
"db_port": 5432,
"db_name": "calejo",
"db_user": "calejo",
"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"
}
}
response = client.post("/api/v1/dashboard/config", json=config_data)
assert response.status_code == 200
data = response.json()
assert data["valid"] == False
assert data["errors"] == ["Database host is required", "Invalid port"]
assert len(data["warnings"]) == 0
def test_get_system_status(self, client):
"""Test GET /api/v1/dashboard/status endpoint"""
response = client.get("/api/v1/dashboard/status")
assert response.status_code == 200
data = response.json()
# Check all status fields are present
assert "application_status" in data
assert "database_status" in data
assert "opcua_status" in data
assert "modbus_status" in data
assert "rest_api_status" in data
assert "monitoring_status" in data
# Check status values are strings
assert isinstance(data["application_status"], str)
assert isinstance(data["database_status"], str)
assert isinstance(data["opcua_status"], str)
assert isinstance(data["modbus_status"], str)
assert isinstance(data["rest_api_status"], str)
assert isinstance(data["monitoring_status"], str)
def test_restart_system(self, client):
"""Test POST /api/v1/dashboard/restart endpoint"""
response = client.post("/api/v1/dashboard/restart")
assert response.status_code == 200
data = response.json()
assert "message" in data
assert "status" in data
assert data["status"] == "pending"
def test_create_backup(self, client):
"""Test GET /api/v1/dashboard/backup endpoint"""
response = client.get("/api/v1/dashboard/backup")
assert response.status_code == 200
data = response.json()
assert "message" in data
assert "status" in data
assert data["status"] == "in_progress"
def test_get_system_logs(self, client):
"""Test GET /api/v1/dashboard/logs endpoint"""
response = client.get("/api/v1/dashboard/logs?limit=10")
assert response.status_code == 200
data = response.json()
assert "logs" in data
assert isinstance(data["logs"], list)
# Check log entry structure
if len(data["logs"]) > 0:
log_entry = data["logs"][0]
assert "timestamp" in log_entry
assert "level" in log_entry
assert "message" in log_entry
def test_get_system_logs_with_limit(self, client):
"""Test GET /api/v1/dashboard/logs with limit parameter"""
response = client.get("/api/v1/dashboard/logs?limit=5")
assert response.status_code == 200
data = response.json()
assert "logs" in data
assert isinstance(data["logs"], list)
# Should return limited number of logs
assert len(data["logs"]) <= 5