fix: config modal scrollbar, scheduler-config.js, logging API endpoint, static cache-busting
This commit is contained in:
272
tests/api/test_logging_endpoints.py
Normal file
272
tests/api/test_logging_endpoints.py
Normal file
@@ -0,0 +1,272 @@
|
||||
"""Tests for /api/logging endpoints.
|
||||
|
||||
Covers:
|
||||
- GET /api/logging/config — returns correct shape with log_level and flags
|
||||
- GET /api/logging/files — lists log files
|
||||
- POST /api/logging/test — test log trigger
|
||||
- POST /api/logging/cleanup — file cleanup
|
||||
- GET /api/logging/files/{name}/tail — tail endpoint
|
||||
|
||||
These tests guard against regressions of the 404 that previously caused the
|
||||
config modal to fail on load.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from src.server.fastapi_app import app
|
||||
from src.server.models.config import AppConfig, LoggingConfig
|
||||
from src.server.services.auth_service import auth_service
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_auth_state():
|
||||
"""Reset auth service state before each test."""
|
||||
auth_service._hash = None
|
||||
auth_service._failed.clear()
|
||||
if not auth_service.is_configured():
|
||||
auth_service.setup_master_password("TestPass123!")
|
||||
yield
|
||||
auth_service._hash = None
|
||||
auth_service._failed.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||||
yield ac
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_client(client):
|
||||
"""Authenticated async client."""
|
||||
response = await client.post("/api/auth/login", json={"password": "TestPass123!"})
|
||||
assert response.status_code == 200, response.text
|
||||
token = response.json()["access_token"]
|
||||
client.headers.update({"Authorization": f"Bearer {token}"})
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_service():
|
||||
"""Return a mock config service with minimal logging config."""
|
||||
service = Mock()
|
||||
app_cfg = Mock(spec=AppConfig)
|
||||
app_cfg.logging = LoggingConfig(
|
||||
level="INFO",
|
||||
file=None,
|
||||
max_bytes=None,
|
||||
backup_count=3,
|
||||
)
|
||||
service.load_config = Mock(return_value=app_cfg)
|
||||
return service
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/logging/config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetLoggingConfig:
|
||||
"""GET /api/logging/config returns the expected shape."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_success_shape(self, auth_client, mock_config_service):
|
||||
with patch(
|
||||
"src.server.api.logging.get_config_service",
|
||||
return_value=mock_config_service,
|
||||
):
|
||||
response = await auth_client.get("/api/logging/config")
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["success"] is True
|
||||
assert "config" in body
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_config_contains_log_level(self, auth_client, mock_config_service):
|
||||
with patch(
|
||||
"src.server.api.logging.get_config_service",
|
||||
return_value=mock_config_service,
|
||||
):
|
||||
response = await auth_client.get("/api/logging/config")
|
||||
|
||||
config = response.json()["config"]
|
||||
assert config["log_level"] == "INFO"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_config_contains_boolean_flags(self, auth_client, mock_config_service):
|
||||
with patch(
|
||||
"src.server.api.logging.get_config_service",
|
||||
return_value=mock_config_service,
|
||||
):
|
||||
response = await auth_client.get("/api/logging/config")
|
||||
|
||||
config = response.json()["config"]
|
||||
assert "enable_console_logging" in config
|
||||
assert "enable_fail2ban_logging" in config
|
||||
assert isinstance(config["enable_console_logging"], bool)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_requires_authentication(self, client):
|
||||
response = await client.get("/api/logging/config")
|
||||
assert response.status_code in (401, 403)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/logging/files
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestListLogFiles:
|
||||
"""GET /api/logging/files lists available log files."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_success_and_files_list(self, auth_client, tmp_path, monkeypatch):
|
||||
# Create a fake log file
|
||||
(tmp_path / "server.log").write_text("line1\nline2\n")
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.get("/api/logging/files")
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["success"] is True
|
||||
assert isinstance(body["files"], list)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_metadata_shape(self, auth_client, tmp_path, monkeypatch):
|
||||
(tmp_path / "app.log").write_text("test content")
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.get("/api/logging/files")
|
||||
files = response.json()["files"]
|
||||
|
||||
assert len(files) >= 1
|
||||
f = files[0]
|
||||
assert "name" in f
|
||||
assert "size_mb" in f
|
||||
assert "modified" in f
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_log_dir(self, auth_client, tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.get("/api/logging/files")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["files"] == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/logging/files/{filename}/tail
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTailLogFile:
|
||||
"""GET /api/logging/files/{filename}/tail returns log lines."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_last_n_lines(self, auth_client, tmp_path, monkeypatch):
|
||||
log_content = "\n".join(f"line {i}" for i in range(200))
|
||||
(tmp_path / "server.log").write_text(log_content)
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.get("/api/logging/files/server.log/tail?lines=50")
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["success"] is True
|
||||
assert len(body["lines"]) == 50
|
||||
assert body["showing_lines"] == 50
|
||||
assert body["total_lines"] == 200
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_404_for_missing_file(self, auth_client, tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.get("/api/logging/files/nonexistent.log/tail")
|
||||
assert response.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prevents_path_traversal(self, auth_client, tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.get(
|
||||
"/api/logging/files/..%2F..%2Fetc%2Fpasswd/tail"
|
||||
)
|
||||
# Either 404 (file not found) or 400 is acceptable — must NOT return 200
|
||||
assert response.status_code != 200
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/logging/test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTestLogging:
|
||||
"""POST /api/logging/test writes test log messages."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_success(self, auth_client):
|
||||
response = await auth_client.post("/api/logging/test")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["success"] is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_requires_authentication(self, client):
|
||||
response = await client.post("/api/logging/test")
|
||||
assert response.status_code in (401, 403)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/logging/cleanup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCleanupLogs:
|
||||
"""POST /api/logging/cleanup removes old log files."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_removes_old_files(self, auth_client, tmp_path, monkeypatch):
|
||||
old_file = tmp_path / "old.log"
|
||||
old_file.write_text("old content")
|
||||
# Make it look very old
|
||||
old_ts = time.time() - (40 * 86400)
|
||||
import os
|
||||
os.utime(old_file, (old_ts, old_ts))
|
||||
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.post("/api/logging/cleanup", json={"days": 30})
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["success"] is True
|
||||
assert "old.log" in body["removed"]
|
||||
assert not old_file.exists()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_keeps_recent_files(self, auth_client, tmp_path, monkeypatch):
|
||||
new_file = tmp_path / "new.log"
|
||||
new_file.write_text("recent content")
|
||||
|
||||
monkeypatch.setattr("src.server.api.logging._LOG_DIR", tmp_path)
|
||||
|
||||
response = await auth_client.post("/api/logging/cleanup", json={"days": 30})
|
||||
assert response.status_code == 200
|
||||
assert "new.log" not in response.json()["removed"]
|
||||
assert new_file.exists()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_days_returns_400(self, auth_client):
|
||||
response = await auth_client.post("/api/logging/cleanup", json={"days": 0})
|
||||
assert response.status_code == 400
|
||||
Reference in New Issue
Block a user