268 lines
9.7 KiB
Python
268 lines
9.7 KiB
Python
"""Tests for the server settings router endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.config import Settings
|
|
from app.db import init_db
|
|
from app.main import create_app
|
|
from app.models.server import ServerSettings, ServerSettingsResponse
|
|
from app.services.geo_cache import GeoCache
|
|
from app.utils.session_cache import NoOpSessionCache
|
|
from app.utils.setup_state import set_setup_complete_cache
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SETUP_PAYLOAD = {
|
|
"master_password": "Testpass1!",
|
|
"database_path": "bangui.db",
|
|
"fail2ban_socket": "/var/run/fail2ban/fail2ban.sock",
|
|
"timezone": "UTC",
|
|
"session_duration_minutes": 60,
|
|
}
|
|
|
|
|
|
async def _write_password_hash(db: aiosqlite.Connection, password: str) -> str:
|
|
"""Hash password and write to settings table."""
|
|
import asyncio
|
|
|
|
import bcrypt
|
|
|
|
pw_bytes = password.encode()
|
|
hashed = await asyncio.get_event_loop().run_in_executor(
|
|
None, lambda: bcrypt.hashpw(pw_bytes, bcrypt.gensalt()).decode()
|
|
)
|
|
await db.execute(
|
|
"INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)",
|
|
("master_password_hash", hashed),
|
|
)
|
|
await db.commit()
|
|
return hashed
|
|
|
|
|
|
@pytest.fixture
|
|
async def server_client(tmp_path: Path) -> AsyncClient: # type: ignore[misc]
|
|
"""Provide an authenticated ``AsyncClient`` for server endpoint tests."""
|
|
import os
|
|
|
|
os.makedirs(tmp_path / "fail2ban", exist_ok=True)
|
|
settings = Settings(
|
|
database_path=str(tmp_path / "server_test.db"),
|
|
fail2ban_socket="/tmp/fake.sock",
|
|
fail2ban_config_dir=str(tmp_path / "fail2ban"),
|
|
session_secret="test-server-secret-0000000000000000000000",
|
|
session_duration_minutes=60,
|
|
timezone="UTC",
|
|
log_level="debug",
|
|
session_cookie_secure=False,
|
|
)
|
|
app = create_app(settings=settings)
|
|
set_setup_complete_cache(app, True)
|
|
|
|
db: aiosqlite.Connection = await aiosqlite.connect(settings.database_path)
|
|
db.row_factory = aiosqlite.Row
|
|
await init_db(db)
|
|
await _write_password_hash(db, _SETUP_PAYLOAD["master_password"])
|
|
app.state.db = db
|
|
app.state.http_session = MagicMock()
|
|
app.state.session_cache = NoOpSessionCache()
|
|
app.state.geo_cache = GeoCache()
|
|
|
|
async def _override_get_db():
|
|
yield db
|
|
|
|
from app.dependencies import get_db, get_session_cache
|
|
|
|
app.dependency_overrides[get_db] = _override_get_db
|
|
app.dependency_overrides[get_session_cache] = lambda: NoOpSessionCache()
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test", headers={"X-BanGUI-Request": "1"}) as ac:
|
|
login = await ac.post(
|
|
"/api/v1/auth/login",
|
|
json={"password": _SETUP_PAYLOAD["master_password"]},
|
|
)
|
|
assert login.status_code == 200
|
|
yield ac
|
|
|
|
await db.close()
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def _make_settings() -> ServerSettingsResponse:
|
|
return ServerSettingsResponse(
|
|
settings=ServerSettings(
|
|
log_level="INFO",
|
|
log_target="/var/log/fail2ban.log",
|
|
syslog_socket=None,
|
|
db_path="/var/lib/fail2ban/fail2ban.sqlite3",
|
|
db_purge_age=86400,
|
|
db_max_matches=10,
|
|
),
|
|
warnings={"db_purge_age_too_low": False},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/server/settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetServerSettings:
|
|
"""Tests for ``GET /api/server/settings``."""
|
|
|
|
async def test_200_returns_settings(self, server_client: AsyncClient) -> None:
|
|
"""GET /api/server/settings returns 200 with ServerSettingsResponse."""
|
|
mock_response = _make_settings()
|
|
with patch(
|
|
"app.routers.server.server_service.get_settings",
|
|
AsyncMock(return_value=mock_response),
|
|
):
|
|
resp = await server_client.get("/api/v1/server/settings")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["settings"]["log_level"] == "INFO"
|
|
assert data["settings"]["db_purge_age"] == 86400
|
|
assert data["warnings"]["db_purge_age_too_low"] is False
|
|
|
|
async def test_401_when_unauthenticated(self, server_client: AsyncClient) -> None:
|
|
"""GET /api/server/settings returns 401 without session."""
|
|
resp = await AsyncClient(
|
|
transport=ASGITransport(app=server_client._transport.app), # type: ignore[attr-defined]
|
|
base_url="http://test",
|
|
).get("/api/v1/server/settings")
|
|
assert resp.status_code == 401
|
|
|
|
async def test_502_on_connection_error(self, server_client: AsyncClient) -> None:
|
|
"""GET /api/server/settings returns 502 when fail2ban is unreachable."""
|
|
from app.exceptions import Fail2BanConnectionError
|
|
|
|
with patch(
|
|
"app.routers.server.server_service.get_settings",
|
|
AsyncMock(side_effect=Fail2BanConnectionError("down", "/tmp/fake.sock")),
|
|
):
|
|
resp = await server_client.get("/api/v1/server/settings")
|
|
|
|
assert resp.status_code == 502
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PUT /api/server/settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUpdateServerSettings:
|
|
"""Tests for ``PUT /api/server/settings``."""
|
|
|
|
async def test_204_on_success(self, server_client: AsyncClient) -> None:
|
|
"""PUT /api/server/settings returns 204 on success."""
|
|
with patch(
|
|
"app.routers.server.server_service.update_settings",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
resp = await server_client.put(
|
|
"/api/v1/server/settings",
|
|
json={"log_level": "DEBUG"},
|
|
)
|
|
|
|
assert resp.status_code == 204
|
|
|
|
async def test_400_on_operation_error(self, server_client: AsyncClient) -> None:
|
|
"""PUT /api/server/settings returns 400 when set command fails."""
|
|
from app.services.server_service import ServerOperationError
|
|
|
|
with patch(
|
|
"app.routers.server.server_service.update_settings",
|
|
AsyncMock(side_effect=ServerOperationError("set failed")),
|
|
):
|
|
resp = await server_client.put(
|
|
"/api/v1/server/settings",
|
|
json={"log_level": "DEBUG"},
|
|
)
|
|
|
|
assert resp.status_code == 400
|
|
|
|
async def test_401_when_unauthenticated(self, server_client: AsyncClient) -> None:
|
|
"""PUT /api/server/settings returns 401 without session."""
|
|
resp = await AsyncClient(
|
|
transport=ASGITransport(app=server_client._transport.app), # type: ignore[attr-defined]
|
|
base_url="http://test",
|
|
).put("/api/v1/server/settings", json={"log_level": "DEBUG"})
|
|
assert resp.status_code == 401
|
|
|
|
async def test_502_on_connection_error(self, server_client: AsyncClient) -> None:
|
|
"""PUT /api/server/settings returns 502 when fail2ban is unreachable."""
|
|
from app.exceptions import Fail2BanConnectionError
|
|
|
|
with patch(
|
|
"app.routers.server.server_service.update_settings",
|
|
AsyncMock(side_effect=Fail2BanConnectionError("down", "/tmp/fake.sock")),
|
|
):
|
|
resp = await server_client.put(
|
|
"/api/v1/server/settings",
|
|
json={"log_level": "INFO"},
|
|
)
|
|
|
|
assert resp.status_code == 502
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/server/flush-logs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFlushLogs:
|
|
"""Tests for ``POST /api/server/flush-logs``."""
|
|
|
|
async def test_200_returns_message(self, server_client: AsyncClient) -> None:
|
|
"""POST /api/server/flush-logs returns 200 with a message."""
|
|
with patch(
|
|
"app.routers.server.server_service.flush_logs",
|
|
AsyncMock(return_value="OK"),
|
|
):
|
|
resp = await server_client.post("/api/v1/server/flush-logs")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "OK"
|
|
|
|
async def test_400_on_operation_error(self, server_client: AsyncClient) -> None:
|
|
"""POST /api/server/flush-logs returns 400 when flushlogs fails."""
|
|
from app.services.server_service import ServerOperationError
|
|
|
|
with patch(
|
|
"app.routers.server.server_service.flush_logs",
|
|
AsyncMock(side_effect=ServerOperationError("flushlogs failed")),
|
|
):
|
|
resp = await server_client.post("/api/v1/server/flush-logs")
|
|
|
|
assert resp.status_code == 400
|
|
|
|
async def test_401_when_unauthenticated(self, server_client: AsyncClient) -> None:
|
|
"""POST /api/server/flush-logs returns 401 without session."""
|
|
resp = await AsyncClient(
|
|
transport=ASGITransport(app=server_client._transport.app), # type: ignore[attr-defined]
|
|
base_url="http://test",
|
|
).post("/api/v1/server/flush-logs")
|
|
assert resp.status_code == 401
|
|
|
|
async def test_502_on_connection_error(self, server_client: AsyncClient) -> None:
|
|
"""POST /api/server/flush-logs returns 502 when fail2ban is unreachable."""
|
|
from app.exceptions import Fail2BanConnectionError
|
|
|
|
with patch(
|
|
"app.routers.server.server_service.flush_logs",
|
|
AsyncMock(side_effect=Fail2BanConnectionError("down", "/tmp/fake.sock")),
|
|
):
|
|
resp = await server_client.post("/api/v1/server/flush-logs")
|
|
|
|
assert resp.status_code == 502
|