diff --git a/controller/conftest.py b/controller/conftest.py
index da65f586..8e304885 100644
--- a/controller/conftest.py
+++ b/controller/conftest.py
@@ -1,6 +1,7 @@
"""
Pytest fixtures for Controller tests.
"""
+import asyncio
import copy
import tempfile
from pathlib import Path
@@ -13,6 +14,50 @@ from fastapi.testclient import TestClient
from controller.defaults import DEFAULT_CONFIG
+_FAKE_CACHED_STATUS = {
+ "cpu_stats": {
+ "percent": 0.0, "cores": 1,
+ "cur_frequency": 0, "max_frequency": 0, "min_frequency": 0,
+ "temperature": None, "high_temperature": None, "critical_temperature": None,
+ },
+ "memory_stats": {"total": 1, "used": 0, "free": 1, "cached": 0},
+ "load_stats": {"minute_1": "0.00", "minute_5": "0.00", "minute_15": "0.00"},
+ "drives_stats": [],
+ "nic_bandwidth_stats": {},
+ "power_stats": {"under_voltage": False, "over_current": False},
+ "iostat_stats": {"percent_iowait": None},
+ "processes_stats": [],
+ "disk_bandwidth_stats": {},
+ "uptime_stats": {"uptime_seconds": 0},
+ "last_status": "1970-01-01T00:00:00",
+}
+
+
+async def _noop_status_worker(app, base_sleep: float = 5.0):
+ """Replacement for status_worker_loop during tests.
+
+ The real worker calls blocking samplers (psutil with interval=1.0, iostat)
+ via asyncio.to_thread; those threads can't be cancelled, so TestClient
+ teardown waits ~2s for each test. Also pre-populates cached_status so the
+ dashboard route doesn't fall back to direct sampling on the request path.
+ """
+ app.state.cached_status = dict(_FAKE_CACHED_STATUS)
+ try:
+ await asyncio.Event().wait()
+ except asyncio.CancelledError:
+ raise
+
+
+@pytest.fixture(autouse=True)
+def _patch_status_worker():
+ """Replace the real status worker with a no-op for every test.
+
+ See _noop_status_worker for why this is necessary.
+ """
+ with mock.patch("controller.main.status_worker_loop", _noop_status_worker):
+ yield
+
+
@pytest.fixture
def test_directory() -> Generator[Path, None, None]:
"""Create a temporary directory for testing."""
diff --git a/controller/test/test_admin.py b/controller/test/test_admin.py
index ddce8b07..374e3b7b 100644
--- a/controller/test/test_admin.py
+++ b/controller/test/test_admin.py
@@ -119,78 +119,37 @@ class TestGetBluetoothStatus:
assert result == BluetoothStatus.unknown
-class TestEnableBluetooth:
- """Tests for enable_bluetooth function."""
+class TestBluetoothEnableDisable:
+ """Tests for enable_bluetooth / disable_bluetooth — they share the same shape."""
@pytest.mark.asyncio
- async def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = await enable_bluetooth()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
- @pytest.mark.asyncio
- async def test_calls_rfkill_unblock(self):
- """Should call rfkill unblock bluetooth."""
+ @pytest.mark.parametrize("func,rfkill_arg", [
+ (enable_bluetooth, "unblock"),
+ (disable_bluetooth, "block"),
+ ])
+ async def test_calls_rfkill_with_correct_action(self, func, rfkill_arg):
+ """Each function should invoke rfkill with the expected action."""
mock_proc = mock.AsyncMock()
mock_proc.communicate.return_value = (b"", b"")
mock_proc.returncode = 0
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc) as mock_exec:
- result = await enable_bluetooth()
+ result = await func()
assert result["success"] is True
mock_exec.assert_called_once_with(
- "/usr/sbin/rfkill", "unblock", "bluetooth",
+ "/usr/sbin/rfkill", rfkill_arg, "bluetooth",
stdout=mock.ANY,
stderr=mock.ANY,
)
@pytest.mark.asyncio
- async def test_handles_rfkill_not_found(self):
+ @pytest.mark.parametrize("func", [enable_bluetooth, disable_bluetooth])
+ async def test_handles_rfkill_not_found(self, func):
"""Should handle rfkill not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
with mock.patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError()):
- result = await enable_bluetooth()
- assert result["success"] is False
- assert "rfkill" in result.get("error", "").lower()
-
-
-class TestDisableBluetooth:
- """Tests for disable_bluetooth function."""
-
- @pytest.mark.asyncio
- async def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = await disable_bluetooth()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
- @pytest.mark.asyncio
- async def test_calls_rfkill_block(self):
- """Should call rfkill block bluetooth."""
- mock_proc = mock.AsyncMock()
- mock_proc.communicate.return_value = (b"", b"")
- mock_proc.returncode = 0
-
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
- with mock.patch("asyncio.create_subprocess_exec", return_value=mock_proc) as mock_exec:
- result = await disable_bluetooth()
- assert result["success"] is True
- mock_exec.assert_called_once_with(
- "/usr/sbin/rfkill", "block", "bluetooth",
- stdout=mock.ANY,
- stderr=mock.ANY,
- )
-
- @pytest.mark.asyncio
- async def test_handles_rfkill_not_found(self):
- """Should handle rfkill not being available."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
- with mock.patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError()):
- result = await disable_bluetooth()
+ result = await func()
assert result["success"] is False
@@ -336,16 +295,35 @@ class TestGetHotspotStatusDict:
assert result["enabled"] is True
+class TestDockerModeErrors:
+ """Admin functions should return an error result when running in Docker mode."""
+
+ @pytest.mark.parametrize("func,is_async,args", [
+ (enable_bluetooth, True, ()),
+ (disable_bluetooth, True, ()),
+ (enable_hotspot, False, ()),
+ (disable_hotspot, False, ()),
+ (enable_throttle, False, ()),
+ (disable_throttle, False, ()),
+ (shutdown_system, False, ()),
+ (reboot_system, False, ()),
+ (restart_all_services, True, ()),
+ (set_timezone, False, ("America/Denver",)),
+ ])
+ @pytest.mark.asyncio
+ async def test_returns_error_in_docker_mode(self, func, is_async, args):
+ """In Docker mode each function should refuse with a Docker-mentioning error."""
+ with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
+ result = await func(*args) if is_async else func(*args)
+ assert result["success"] is False
+ # set_timezone uses "error" string verbatim; others use the same shape.
+ error = result.get("error", "")
+ assert "Docker" in error
+
+
class TestEnableHotspot:
"""Tests for enable_hotspot function."""
- def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = enable_hotspot()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
def test_handles_nmcli_not_found(self):
"""Should handle nmcli not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
@@ -424,13 +402,6 @@ class TestEnableHotspot:
class TestDisableHotspot:
"""Tests for disable_hotspot function."""
- def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = disable_hotspot()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
def test_handles_nmcli_not_found(self):
"""Should handle nmcli not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
@@ -478,38 +449,9 @@ class TestGetThrottleStatus:
assert isinstance(result, GovernorStatus)
-class TestEnableThrottle:
- """Tests for enable_throttle function."""
-
- def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = enable_throttle()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
-
-class TestDisableThrottle:
- """Tests for disable_throttle function."""
-
- def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = disable_throttle()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
-
class TestShutdownSystem:
"""Tests for shutdown_system function."""
- def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = shutdown_system()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
def test_handles_shutdown_not_found(self):
"""Should handle shutdown command not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
@@ -537,13 +479,6 @@ class TestShutdownSystem:
class TestRebootSystem:
"""Tests for reboot_system function."""
- def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = reboot_system()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
def test_handles_reboot_not_found(self):
"""Should handle reboot command not being available."""
with mock.patch("controller.lib.admin.is_docker_mode", return_value=False):
@@ -571,14 +506,6 @@ class TestRebootSystem:
class TestRestartAllServices:
"""Tests for restart_all_services function."""
- @pytest.mark.asyncio
- async def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = await restart_all_services()
- assert result["success"] is False
- assert "Docker" in result.get("error", "")
-
@pytest.mark.asyncio
async def test_restarts_services(self):
"""Should attempt to restart all services."""
@@ -647,13 +574,6 @@ class TestGetTimezoneStatus:
class TestSetTimezone:
"""Tests for set_timezone function."""
- def test_returns_error_in_docker_mode(self):
- """Should return error when in Docker mode."""
- with mock.patch("controller.lib.admin.is_docker_mode", return_value=True):
- result = set_timezone("America/Denver")
- assert result["success"] is False
- assert "Docker" in result["error"]
-
def test_sets_timezone_successfully(self):
"""Should set timezone via timedatectl."""
mock_result = mock.Mock()
diff --git a/controller/test/test_admin_api.py b/controller/test/test_admin_api.py
index a5ccbcce..d5a3cafc 100644
--- a/controller/test/test_admin_api.py
+++ b/controller/test/test_admin_api.py
@@ -2,233 +2,61 @@
Integration tests for Controller admin API endpoints.
"""
+import pytest
-class TestHotspotStatusEndpoint:
- """Tests for /api/hotspot/status endpoint."""
- def test_hotspot_status_returns_200(self, test_client):
- """Hotspot status endpoint should return 200 OK."""
- response = test_client.get("/api/hotspot/status")
+class TestStatusEndpoints:
+ """Status endpoints (hotspot/bluetooth/throttle/timezone) — shape + docker behaviour."""
+
+ @pytest.mark.parametrize("endpoint,required_fields", [
+ ("/api/hotspot/status", ["enabled", "available"]),
+ ("/api/bluetooth/status", ["enabled", "available"]),
+ ("/api/throttle/status", ["enabled", "available"]),
+ ("/api/timezone/status", ["available", "timezone"]),
+ ])
+ def test_status_endpoint_shape(self, test_client, endpoint, required_fields):
+ """Each status endpoint should return 200 with the documented fields."""
+ response = test_client.get(endpoint)
assert response.status_code == 200
-
- def test_hotspot_status_returns_expected_fields(self, test_client):
- """Hotspot status should return enabled and available fields."""
- response = test_client.get("/api/hotspot/status")
data = response.json()
- assert "enabled" in data
- assert "available" in data
+ for field in required_fields:
+ assert field in data, f"{endpoint} missing field {field!r}"
- def test_hotspot_status_shows_unavailable_in_docker(self, test_client_docker_mode):
- """Hotspot should be unavailable in Docker mode."""
- response = test_client_docker_mode.get("/api/hotspot/status")
- data = response.json()
+ @pytest.mark.parametrize("endpoint", [
+ "/api/hotspot/status",
+ "/api/bluetooth/status",
+ "/api/throttle/status",
+ "/api/timezone/status",
+ ])
+ def test_status_endpoint_unavailable_in_docker(self, test_client_docker_mode, endpoint):
+ """Status endpoints should report available=False in Docker mode."""
+ data = test_client_docker_mode.get(endpoint).json()
assert data["available"] is False
-class TestHotspotEnableEndpoint:
- """Tests for /api/hotspot/enable endpoint."""
+class TestDockerModeRejectsAdminActions:
+ """Admin action endpoints should reject requests when running in Docker."""
- def test_hotspot_enable_fails_in_docker(self, test_client_docker_mode):
- """Hotspot enable should fail in Docker mode."""
- response = test_client_docker_mode.post("/api/hotspot/enable")
- assert response.status_code == 500
+ @pytest.mark.parametrize("method,endpoint,expected_status,payload", [
+ # Subsystem enable/disable — return 500 with a Docker-flavoured error.
+ ("post", "/api/hotspot/enable", 500, None),
+ ("post", "/api/hotspot/disable", 500, None),
+ ("post", "/api/bluetooth/enable", 500, None),
+ ("post", "/api/bluetooth/disable", 500, None),
+ ("post", "/api/throttle/enable", 500, None),
+ ("post", "/api/throttle/disable", 500, None),
+ ("post", "/api/timezone/set", 500, {"timezone": "America/Denver"}),
+ # System control — return 501 (not implemented under Docker).
+ ("post", "/api/shutdown", 501, None),
+ ("post", "/api/reboot", 501, None),
+ ("post", "/api/restart", 501, None),
+ ])
+ def test_action_rejected_in_docker(self, test_client_docker_mode, method, endpoint, expected_status, payload):
+ """Admin endpoints should reject Docker-mode requests with a Docker-mentioning error."""
+ client_call = getattr(test_client_docker_mode, method)
+ response = client_call(endpoint, json=payload) if payload else client_call(endpoint)
+ assert response.status_code == expected_status
assert "Docker" in response.json()["detail"]
-class TestHotspotDisableEndpoint:
- """Tests for /api/hotspot/disable endpoint."""
-
- def test_hotspot_disable_fails_in_docker(self, test_client_docker_mode):
- """Hotspot disable should fail in Docker mode."""
- response = test_client_docker_mode.post("/api/hotspot/disable")
- assert response.status_code == 500
- assert "Docker" in response.json()["detail"]
-
-
-class TestBluetoothStatusEndpoint:
- """Tests for /api/bluetooth/status endpoint."""
-
- def test_bluetooth_status_returns_200(self, test_client):
- """Bluetooth status endpoint should return 200 OK."""
- response = test_client.get("/api/bluetooth/status")
- assert response.status_code == 200
-
- def test_bluetooth_status_returns_expected_fields(self, test_client):
- """Bluetooth status should return enabled and available fields."""
- response = test_client.get("/api/bluetooth/status")
- data = response.json()
- assert "enabled" in data
- assert "available" in data
-
- def test_bluetooth_status_shows_unavailable_in_docker(self, test_client_docker_mode):
- """Bluetooth should be unavailable in Docker mode."""
- response = test_client_docker_mode.get("/api/bluetooth/status")
- data = response.json()
- assert data["available"] is False
-
-
-class TestBluetoothEnableEndpoint:
- """Tests for /api/bluetooth/enable endpoint."""
-
- def test_bluetooth_enable_fails_in_docker(self, test_client_docker_mode):
- """Bluetooth enable should fail in Docker mode."""
- response = test_client_docker_mode.post("/api/bluetooth/enable")
- assert response.status_code == 500
- assert "Docker" in response.json()["detail"]
-
-
-class TestBluetoothDisableEndpoint:
- """Tests for /api/bluetooth/disable endpoint."""
-
- def test_bluetooth_disable_fails_in_docker(self, test_client_docker_mode):
- """Bluetooth disable should fail in Docker mode."""
- response = test_client_docker_mode.post("/api/bluetooth/disable")
- assert response.status_code == 500
- assert "Docker" in response.json()["detail"]
-
-
-class TestThrottleStatusEndpoint:
- """Tests for /api/throttle/status endpoint."""
-
- def test_throttle_status_returns_200(self, test_client):
- """Throttle status endpoint should return 200 OK."""
- response = test_client.get("/api/throttle/status")
- assert response.status_code == 200
-
- def test_throttle_status_returns_expected_fields(self, test_client):
- """Throttle status should return enabled and available fields."""
- response = test_client.get("/api/throttle/status")
- data = response.json()
- assert "enabled" in data
- assert "available" in data
-
- def test_throttle_status_shows_unavailable_in_docker(self, test_client_docker_mode):
- """Throttle should be unavailable in Docker mode."""
- response = test_client_docker_mode.get("/api/throttle/status")
- data = response.json()
- assert data["available"] is False
-
-
-class TestThrottleEnableEndpoint:
- """Tests for /api/throttle/enable endpoint."""
-
- def test_throttle_enable_fails_in_docker(self, test_client_docker_mode):
- """Throttle enable should fail in Docker mode."""
- response = test_client_docker_mode.post("/api/throttle/enable")
- assert response.status_code == 500
- assert "Docker" in response.json()["detail"]
-
-
-class TestThrottleDisableEndpoint:
- """Tests for /api/throttle/disable endpoint."""
-
- def test_throttle_disable_fails_in_docker(self, test_client_docker_mode):
- """Throttle disable should fail in Docker mode."""
- response = test_client_docker_mode.post("/api/throttle/disable")
- assert response.status_code == 500
- assert "Docker" in response.json()["detail"]
-
-
-class TestShutdownEndpoint:
- """Tests for /api/shutdown endpoint."""
-
- def test_shutdown_returns_501_in_docker(self, test_client_docker_mode):
- """Shutdown should return 501 in Docker mode."""
- response = test_client_docker_mode.post("/api/shutdown")
- assert response.status_code == 501
- assert "Docker" in response.json()["detail"]
-
-
-class TestRebootEndpoint:
- """Tests for /api/reboot endpoint."""
-
- def test_reboot_returns_501_in_docker(self, test_client_docker_mode):
- """Reboot should return 501 in Docker mode."""
- response = test_client_docker_mode.post("/api/reboot")
- assert response.status_code == 501
- assert "Docker" in response.json()["detail"]
-
-
-class TestRestartEndpoint:
- """Tests for /api/restart endpoint."""
-
- def test_restart_returns_501_in_docker(self, test_client_docker_mode):
- """Restart should return 501 in Docker mode."""
- response = test_client_docker_mode.post("/api/restart")
- assert response.status_code == 501
- assert "Docker" in response.json()["detail"]
-
-
-class TestTimezoneStatusEndpoint:
- """Tests for /api/timezone/status endpoint."""
-
- def test_timezone_status_returns_200(self, test_client):
- """Timezone status endpoint should return 200 OK."""
- response = test_client.get("/api/timezone/status")
- assert response.status_code == 200
-
- def test_timezone_status_returns_expected_fields(self, test_client):
- """Timezone status should return available and timezone fields."""
- response = test_client.get("/api/timezone/status")
- data = response.json()
- assert "available" in data
- assert "timezone" in data
-
- def test_timezone_status_shows_unavailable_in_docker(self, test_client_docker_mode):
- """Timezone should be unavailable in Docker mode."""
- response = test_client_docker_mode.get("/api/timezone/status")
- data = response.json()
- assert data["available"] is False
-
-
-class TestTimezoneSetEndpoint:
- """Tests for /api/timezone/set endpoint."""
-
- def test_timezone_set_fails_in_docker(self, test_client_docker_mode):
- """Timezone set should fail in Docker mode."""
- response = test_client_docker_mode.post("/api/timezone/set", json={"timezone": "America/Denver"})
- assert response.status_code == 500
- assert "Docker" in response.json()["detail"]
-
-
-class TestOpenAPIIncludesAdminEndpoints:
- """Tests that OpenAPI documentation includes admin endpoints."""
-
- def test_openapi_has_hotspot_paths(self, test_client):
- """OpenAPI schema should include hotspot paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/hotspot/status" in data["paths"]
- assert "/api/hotspot/enable" in data["paths"]
- assert "/api/hotspot/disable" in data["paths"]
-
- def test_openapi_has_bluetooth_paths(self, test_client):
- """OpenAPI schema should include bluetooth paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/bluetooth/status" in data["paths"]
- assert "/api/bluetooth/enable" in data["paths"]
- assert "/api/bluetooth/disable" in data["paths"]
-
- def test_openapi_has_throttle_paths(self, test_client):
- """OpenAPI schema should include throttle paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/throttle/status" in data["paths"]
- assert "/api/throttle/enable" in data["paths"]
- assert "/api/throttle/disable" in data["paths"]
-
- def test_openapi_has_timezone_paths(self, test_client):
- """OpenAPI schema should include timezone paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/timezone/status" in data["paths"]
- assert "/api/timezone/set" in data["paths"]
-
- def test_openapi_has_system_control_paths(self, test_client):
- """OpenAPI schema should include system control paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/shutdown" in data["paths"]
- assert "/api/reboot" in data["paths"]
- assert "/api/restart" in data["paths"]
+# OpenAPI endpoint-presence tests are consolidated in test_api.py::TestOpenAPI.
diff --git a/controller/test/test_api.py b/controller/test/test_api.py
index b40beb87..51f1c5bf 100644
--- a/controller/test/test_api.py
+++ b/controller/test/test_api.py
@@ -1,6 +1,7 @@
"""
Integration tests for Controller API endpoints.
"""
+import re
from unittest import mock
import pytest
@@ -37,133 +38,99 @@ class TestHealthEndpoint:
class TestDashboardEndpoint:
"""Tests for / dashboard endpoint (UI)."""
- def test_dashboard_returns_200(self, test_client):
- """Dashboard endpoint should return 200 OK."""
+ def test_dashboard_basic_response(self, test_client):
+ """Dashboard should return 200 with HTML content-type and a recognisable body."""
response = test_client.get("/")
assert response.status_code == 200
-
- def test_dashboard_returns_html(self, test_client):
- """Dashboard endpoint should return HTML."""
- response = test_client.get("/")
assert "text/html" in response.headers["content-type"]
-
- def test_dashboard_contains_title(self, test_client):
- """Dashboard should contain title."""
- response = test_client.get("/")
- # Title format is "Controller - {hostname} WROLPi"
assert b"Controller" in response.content
assert b"WROLPi" in response.content
- def test_dashboard_contains_status_cards(self, test_client):
- """Dashboard should contain status cards."""
+ @pytest.mark.parametrize("needle", [
+ # Status cards.
+ "CPU", "Memory", "Load", "Storage",
+ # Sections / structural ids.
+ "Disks", 'id="disks"', "Services", 'id="services"',
+ # System actions.
+ "System Actions", "Restart All Services",
+ # Mount modal.
+ 'id="mount-modal"', 'id="mount-point-input"', 'id="mount-persist-checkbox"',
+ "Persistent (survive reboots)",
+ # Disks JS table.
+ "| Persist | ", "function togglePersist", "api/disks/fstab",
+ ])
+ def test_dashboard_contains(self, test_client, needle):
+ """Dashboard body should contain each expected fragment (native mode)."""
response = test_client.get("/")
- content = response.text
- assert "CPU" in content
- assert "Memory" in content
- assert "Load" in content
- assert "Storage" in content
+ assert needle in response.text
- def test_dashboard_contains_disks_section(self, test_client):
- """Dashboard should contain disks section."""
- response = test_client.get("/")
- content = response.text
- assert "Disks" in content
- assert 'id="disks"' in content
-
- def test_dashboard_contains_services_section(self, test_client):
- """Dashboard should contain services section."""
- response = test_client.get("/")
- content = response.text
- assert "Services" in content
- assert 'id="services"' in content
-
- def test_dashboard_contains_system_actions(self, test_client):
- """Dashboard should contain system actions section."""
- response = test_client.get("/")
- content = response.text
- assert "System Actions" in content
- assert "Restart All Services" in content
-
- def test_dashboard_shows_docker_banner_when_dockerized(self, test_client_docker_mode):
- """Dashboard should show Docker mode banner when in Docker."""
- response = test_client_docker_mode.get("/")
- content = response.text
- assert "Running in Docker mode" in content
-
- def test_dashboard_no_docker_banner_when_not_dockerized(self, test_client):
- """Dashboard should not show Docker mode banner when not in Docker."""
- response = test_client.get("/")
- content = response.text
- assert "Running in Docker mode" not in content
-
- def test_dashboard_shows_boot_column_when_not_dockerized(self, test_client):
- """Dashboard should show Boot column in services table when not in Docker."""
+ def test_dashboard_native_mode_shows_boot_column_and_hides_docker_banner(self, test_client):
+ """Native (non-Docker) mode: Boot column visible, no Docker banner."""
response = test_client.get("/")
content = response.text
assert "Boot | " in content
- # Should have toggle switches
assert 'class="toggle"' in content
+ assert "Running in Docker mode" not in content
- def test_dashboard_no_boot_column_when_dockerized(self, test_client_docker_mode):
- """Dashboard should not show Boot column in services table when in Docker."""
+ def test_dashboard_docker_mode_shows_banner_and_hides_boot_column(self, test_client_docker_mode):
+ """Docker mode: Docker banner visible, no Boot toggles in rendered tbody."""
response = test_client_docker_mode.get("/")
content = response.text
+ assert "Running in Docker mode" in content
assert "Boot | " not in content
- # The table body (rendered server-side) should not have boot toggles
- # Note: JavaScript template literals may contain 'id="boot-' for dynamic updates,
- # but the actual rendered should not have boot toggles in Docker mode
- import re
+ # JS template literals may mention 'id="boot-' but the server-rendered tbody must not.
tbody_match = re.search(r'(.*?)', content, re.DOTALL)
if tbody_match:
- tbody_content = tbody_match.group(1)
- assert 'id="boot-' not in tbody_content
-
- def test_dashboard_has_mount_modal(self, test_client):
- """Dashboard should have mount modal for disk mounting."""
- response = test_client.get("/")
- content = response.text
- assert 'id="mount-modal"' in content
- assert 'id="mount-point-input"' in content
- assert 'id="mount-persist-checkbox"' in content
- assert "Persistent (survive reboots)" in content
-
- def test_dashboard_has_persist_column_in_disks_js(self, test_client):
- """Dashboard JavaScript should include Persist column in disk table."""
- response = test_client.get("/")
- content = response.text
- # The loadDisks() function should create a table with Persist column
- assert "Persist | " in content
- # Should have togglePersist function
- assert "function togglePersist" in content
- # Should fetch fstab entries
- assert "api/disks/fstab" in content
+ assert 'id="boot-' not in tbody_match.group(1)
-class TestOpenAPIEndpoint:
- """Tests for OpenAPI documentation endpoint."""
+class TestOpenAPI:
+ """Tests for the OpenAPI schema and its expected endpoint coverage."""
- def test_openapi_returns_200(self, test_client):
- """OpenAPI endpoint should return 200 OK."""
+ @pytest.fixture
+ def openapi(self, test_client):
+ """Fetch the OpenAPI schema once per parametrized test."""
response = test_client.get("/openapi.json")
assert response.status_code == 200
+ return response.json()
- def test_openapi_has_title(self, test_client):
- """OpenAPI schema should have correct title."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert data["info"]["title"] == "WROLPi Controller"
+ def test_openapi_metadata(self, openapi):
+ """Schema title and version should match the package metadata."""
+ assert openapi["info"]["title"] == "WROLPi Controller"
+ assert openapi["info"]["version"] == __version__
- def test_openapi_has_version(self, test_client):
- """OpenAPI schema should have version."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert data["info"]["version"] == __version__
+ @pytest.mark.parametrize("path", [
+ # Health.
+ "/api/health",
+ # Stats (per-resource + aggregated).
+ "/api/stats/cpu", "/api/stats/memory", "/api/stats/load",
+ "/api/stats/drives/primary", "/api/stats/network", "/api/stats/power",
+ # Admin: hotspot, bluetooth, throttle, timezone, power.
+ "/api/hotspot/status", "/api/hotspot/enable", "/api/hotspot/disable",
+ "/api/bluetooth/status", "/api/bluetooth/enable", "/api/bluetooth/disable",
+ "/api/throttle/status", "/api/throttle/enable", "/api/throttle/disable",
+ "/api/timezone/status", "/api/timezone/set",
+ "/api/shutdown", "/api/reboot", "/api/restart",
+ # Disks.
+ "/api/disks", "/api/disks/mounts", "/api/disks/mount", "/api/disks/unmount",
+ "/api/disks/fstab", "/api/disks/smart",
+ # Services.
+ "/api/services", "/api/services/{name}",
+ "/api/services/{name}/start", "/api/services/{name}/stop",
+ "/api/services/{name}/restart", "/api/services/{name}/enable",
+ "/api/services/{name}/disable", "/api/services/{name}/logs",
+ # Scripts.
+ "/api/scripts", "/api/scripts/status",
+ "/api/scripts/{name}/start", "/api/scripts/{name}/output",
+ ])
+ def test_openapi_has_path(self, openapi, path):
+ """Every documented API path should be present in the OpenAPI schema."""
+ assert path in openapi["paths"], f"OpenAPI schema is missing {path}"
- def test_openapi_has_api_paths(self, test_client):
- """OpenAPI schema should have API paths defined."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/health" in data["paths"]
+ def test_openapi_has_aggregated_stats(self, openapi):
+ """Aggregated /api/stats endpoint (may register with or without trailing slash)."""
+ paths = openapi["paths"]
+ assert "/api/stats" in paths or "/api/stats/" in paths
class TestDocsEndpoint:
diff --git a/controller/test/test_disks.py b/controller/test/test_disks.py
index 9d8bba35..76e271a7 100644
--- a/controller/test/test_disks.py
+++ b/controller/test/test_disks.py
@@ -268,85 +268,31 @@ class TestMountDrive:
assert result["success"] is False
assert "mount failed" in result["error"]
- def test_exfat_mount_includes_uid_gid(self):
- """Should add uid/gid options when mounting exfat."""
- with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
- with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
- with mock.patch("pathlib.Path.mkdir"):
- with mock.patch("subprocess.run") as mock_run:
- mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
- result = mount_drive("/dev/sda1", "/media/test", fstype="exfat")
- assert result["success"] is True
- # Verify mount was called with uid/gid options
- call_args = mock_run.call_args[0][0]
- assert "-o" in call_args
- options_idx = call_args.index("-o") + 1
- options = call_args[options_idx]
- assert "uid=1001" in options
- assert "gid=1001" in options
-
- def test_ext4_mount_does_not_include_uid_gid(self):
- """Should not add uid/gid options when mounting ext4."""
- with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
- with mock.patch("pathlib.Path.mkdir"):
- with mock.patch("subprocess.run") as mock_run:
- mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
- result = mount_drive("/dev/sda1", "/media/test", fstype="ext4")
- assert result["success"] is True
- # Verify mount was called without uid/gid options
- call_args = mock_run.call_args[0][0]
- assert "-o" in call_args
- options_idx = call_args.index("-o") + 1
- options = call_args[options_idx]
- assert "uid=" not in options
- assert "gid=" not in options
-
- def test_vfat_mount_includes_uid_gid(self):
- """Should add uid/gid options when mounting vfat (FAT32)."""
- with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
- with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
- with mock.patch("pathlib.Path.mkdir"):
- with mock.patch("subprocess.run") as mock_run:
- mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
- result = mount_drive("/dev/sda1", "/media/test", fstype="vfat")
- assert result["success"] is True
- # Verify mount was called with uid/gid options
- call_args = mock_run.call_args[0][0]
- assert "-o" in call_args
- options_idx = call_args.index("-o") + 1
- options = call_args[options_idx]
- assert "uid=1001" in options
- assert "gid=1001" in options
-
- def test_ntfs_mount_includes_uid_gid(self):
- """Should add uid/gid options when mounting ntfs (Windows-formatted drive)."""
- with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
- with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
- with mock.patch("pathlib.Path.mkdir"):
- with mock.patch("subprocess.run") as mock_run:
- mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
- result = mount_drive("/dev/sda1", "/media/test", fstype="ntfs")
- assert result["success"] is True
- call_args = mock_run.call_args[0][0]
- options_idx = call_args.index("-o") + 1
- options = call_args[options_idx]
- assert "uid=1001" in options
- assert "gid=1001" in options
-
- def test_ntfs3_mount_includes_uid_gid(self):
- """Same handling for the in-kernel ntfs3 driver (kernel 5.15+)."""
- with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
- with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
- with mock.patch("pathlib.Path.mkdir"):
- with mock.patch("subprocess.run") as mock_run:
- mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
- result = mount_drive("/dev/sda1", "/media/test", fstype="ntfs3")
- assert result["success"] is True
- call_args = mock_run.call_args[0][0]
- options_idx = call_args.index("-o") + 1
- options = call_args[options_idx]
- assert "uid=1001" in options
- assert "gid=1001" in options
+ @pytest.mark.parametrize("fstype,expect_uid_gid", [
+ ("exfat", True),
+ ("vfat", True), # FAT32
+ ("ntfs", True), # ntfs-3g userspace driver
+ ("ntfs3", True), # in-kernel ntfs3 driver (kernel 5.15+)
+ ("ext4", False), # POSIX permissions; no uid/gid mount opts
+ ])
+ def test_mount_uid_gid_options_by_fstype(self, fstype, expect_uid_gid):
+ """Foreign-fs mounts (exfat/vfat/ntfs/ntfs3) inject uid/gid; ext4 must not."""
+ with mock.patch("controller.lib.disks.is_docker_mode", return_value=False), \
+ mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)), \
+ mock.patch("pathlib.Path.mkdir"), \
+ mock.patch("subprocess.run") as mock_run:
+ mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
+ result = mount_drive("/dev/sda1", "/media/test", fstype=fstype)
+ assert result["success"] is True
+ call_args = mock_run.call_args[0][0]
+ assert "-o" in call_args
+ options = call_args[call_args.index("-o") + 1]
+ if expect_uid_gid:
+ assert "uid=1001" in options
+ assert "gid=1001" in options
+ else:
+ assert "uid=" not in options
+ assert "gid=" not in options
class TestGetWrolpiUidGid:
diff --git a/controller/test/test_disks_api.py b/controller/test/test_disks_api.py
index 977cf909..d5e76d88 100644
--- a/controller/test/test_disks_api.py
+++ b/controller/test/test_disks_api.py
@@ -4,15 +4,32 @@ Integration tests for Controller disks API endpoints.
from unittest import mock
+import pytest
+
+
+class TestDockerModeRejectsDisksEndpoints:
+ """All /api/disks/* endpoints should return 501 in Docker mode."""
+
+ @pytest.mark.parametrize("method,endpoint,payload", [
+ ("get", "/api/disks", None),
+ ("get", "/api/disks/mounts", None),
+ ("post", "/api/disks/mount", {"device": "/dev/sda1", "mount_point": "/media/test"}),
+ ("post", "/api/disks/unmount", {"mount_point": "/media/test"}),
+ ("get", "/api/disks/fstab", None),
+ ("post", "/api/disks/fstab", {"device": "/dev/sda1", "mount_point": "/media/test", "fstype": "ext4"}),
+ ("delete", "/api/disks/fstab/media/test", None),
+ ("get", "/api/disks/smart", None),
+ ])
+ def test_disks_endpoint_rejected_in_docker(self, test_client_docker_mode, method, endpoint, payload):
+ """Docker mode rejects every disks endpoint with 501."""
+ client_call = getattr(test_client_docker_mode, method)
+ response = client_call(endpoint, json=payload) if payload else client_call(endpoint)
+ assert response.status_code == 501
+
class TestDisksListEndpoint:
"""Tests for /api/disks endpoint."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.get("/api/disks")
- assert response.status_code == 501
-
def test_returns_list_in_native_mode(self, test_client):
"""Should return list of disks in native mode."""
with mock.patch("controller.api.disks.get_block_devices", return_value=[]):
@@ -24,11 +41,6 @@ class TestDisksListEndpoint:
class TestMountsEndpoint:
"""Tests for /api/disks/mounts endpoint."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.get("/api/disks/mounts")
- assert response.status_code == 501
-
def test_returns_mounts(self, test_client):
"""Should return current mounts."""
mock_mounts = [{"mount_point": "/media/wrolpi", "device": "/dev/sda1"}]
@@ -41,14 +53,6 @@ class TestMountsEndpoint:
class TestMountEndpoint:
"""Tests for /api/disks/mount endpoint."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.post(
- "/api/disks/mount",
- json={"device": "/dev/sda1", "mount_point": "/media/test"}
- )
- assert response.status_code == 501
-
def test_mounts_drive_successfully(self, test_client):
"""Should mount drive successfully."""
with mock.patch(
@@ -111,14 +115,6 @@ class TestMountEndpoint:
class TestUnmountEndpoint:
"""Tests for /api/disks/unmount endpoint."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.post(
- "/api/disks/unmount",
- json={"mount_point": "/media/test"}
- )
- assert response.status_code == 501
-
def test_unmounts_drive_successfully(self, test_client):
"""Should unmount drive successfully."""
with mock.patch(
@@ -135,11 +131,6 @@ class TestUnmountEndpoint:
class TestFstabEndpoints:
"""Tests for /api/disks/fstab endpoints."""
- def test_list_fstab_returns_501_in_docker(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.get("/api/disks/fstab")
- assert response.status_code == 501
-
def test_list_fstab_returns_entries(self, test_client):
"""Should return fstab entries."""
mock_entries = [{"mount_point": "/media/wrolpi", "device": "UUID=1234"}]
@@ -148,14 +139,6 @@ class TestFstabEndpoints:
assert response.status_code == 200
assert response.json() == mock_entries
- def test_add_fstab_returns_501_in_docker(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.post(
- "/api/disks/fstab",
- json={"device": "/dev/sda1", "mount_point": "/media/test", "fstype": "ext4"}
- )
- assert response.status_code == 501
-
def test_add_fstab_entry(self, test_client):
"""Should add fstab entry."""
with mock.patch(
@@ -168,20 +151,10 @@ class TestFstabEndpoints:
)
assert response.status_code == 200
- def test_delete_fstab_returns_501_in_docker(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.delete("/api/disks/fstab/media/test")
- assert response.status_code == 501
-
class TestSmartEndpoints:
"""Tests for /api/disks/smart endpoints."""
- def test_list_smart_returns_501_in_docker(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.get("/api/disks/smart")
- assert response.status_code == 501
-
def test_list_smart_when_not_available(self, test_client):
"""Should return unavailable when SMART not available."""
with mock.patch("controller.api.disks.is_smart_available", return_value=False):
@@ -339,16 +312,4 @@ class TestDeleteFstabRemovesConfig:
assert mounts[0]["mount_point"] == "/media/other"
-class TestOpenAPIIncludesDisksEndpoints:
- """Tests that OpenAPI documentation includes disks endpoints."""
-
- def test_openapi_has_disks_paths(self, test_client):
- """OpenAPI schema should include disks paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/disks" in data["paths"]
- assert "/api/disks/mounts" in data["paths"]
- assert "/api/disks/mount" in data["paths"]
- assert "/api/disks/unmount" in data["paths"]
- assert "/api/disks/fstab" in data["paths"]
- assert "/api/disks/smart" in data["paths"]
+# OpenAPI endpoint-presence tests are consolidated in test_api.py::TestOpenAPI.
diff --git a/controller/test/test_onboarding_api.py b/controller/test/test_onboarding_api.py
index 39e905ce..97ecc824 100644
--- a/controller/test/test_onboarding_api.py
+++ b/controller/test/test_onboarding_api.py
@@ -4,23 +4,38 @@ Integration tests for onboarding API endpoints.
from unittest import mock
+import pytest
+
+
+class TestOnboardingDockerAndMounted:
+ """Cross-cutting behaviour: Docker mode → 501; drive already mounted → 409."""
+
+ @pytest.mark.parametrize("method,endpoint,payload", [
+ ("get", "/api/onboarding/candidates", None),
+ ("post", "/api/onboarding/probe", {"device_path": "/dev/sda1", "fstype": "ext4"}),
+ ("post", "/api/onboarding/commit", {"device_path": "/dev/sda1", "fstype": "ext4"}),
+ ("post", "/api/onboarding/cancel", None),
+ ])
+ def test_endpoint_returns_501_in_docker(self, test_client_docker_mode, method, endpoint, payload):
+ client_call = getattr(test_client_docker_mode, method)
+ response = client_call(endpoint, json=payload) if payload else client_call(endpoint)
+ assert response.status_code == 501
+
+ @pytest.mark.parametrize("method,endpoint,payload", [
+ ("get", "/api/onboarding/candidates", None),
+ ("post", "/api/onboarding/probe", {"device_path": "/dev/sda1", "fstype": "ext4"}),
+ ])
+ def test_endpoint_returns_409_when_drive_mounted(self, test_client, method, endpoint, payload):
+ """Onboarding endpoints reject when the primary drive is already mounted."""
+ with mock.patch("controller.api.onboarding.is_primary_drive_mounted", return_value=True):
+ client_call = getattr(test_client, method)
+ response = client_call(endpoint, json=payload) if payload else client_call(endpoint)
+ assert response.status_code == 409
+
class TestOnboardingCandidatesEndpoint:
"""Tests for GET /api/onboarding/candidates."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.get("/api/onboarding/candidates")
- assert response.status_code == 501
-
- def test_returns_409_when_drive_mounted(self, test_client):
- """Should return 409 when primary drive is already mounted."""
- with mock.patch(
- "controller.api.onboarding.is_primary_drive_mounted", return_value=True,
- ):
- response = test_client.get("/api/onboarding/candidates")
- assert response.status_code == 409
-
def test_returns_candidates(self, test_client):
"""Should return list of unmounted drives."""
with mock.patch(
@@ -53,25 +68,6 @@ class TestOnboardingCandidatesEndpoint:
class TestOnboardingProbeEndpoint:
"""Tests for POST /api/onboarding/probe."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.post(
- "/api/onboarding/probe",
- json={"device_path": "/dev/sda1", "fstype": "ext4"},
- )
- assert response.status_code == 501
-
- def test_returns_409_when_drive_mounted(self, test_client):
- """Should return 409 when primary drive is already mounted."""
- with mock.patch(
- "controller.api.onboarding.is_primary_drive_mounted", return_value=True,
- ):
- response = test_client.post(
- "/api/onboarding/probe",
- json={"device_path": "/dev/sda1", "fstype": "ext4"},
- )
- assert response.status_code == 409
-
def test_probe_success_with_config(self, test_client):
"""Should return probe result with config found."""
with mock.patch(
@@ -113,14 +109,6 @@ class TestOnboardingProbeEndpoint:
class TestOnboardingCommitEndpoint:
"""Tests for POST /api/onboarding/commit."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.post(
- "/api/onboarding/commit",
- json={"device_path": "/dev/sda1", "fstype": "ext4"},
- )
- assert response.status_code == 501
-
def test_commit_success(self, test_client):
"""Should return success on commit."""
with mock.patch(
@@ -200,11 +188,6 @@ class TestOnboardingCommitEndpoint:
class TestOnboardingCancelEndpoint:
"""Tests for POST /api/onboarding/cancel."""
- def test_returns_501_in_docker_mode(self, test_client_docker_mode):
- """Should return 501 in Docker mode."""
- response = test_client_docker_mode.post("/api/onboarding/cancel")
- assert response.status_code == 501
-
def test_cancel_success(self, test_client):
"""Should return success on cancel."""
with mock.patch(
diff --git a/controller/test/test_scripts_api.py b/controller/test/test_scripts_api.py
index f195f574..41a07951 100644
--- a/controller/test/test_scripts_api.py
+++ b/controller/test/test_scripts_api.py
@@ -4,6 +4,21 @@ Integration tests for Controller scripts API endpoints.
from unittest import mock
+import pytest
+
+
+class TestDockerModeRejectsScriptsEndpoints:
+ """Scripts action endpoints should return 501 in Docker mode."""
+
+ @pytest.mark.parametrize("method,endpoint", [
+ ("post", "/api/scripts/repair/start"),
+ ("get", "/api/scripts/repair/output"),
+ ])
+ def test_action_rejected_in_docker(self, test_client_docker_mode, method, endpoint):
+ response = getattr(test_client_docker_mode, method)(endpoint)
+ assert response.status_code == 501
+ assert "Docker" in response.json()["detail"]
+
class TestScriptsListEndpoint:
"""Tests for /api/scripts endpoint."""
@@ -69,12 +84,6 @@ class TestScriptsStatusEndpoint:
class TestScriptsStartEndpoint:
"""Tests for /api/scripts/{name}/start endpoint."""
- def test_scripts_start_returns_501_in_docker(self, test_client_docker_mode):
- """Scripts start should return 501 in Docker mode."""
- response = test_client_docker_mode.post("/api/scripts/repair/start")
- assert response.status_code == 501
- assert "Docker" in response.json()["detail"]
-
def test_scripts_start_returns_400_for_unknown_script(self, test_client):
"""Scripts start should return 400 for unknown script."""
with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
@@ -96,12 +105,6 @@ class TestScriptsStartEndpoint:
class TestScriptsOutputEndpoint:
"""Tests for /api/scripts/{name}/output endpoint."""
- def test_scripts_output_returns_501_in_docker(self, test_client_docker_mode):
- """Scripts output should return 501 in Docker mode."""
- response = test_client_docker_mode.get("/api/scripts/repair/output")
- assert response.status_code == 501
- assert "Docker" in response.json()["detail"]
-
def test_scripts_output_returns_200(self, test_client):
"""Scripts output should return 200 when not in Docker mode."""
with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
@@ -152,14 +155,4 @@ class TestScriptsOutputEndpoint:
assert "5000" in call_args
-class TestOpenAPIIncludesScriptsEndpoints:
- """Tests that OpenAPI documentation includes scripts endpoints."""
-
- def test_openapi_has_scripts_paths(self, test_client):
- """OpenAPI schema should include scripts paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/scripts" in data["paths"]
- assert "/api/scripts/status" in data["paths"]
- assert "/api/scripts/{name}/start" in data["paths"]
- assert "/api/scripts/{name}/output" in data["paths"]
+# OpenAPI endpoint-presence tests are consolidated in test_api.py::TestOpenAPI.
diff --git a/controller/test/test_services_api.py b/controller/test/test_services_api.py
index ccb96721..390dddd5 100644
--- a/controller/test/test_services_api.py
+++ b/controller/test/test_services_api.py
@@ -162,18 +162,4 @@ class TestServiceLogsEndpoint:
assert response.status_code == 200
-class TestOpenAPIIncludesServicesEndpoints:
- """Tests that OpenAPI documentation includes services endpoints."""
-
- def test_openapi_has_services_paths(self, test_client):
- """OpenAPI schema should include services paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/services" in data["paths"]
- assert "/api/services/{name}" in data["paths"]
- assert "/api/services/{name}/start" in data["paths"]
- assert "/api/services/{name}/stop" in data["paths"]
- assert "/api/services/{name}/restart" in data["paths"]
- assert "/api/services/{name}/enable" in data["paths"]
- assert "/api/services/{name}/disable" in data["paths"]
- assert "/api/services/{name}/logs" in data["paths"]
+# OpenAPI endpoint-presence tests are consolidated in test_api.py::TestOpenAPI.
diff --git a/controller/test/test_status_api.py b/controller/test/test_status_api.py
index b5653366..c1b4ff58 100644
--- a/controller/test/test_status_api.py
+++ b/controller/test/test_status_api.py
@@ -5,187 +5,60 @@ The stats endpoints return system statistics (CPU, memory, etc).
Note: These are separate from the main WROLPi API's /api/status endpoint.
"""
+import pytest
-class TestCpuEndpoint:
- """Tests for /api/stats/cpu endpoint."""
- def test_cpu_returns_200(self, test_client):
- """CPU endpoint should return 200 OK."""
- response = test_client.get("/api/stats/cpu")
+class TestStatsEndpoints:
+ """Tests for /api/stats/* endpoints — status code, payload shape, field set."""
+
+ @pytest.mark.parametrize("endpoint,expected_type,required_fields", [
+ ("/api/stats/cpu", dict, [
+ "percent", "cores",
+ "cur_frequency", "max_frequency", "min_frequency",
+ "temperature", "high_temperature", "critical_temperature",
+ ]),
+ ("/api/stats/memory", dict, ["total", "used", "free", "cached"]),
+ ("/api/stats/load", dict, ["minute_1", "minute_5", "minute_15"]),
+ ("/api/stats/drives/primary", dict, ["mounted"]),
+ ("/api/stats/network", dict, []),
+ ("/api/stats/power", dict, ["under_voltage", "over_current"]),
+ ])
+ def test_endpoint_shape(self, test_client, endpoint, expected_type, required_fields):
+ """Each stats endpoint should return 200 with the expected payload shape and fields."""
+ response = test_client.get(endpoint)
assert response.status_code == 200
-
- def test_cpu_returns_expected_fields(self, test_client):
- """CPU endpoint should return expected fields (main API format)."""
- response = test_client.get("/api/stats/cpu")
data = response.json()
- assert "percent" in data
- assert "cores" in data
- assert "cur_frequency" in data
- assert "max_frequency" in data
- assert "min_frequency" in data
- assert "temperature" in data
- assert "high_temperature" in data
- assert "critical_temperature" in data
+ assert isinstance(data, expected_type)
+ for field in required_fields:
+ assert field in data, f"{endpoint} missing field {field!r}"
+
+
+class TestStatsValueSemantics:
+ """Tests that verify value ranges/types — not just presence."""
def test_cpu_percent_valid_range(self, test_client):
"""CPU percent should be between 0 and 100."""
- response = test_client.get("/api/stats/cpu")
- data = response.json()
+ data = test_client.get("/api/stats/cpu").json()
assert 0 <= data["percent"] <= 100
-
-class TestMemoryEndpoint:
- """Tests for /api/stats/memory endpoint."""
-
- def test_memory_returns_200(self, test_client):
- """Memory endpoint should return 200 OK."""
- response = test_client.get("/api/stats/memory")
- assert response.status_code == 200
-
- def test_memory_returns_expected_fields(self, test_client):
- """Memory endpoint should return expected fields (main API format)."""
- response = test_client.get("/api/stats/memory")
- data = response.json()
- assert "total" in data
- assert "used" in data
- assert "free" in data
- assert "cached" in data
-
- def test_memory_bytes_positive(self, test_client):
+ def test_memory_total_positive(self, test_client):
"""Memory total should be positive."""
- response = test_client.get("/api/stats/memory")
- data = response.json()
+ data = test_client.get("/api/stats/memory").json()
assert data["total"] > 0
-
-class TestLoadEndpoint:
- """Tests for /api/stats/load endpoint."""
-
- def test_load_returns_200(self, test_client):
- """Load endpoint should return 200 OK."""
- response = test_client.get("/api/stats/load")
- assert response.status_code == 200
-
- def test_load_returns_expected_fields(self, test_client):
- """Load endpoint should return expected fields (main API format)."""
- response = test_client.get("/api/stats/load")
- data = response.json()
- assert "minute_1" in data
- assert "minute_5" in data
- assert "minute_15" in data
-
def test_load_values_non_negative(self, test_client):
- """Load values should be non-negative."""
- response = test_client.get("/api/stats/load")
- data = response.json()
- # Values are strings in main API format
- assert float(data["minute_1"]) >= 0
- assert float(data["minute_5"]) >= 0
- assert float(data["minute_15"]) >= 0
-
-
-class TestPrimaryDriveEndpoint:
- """Tests for /api/stats/drives/primary endpoint."""
-
- def test_primary_drive_returns_200(self, test_client):
- """Primary drive endpoint should return 200 OK."""
- response = test_client.get("/api/stats/drives/primary")
- assert response.status_code == 200
-
- def test_primary_drive_returns_mounted_false_when_not_mounted(self, test_client):
- """Primary drive should return mounted=false when not present."""
- response = test_client.get("/api/stats/drives/primary")
- data = response.json()
- # In test environment, /media/wrolpi likely not mounted
- assert "mounted" in data
-
-
-class TestNetworkEndpoint:
- """Tests for /api/stats/network endpoint."""
-
- def test_network_returns_200(self, test_client):
- """Network endpoint should return 200 OK."""
- response = test_client.get("/api/stats/network")
- assert response.status_code == 200
-
- def test_network_returns_dict(self, test_client):
- """Network endpoint should return a dict (main API format)."""
- response = test_client.get("/api/stats/network")
- data = response.json()
- assert isinstance(data, dict)
+ """Load values are strings in main API format; all should be non-negative numbers."""
+ data = test_client.get("/api/stats/load").json()
+ for key in ("minute_1", "minute_5", "minute_15"):
+ assert float(data[key]) >= 0
def test_network_excludes_loopback(self, test_client):
- """Network should not include loopback interface."""
- response = test_client.get("/api/stats/network")
- data = response.json()
+ """Network should not include the loopback interface."""
+ data = test_client.get("/api/stats/network").json()
assert "lo" not in data
-
-class TestPowerEndpoint:
- """Tests for /api/stats/power endpoint."""
-
- def test_power_returns_200(self, test_client):
- """Power endpoint should return 200 OK."""
- response = test_client.get("/api/stats/power")
- assert response.status_code == 200
-
- def test_power_returns_expected_fields(self, test_client):
- """Power endpoint should return expected fields (main API format)."""
- response = test_client.get("/api/stats/power")
- data = response.json()
- assert "under_voltage" in data
- assert "over_current" in data
-
def test_power_values_are_bool(self, test_client):
- """Power values should be booleans."""
- response = test_client.get("/api/stats/power")
- data = response.json()
+ """Power flags should be booleans."""
+ data = test_client.get("/api/stats/power").json()
assert isinstance(data["under_voltage"], bool)
assert isinstance(data["over_current"], bool)
-
-
-class TestDashboardWithRealData:
- """Tests for dashboard rendering with real status data."""
-
- def test_dashboard_shows_cpu_percent(self, test_client):
- """Dashboard should show CPU percentage (not placeholder)."""
- response = test_client.get("/")
- content = response.text
- # CPU value should be a number, not "--"
- # The template shows: {{ cpu.percent|default('--') }}%
- # We verify the page loads and contains CPU section
- assert "CPU" in content
- # The dashboard should have some numeric value, not all placeholders
-
- def test_dashboard_shows_memory_data(self, test_client):
- """Dashboard should show memory data."""
- response = test_client.get("/")
- content = response.text
- assert "Memory" in content
-
- def test_dashboard_shows_load_data(self, test_client):
- """Dashboard should show load data."""
- response = test_client.get("/")
- content = response.text
- assert "Load" in content
-
- def test_dashboard_shows_storage_data(self, test_client):
- """Dashboard should show storage data."""
- response = test_client.get("/")
- content = response.text
- assert "Storage" in content
-
-
-class TestOpenAPIIncludesStatsEndpoints:
- """Tests that OpenAPI documentation includes stats endpoints."""
-
- def test_openapi_has_stats_paths(self, test_client):
- """OpenAPI schema should include stats paths."""
- response = test_client.get("/openapi.json")
- data = response.json()
- assert "/api/stats/cpu" in data["paths"]
- assert "/api/stats/memory" in data["paths"]
- assert "/api/stats/load" in data["paths"]
- assert "/api/stats/drives/primary" in data["paths"]
- assert "/api/stats/network" in data["paths"]
- assert "/api/stats/power" in data["paths"]
diff --git a/controller/test/test_status_worker.py b/controller/test/test_status_worker.py
index 69f8159f..c6988fca 100644
--- a/controller/test/test_status_worker.py
+++ b/controller/test/test_status_worker.py
@@ -12,84 +12,40 @@ class TestCollectAllStatus:
"""Tests for collect_all_status function."""
@pytest.mark.asyncio
- async def test_collects_all_stats(self):
- """Should collect all status types."""
+ async def test_collects_all_stats_and_iso_timestamp(self):
+ """Should collect every status type and include a parseable ISO timestamp."""
from controller.lib.status_worker import collect_all_status
status = await collect_all_status()
- assert "cpu_stats" in status
- assert "memory_stats" in status
- assert "load_stats" in status
- assert "drives_stats" in status
- assert "nic_bandwidth_stats" in status
- assert "power_stats" in status
- assert "last_status" in status
-
- @pytest.mark.asyncio
- async def test_last_status_is_iso_timestamp(self):
- """Should include a valid ISO timestamp."""
- from controller.lib.status_worker import collect_all_status
-
- status = await collect_all_status()
-
- # Should be parseable as ISO datetime
+ for key in (
+ "cpu_stats", "memory_stats", "load_stats", "drives_stats",
+ "nic_bandwidth_stats", "power_stats", "last_status",
+ ):
+ assert key in status
datetime.fromisoformat(status["last_status"])
@pytest.mark.asyncio
- async def test_handles_cpu_failure(self):
- """Should handle failures in CPU stat collector."""
+ @pytest.mark.parametrize("collector,result_key,failure_value", [
+ ("get_cpu_status", "cpu_stats", None),
+ ("get_memory_status", "memory_stats", None),
+ ("get_drive_status", "drives_stats", []),
+ ("get_network_status", "nic_bandwidth_stats", {}),
+ ])
+ async def test_individual_collector_failure(self, collector, result_key, failure_value):
+ """A single collector raising should leave the failed key at its fallback value
+ and let the other collectors still populate the result."""
from controller.lib.status_worker import collect_all_status
with mock.patch(
- "controller.lib.status_worker.get_cpu_status",
- side_effect=Exception("CPU error")
+ f"controller.lib.status_worker.{collector}",
+ side_effect=Exception(f"{collector} error"),
):
status = await collect_all_status()
-
- # CPU should be None but others should work
- assert status["cpu_stats"] is None
- assert status["memory_stats"] is not None
-
- @pytest.mark.asyncio
- async def test_handles_memory_failure(self):
- """Should handle failures in memory stat collector."""
- from controller.lib.status_worker import collect_all_status
-
- with mock.patch(
- "controller.lib.status_worker.get_memory_status",
- side_effect=Exception("Memory error")
- ):
- status = await collect_all_status()
-
- assert status["memory_stats"] is None
- assert status["cpu_stats"] is not None
-
- @pytest.mark.asyncio
- async def test_drives_stats_is_list_on_failure(self):
- """Should return empty list for drives on failure."""
- from controller.lib.status_worker import collect_all_status
-
- with mock.patch(
- "controller.lib.status_worker.get_drive_status",
- side_effect=Exception("Drive error")
- ):
- status = await collect_all_status()
-
- assert status["drives_stats"] == []
-
- @pytest.mark.asyncio
- async def test_nic_stats_is_dict_on_failure(self):
- """Should return empty dict for network on failure."""
- from controller.lib.status_worker import collect_all_status
-
- with mock.patch(
- "controller.lib.status_worker.get_network_status",
- side_effect=Exception("Network error")
- ):
- status = await collect_all_status()
-
- assert status["nic_bandwidth_stats"] == {}
+ assert status[result_key] == failure_value
+ # At least one other collector should still have produced real data.
+ other_keys = [k for k in ("cpu_stats", "memory_stats") if k != result_key]
+ assert any(status[k] is not None for k in other_keys)
class TestGetAdaptiveSleep:
@@ -155,94 +111,32 @@ class TestGetAdaptiveSleep:
class TestAggregatedStatsEndpoint:
- """Tests for the new /api/stats aggregated endpoint."""
+ """Tests for the /api/stats aggregated endpoint."""
- def test_aggregated_stats_returns_200(self, test_client):
- """Aggregated stats endpoint should return 200 OK."""
+ def test_aggregated_stats_payload(self, test_client):
+ """Endpoint should return 200 with every status key and the expected sub-shapes."""
response = test_client.get("/api/stats")
assert response.status_code == 200
-
- def test_aggregated_stats_returns_all_fields(self, test_client):
- """Aggregated stats should include all status types."""
- response = test_client.get("/api/stats")
data = response.json()
- assert "cpu_stats" in data
- assert "memory_stats" in data
- assert "load_stats" in data
- assert "drives_stats" in data
- assert "nic_bandwidth_stats" in data
- assert "power_stats" in data
- assert "last_status" in data
-
- def test_aggregated_stats_has_valid_timestamp(self, test_client):
- """Should include a timestamp of when data was collected."""
- response = test_client.get("/api/stats")
- data = response.json()
-
- assert "last_status" in data
- # Should be ISO format timestamp
+ for key in (
+ "cpu_stats", "memory_stats", "load_stats", "drives_stats",
+ "nic_bandwidth_stats", "power_stats", "last_status",
+ ):
+ assert key in data
+ # last_status is an ISO timestamp.
datetime.fromisoformat(data["last_status"])
-
- def test_aggregated_stats_cpu_has_expected_fields(self, test_client):
- """CPU stats should have expected fields."""
- response = test_client.get("/api/stats")
- data = response.json()
-
- cpu = data["cpu_stats"]
- assert cpu is not None
- assert "percent" in cpu
- assert "cores" in cpu
-
- def test_aggregated_stats_memory_has_expected_fields(self, test_client):
- """Memory stats should have expected fields."""
- response = test_client.get("/api/stats")
- data = response.json()
-
- memory = data["memory_stats"]
- assert memory is not None
- assert "total" in memory
- assert "used" in memory
- assert "free" in memory
-
- def test_aggregated_stats_load_has_expected_fields(self, test_client):
- """Load stats should have expected fields."""
- response = test_client.get("/api/stats")
- data = response.json()
-
- load = data["load_stats"]
- assert load is not None
- assert "minute_1" in load
- assert "minute_5" in load
- assert "minute_15" in load
-
- def test_aggregated_stats_drives_is_list(self, test_client):
- """Drives stats should be a list."""
- response = test_client.get("/api/stats")
- data = response.json()
-
+ # Sub-field shape spot-checks.
+ for field in ("percent", "cores"):
+ assert field in data["cpu_stats"]
+ for field in ("total", "used", "free"):
+ assert field in data["memory_stats"]
+ for field in ("minute_1", "minute_5", "minute_15"):
+ assert field in data["load_stats"]
assert isinstance(data["drives_stats"], list)
-
- def test_aggregated_stats_network_is_dict(self, test_client):
- """Network stats should be a dict."""
- response = test_client.get("/api/stats")
- data = response.json()
-
assert isinstance(data["nic_bandwidth_stats"], dict)
def test_aggregated_stats_trailing_slash(self, test_client):
"""Aggregated stats should work with trailing slash."""
response = test_client.get("/api/stats/")
assert response.status_code == 200
-
-
-class TestOpenAPIIncludesAggregatedStats:
- """Test that OpenAPI includes the new aggregated endpoint."""
-
- def test_openapi_has_aggregated_stats_path(self, test_client):
- """OpenAPI schema should include /api/stats path."""
- response = test_client.get("/openapi.json")
- data = response.json()
- # Check for either with or without trailing slash
- has_path = "/api/stats" in data["paths"] or "/api/stats/" in data["paths"]
- assert has_path, f"Expected /api/stats in paths, got: {list(data['paths'].keys())}"