""" 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