273 lines
9.4 KiB
Python
273 lines
9.4 KiB
Python
"""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
|