lukas.pupkalipinski/BanGUI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixed tests

Browse Source
This commit is contained in:
Lukas
2026-05-15 20:41:05 +02:00
parent 96ce516ecf
commit 77df5d5d65
50 changed files with 1482 additions and 5089 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
backend/tests/test_routers/test_bans.py
View File

@@ -7,25 +7,34 @@ from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import aiosqlite
import bcrypt
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.ban import ActiveBan, ActiveBanListResponse
from app.exceptions import Fail2BanConnectionError
from app.main import create_app
from app.models.ban_domain import DomainActiveBan, DomainActiveBanList
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!",
"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."""
pw_bytes = password.encode()
import asyncio
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
@@ -41,24 +50,30 @@ async def bans_client(tmp_path: Path) -> AsyncClient: # type: ignore[misc]
log_level="debug",
fail2ban_config_dir=str(tmp_path / "fail2ban"),
session_cache_enabled=False,
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() -> AsyncGenerator[aiosqlite.Connection, None]:
yield db
from app.dependencies import get_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") as ac:
await ac.post("/api/v1/setup", json=_SETUP_PAYLOAD)
login = await ac.post(
"/api/v1/auth/login",
json={"password": _SETUP_PAYLOAD["master_password"]},
@@ -70,6 +85,19 @@ async def bans_client(tmp_path: Path) -> AsyncClient: # type: ignore[misc]
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
_SETUP_PAYLOAD = {
"master_password": "Testpass1!",
"fail2ban_socket": "/var/run/fail2ban/fail2ban.sock",
"timezone": "UTC",
"session_duration_minutes": 60,
"database_path": "bans_test.db",
}
# ---------------------------------------------------------------------------
# GET /api/bans/active
# ---------------------------------------------------------------------------
@@ -80,9 +108,11 @@ class TestGetActiveBans:
async def test_200_when_authenticated(self, bans_client: AsyncClient) -> None:
"""GET /api/bans/active returns 200 with an ActiveBanListResponse."""
mock_response = ActiveBanListResponse(
from app.models.ban_domain import DomainActiveBan, DomainActiveBanList
mock_response = DomainActiveBanList(
bans=[
ActiveBan(
DomainActiveBan(
ip="1.2.3.4",
jail="sshd",
banned_at="2025-01-01T12:00:00+00:00",
@@ -102,20 +132,21 @@ class TestGetActiveBans:
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 1
assert data["bans"][0]["ip"] == "1.2.3.4"
assert data["bans"][0]["jail"] == "sshd"
assert data["items"][0]["ip"] == "1.2.3.4"
assert data["items"][0]["jail"] == "sshd"
async def test_401_when_unauthenticated(
self, bans_client: AsyncClient, monkeypatch: pytest.MonkeyPatch
) -> None:
async def test_401_when_unauthenticated(self, bans_client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None:
"""GET /api/bans/active returns 401 without session."""
import logging
from unittest.mock import MagicMock
class FakeLogger:
def error(self, *args, **kwargs): pass
def warning(self, *args, **kwargs): pass
def info(self, *args, **kwargs): pass
def error(self, *args, **kwargs):
pass
def warning(self, *args, **kwargs):
pass
def info(self, *args, **kwargs):
pass
monkeypatch.setattr("app.main.log", FakeLogger())
resp = await AsyncClient(
@@ -126,7 +157,7 @@ class TestGetActiveBans:
async def test_empty_when_no_bans(self, bans_client: AsyncClient) -> None:
"""GET /api/bans/active returns empty list when no bans are active."""
mock_response = ActiveBanListResponse(bans=[], total=0)
mock_response = DomainActiveBanList(bans=[], total=0)
with patch(
"app.routers.bans.ban_service.get_active_bans",
AsyncMock(return_value=mock_response),
@@ -135,13 +166,13 @@ class TestGetActiveBans:
assert resp.status_code == 200
assert resp.json()["total"] == 0
assert resp.json()["bans"] == []
assert resp.json()["items"] == []
async def test_response_shape(self, bans_client: AsyncClient) -> None:
"""GET /api/bans/active returns expected fields per ban entry."""
mock_response = ActiveBanListResponse(
mock_response = DomainActiveBanList(
bans=[
ActiveBan(
DomainActiveBan(
ip="10.0.0.1",
jail="nginx",
banned_at=None,
@@ -158,7 +189,7 @@ class TestGetActiveBans:
):
resp = await bans_client.get("/api/v1/bans/active")
ban = resp.json()["bans"][0]
ban = resp.json()["items"][0]
assert "ip" in ban
assert "jail" in ban
assert "banned_at" in ban
@@ -183,6 +214,7 @@ class TestBanIp:
resp = await bans_client.post(
"/api/v1/bans",
json={"ip": "1.2.3.4", "jail": "sshd"},
headers={"X-BanGUI-Request": "1"},
)
assert resp.status_code == 201
@@ -197,6 +229,7 @@ class TestBanIp:
resp = await bans_client.post(
"/api/v1/bans",
json={"ip": "bad", "jail": "sshd"},
headers={"X-BanGUI-Request": "1"},
)
assert resp.status_code == 400
@@ -212,6 +245,7 @@ class TestBanIp:
resp = await bans_client.post(
"/api/v1/bans",
json={"ip": "1.2.3.4", "jail": "ghost"},
headers={"X-BanGUI-Request": "1"},
)
assert resp.status_code == 404
@@ -243,6 +277,7 @@ class TestUnbanIp:
"DELETE",
"/api/v1/bans",
json={"ip": "1.2.3.4", "unban_all": True},
headers={"X-BanGUI-Request": "1"},
)
assert resp.status_code == 200
@@ -258,6 +293,7 @@ class TestUnbanIp:
"DELETE",
"/api/v1/bans",
json={"ip": "1.2.3.4", "jail": "sshd"},
headers={"X-BanGUI-Request": "1"},
)
assert resp.status_code == 200
@@ -273,6 +309,7 @@ class TestUnbanIp:
"DELETE",
"/api/v1/bans",
json={"ip": "bad", "unban_all": True},
headers={"X-BanGUI-Request": "1"},
)
assert resp.status_code == 400
@@ -289,6 +326,7 @@ class TestUnbanIp:
"DELETE",
"/api/v1/bans",
json={"ip": "1.2.3.4", "jail": "ghost"},
headers={"X-BanGUI-Request": "1"},
)
assert resp.status_code == 404
@@ -308,7 +346,7 @@ class TestUnbanAll:
"app.routers.bans.jail_service.unban_all_ips",
AsyncMock(return_value=3),
):
resp = await bans_client.request("DELETE", "/api/v1/bans/all")
resp = await bans_client.request("DELETE", "/api/v1/bans/all", headers={"X-BanGUI-Request": "1"})
assert resp.status_code == 200
data = resp.json()
@@ -321,14 +359,12 @@ class TestUnbanAll:
"app.routers.bans.jail_service.unban_all_ips",
AsyncMock(return_value=0),
):
resp = await bans_client.request("DELETE", "/api/v1/bans/all")
resp = await bans_client.request("DELETE", "/api/v1/bans/all", headers={"X-BanGUI-Request": "1"})
assert resp.status_code == 200
assert resp.json()["count"] == 0
async def test_502_when_fail2ban_unreachable(
self, bans_client: AsyncClient
) -> None:
async def test_502_when_fail2ban_unreachable(self, bans_client: AsyncClient) -> None:
"""DELETE /api/bans/all returns 502 when fail2ban is unreachable."""
with patch(
"app.routers.bans.jail_service.unban_all_ips",
@@ -339,7 +375,7 @@ class TestUnbanAll:
)
),
):
resp = await bans_client.request("DELETE", "/api/v1/bans/all")
resp = await bans_client.request("DELETE", "/api/v1/bans/all", headers={"X-BanGUI-Request": "1"})
assert resp.status_code == 502