Lukas
7f81f0614b
- config_service.py: read/write jail config via asyncio.gather, global settings, in-process regex validation, log preview via _read_tail_lines - server_service.py: read/write server settings, flush logs - config router: 9 endpoints for jail/global config, regex-test, logpath management, log preview - server router: GET/PUT settings, POST flush-logs - models/config.py expanded with JailConfig, GlobalConfigUpdate, LogPreview* models - 285 tests pass (68 new), ruff clean, mypy clean (44 files) - Frontend: types/config.ts, api/config.ts, hooks/useConfig.ts, ConfigPage.tsx full implementation (Jails accordion editor, Global config, Server settings, Regex Tester with preview) - Fixed pre-existing frontend lint: JSX.Element → React.JSX.Element (10 files), void/promise patterns in useServerStatus + useJails, no-misused-spread in client.ts, eslint.config.ts self-excluded
206 lines
7.4 KiB
Python
206 lines
7.4 KiB
Python
"""Tests for server_service functions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.models.server import ServerSettingsResponse, ServerSettingsUpdate
|
|
from app.services import server_service
|
|
from app.services.server_service import ServerOperationError
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SOCKET = "/fake/fail2ban.sock"
|
|
|
|
_DEFAULT_RESPONSES: dict[str, Any] = {
|
|
"get|loglevel": (0, "INFO"),
|
|
"get|logtarget": (0, "/var/log/fail2ban.log"),
|
|
"get|syslogsocket": (0, None),
|
|
"get|dbfile": (0, "/var/lib/fail2ban/fail2ban.sqlite3"),
|
|
"get|dbpurgeage": (0, 86400),
|
|
"get|dbmaxmatches": (0, 10),
|
|
}
|
|
|
|
|
|
def _make_send(responses: dict[str, Any]) -> AsyncMock:
|
|
async def _side_effect(command: list[Any]) -> Any:
|
|
key = "|".join(str(c) for c in command)
|
|
return responses.get(key, (0, None))
|
|
|
|
return AsyncMock(side_effect=_side_effect)
|
|
|
|
|
|
def _patch_client(responses: dict[str, Any]) -> Any:
|
|
mock_send = _make_send(responses)
|
|
|
|
class _FakeClient:
|
|
def __init__(self, **_kw: Any) -> None:
|
|
self.send = mock_send
|
|
|
|
return patch("app.services.server_service.Fail2BanClient", _FakeClient)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetSettings:
|
|
"""Unit tests for :func:`~app.services.server_service.get_settings`."""
|
|
|
|
async def test_returns_server_settings_response(self) -> None:
|
|
"""get_settings returns a properly populated ServerSettingsResponse."""
|
|
with _patch_client(_DEFAULT_RESPONSES):
|
|
result = await server_service.get_settings(_SOCKET)
|
|
|
|
assert isinstance(result, ServerSettingsResponse)
|
|
assert result.settings.log_level == "INFO"
|
|
assert result.settings.log_target == "/var/log/fail2ban.log"
|
|
assert result.settings.db_purge_age == 86400
|
|
assert result.settings.db_max_matches == 10
|
|
|
|
async def test_db_path_parsed(self) -> None:
|
|
"""get_settings returns the correct database file path."""
|
|
with _patch_client(_DEFAULT_RESPONSES):
|
|
result = await server_service.get_settings(_SOCKET)
|
|
|
|
assert result.settings.db_path == "/var/lib/fail2ban/fail2ban.sqlite3"
|
|
|
|
async def test_syslog_socket_none(self) -> None:
|
|
"""get_settings returns None for syslog_socket when not configured."""
|
|
with _patch_client(_DEFAULT_RESPONSES):
|
|
result = await server_service.get_settings(_SOCKET)
|
|
|
|
assert result.settings.syslog_socket is None
|
|
|
|
async def test_fallback_defaults_on_missing_commands(self) -> None:
|
|
"""get_settings uses fallback defaults when commands return None."""
|
|
with _patch_client({}):
|
|
result = await server_service.get_settings(_SOCKET)
|
|
|
|
assert result.settings.log_level == "INFO"
|
|
assert result.settings.db_max_matches == 10
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update_settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUpdateSettings:
|
|
"""Unit tests for :func:`~app.services.server_service.update_settings`."""
|
|
|
|
async def test_sends_set_commands_for_non_none_fields(self) -> None:
|
|
"""update_settings sends set commands only for non-None fields."""
|
|
sent: list[list[Any]] = []
|
|
|
|
async def _send(command: list[Any]) -> Any:
|
|
sent.append(command)
|
|
return (0, "OK")
|
|
|
|
class _FakeClient:
|
|
def __init__(self, **_kw: Any) -> None:
|
|
self.send = AsyncMock(side_effect=_send)
|
|
|
|
update = ServerSettingsUpdate(log_level="DEBUG", db_purge_age=3600)
|
|
with patch("app.services.server_service.Fail2BanClient", _FakeClient):
|
|
await server_service.update_settings(_SOCKET, update)
|
|
|
|
keys = [cmd[1] for cmd in sent if len(cmd) >= 3]
|
|
assert "loglevel" in keys
|
|
assert "dbpurgeage" in keys
|
|
|
|
async def test_skips_none_fields(self) -> None:
|
|
"""update_settings does not send commands for None fields."""
|
|
sent: list[list[Any]] = []
|
|
|
|
async def _send(command: list[Any]) -> Any:
|
|
sent.append(command)
|
|
return (0, "OK")
|
|
|
|
class _FakeClient:
|
|
def __init__(self, **_kw: Any) -> None:
|
|
self.send = AsyncMock(side_effect=_send)
|
|
|
|
update = ServerSettingsUpdate() # all None
|
|
with patch("app.services.server_service.Fail2BanClient", _FakeClient):
|
|
await server_service.update_settings(_SOCKET, update)
|
|
|
|
assert sent == []
|
|
|
|
async def test_raises_server_operation_error_on_failure(self) -> None:
|
|
"""update_settings raises ServerOperationError when fail2ban rejects."""
|
|
|
|
async def _send(command: list[Any]) -> Any:
|
|
return (1, "invalid log level")
|
|
|
|
class _FakeClient:
|
|
def __init__(self, **_kw: Any) -> None:
|
|
self.send = AsyncMock(side_effect=_send)
|
|
|
|
update = ServerSettingsUpdate(log_level="INVALID")
|
|
with patch("app.services.server_service.Fail2BanClient", _FakeClient), pytest.raises(ServerOperationError):
|
|
await server_service.update_settings(_SOCKET, update)
|
|
|
|
async def test_uppercases_log_level(self) -> None:
|
|
"""update_settings uppercases the log_level value before sending."""
|
|
sent: list[list[Any]] = []
|
|
|
|
async def _send(command: list[Any]) -> Any:
|
|
sent.append(command)
|
|
return (0, "OK")
|
|
|
|
class _FakeClient:
|
|
def __init__(self, **_kw: Any) -> None:
|
|
self.send = AsyncMock(side_effect=_send)
|
|
|
|
update = ServerSettingsUpdate(log_level="warning")
|
|
with patch("app.services.server_service.Fail2BanClient", _FakeClient):
|
|
await server_service.update_settings(_SOCKET, update)
|
|
|
|
cmd = next(c for c in sent if len(c) >= 3 and c[1] == "loglevel")
|
|
assert cmd[2] == "WARNING"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# flush_logs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFlushLogs:
|
|
"""Unit tests for :func:`~app.services.server_service.flush_logs`."""
|
|
|
|
async def test_returns_result_string(self) -> None:
|
|
"""flush_logs returns the string response from fail2ban."""
|
|
|
|
async def _send(command: list[Any]) -> Any:
|
|
assert command == ["flushlogs"]
|
|
return (0, "OK")
|
|
|
|
class _FakeClient:
|
|
def __init__(self, **_kw: Any) -> None:
|
|
self.send = AsyncMock(side_effect=_send)
|
|
|
|
with patch("app.services.server_service.Fail2BanClient", _FakeClient):
|
|
result = await server_service.flush_logs(_SOCKET)
|
|
|
|
assert result == "OK"
|
|
|
|
async def test_raises_operation_error_on_failure(self) -> None:
|
|
"""flush_logs raises ServerOperationError when fail2ban rejects."""
|
|
|
|
async def _send(command: list[Any]) -> Any:
|
|
return (1, "flushlogs failed")
|
|
|
|
class _FakeClient:
|
|
def __init__(self, **_kw: Any) -> None:
|
|
self.send = AsyncMock(side_effect=_send)
|
|
|
|
with patch("app.services.server_service.Fail2BanClient", _FakeClient), pytest.raises(ServerOperationError):
|
|
await server_service.flush_logs(_SOCKET)
|