Gan, Jimmy
258f8fd2f7
Deploy Dashboard (Main) / Backend Tests (push) Successful in 3m26s
Deploy Dashboard (Main) / Frontend Tests (push) Successful in 7s
Deploy Dashboard (Main) / Build Main Image (push) Successful in 4m52s
Deploy Dashboard (Main) / Deploy to Production (push) Successful in 1m0s
50 lines
2.3 KiB
Python
50 lines
2.3 KiB
Python
import pytest
|
|
from unittest.mock import patch, AsyncMock
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_health(test_app, admin_headers):
|
|
with patch("routers.media_mgmt.run_host_command", new_callable=AsyncMock) as mock_run:
|
|
mock_run.return_value = (0, "All checks passed\n", "")
|
|
response = test_app.get("/api/media/health", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
assert data["exit_code"] == 0
|
|
assert "All checks passed" in data["stdout"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_storage(test_app, admin_headers):
|
|
with patch("routers.media_mgmt.run_host_command", new_callable=AsyncMock) as mock_run:
|
|
mock_run.return_value = (0, '{"timestamp": "2026-07-11T05:37:57Z", "volume": {"path": "/volume1", "total": "14T", "used": "12T", "available": "2.2T", "used_pct": 85}}', "")
|
|
response = test_app.get("/api/media/storage", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "volume" in data
|
|
assert data["volume"]["total"] == "14T"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan(test_app, admin_headers):
|
|
with patch("routers.media_mgmt.run_host_command", new_callable=AsyncMock) as mock_run:
|
|
mock_run.return_value = (0, "", "")
|
|
response = test_app.post("/api/media/maintenance/scan?library=DJI-Action6", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "Scan started" in data["message"]
|
|
mock_run.assert_called_once()
|
|
assert "scan-library.sh DJI-Action6" in mock_run.call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_active_status(test_app, admin_headers):
|
|
with patch("routers.media_mgmt.run_host_command", new_callable=AsyncMock) as mock_run:
|
|
def side_effect(cmd):
|
|
if "scan-library.sh" in cmd:
|
|
return (0, "1234\n", "")
|
|
return (1, "", "")
|
|
mock_run.side_effect = side_effect
|
|
response = test_app.get("/api/media/status", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["scan"]["running"] is True
|
|
assert data["scan"]["pid"] == "1234"
|
|
assert data["requeue"]["running"] is False
|