mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-25 18:40:06 -04:00
Consolidating controller tests. Fixing controller test cleanup which slowed down tests significantly.
This commit is contained in:
Roland
parent
b0be30c3d6
commit
2806673e25
11 changed files with 366 additions and 970 deletions
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
"<th>Persist</th>", "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 "<th>Boot</th>" 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 "<th>Boot</th>" 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 <tbody> 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'<tbody>(.*?)</tbody>', 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 "<th>Persist</th>" 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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
|
|
@ -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())}"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue