Add fail2ban DB index management and socket-based path resolution

- New get_fail2ban_db_path() in setup_service resolves DB path from configured socket path
- New ensure_fail2ban_indexes() creates missing performance indexes on bans table
- Call ensure_fail2ban_indexes on every startup before first ban query
- Remove completed tasks from Docs/Tasks.md
- Update Docs/PERFORMANCE.md with index findings
This commit is contained in:
Copilot
2026-05-03 12:17:31 +02:00
committed by Lukas
parent 0133489920
commit 22db607875
6 changed files with 189 additions and 50 deletions

View File

@@ -1,6 +1,60 @@
"""Tests for fail2ban_db_utils module."""
from app.utils.fail2ban_db_utils import escape_like
import sqlite3
from pathlib import Path
import pytest
from app.utils.fail2ban_db_utils import (
ensure_fail2ban_indexes,
escape_like,
)
@pytest.fixture
def tmp_bans_table(tmp_path: Path) -> str:
"""Create a minimal fail2ban-style database with bans table."""
db_path = str(tmp_path / "test_f2b.db")
conn = sqlite3.connect(db_path)
conn.execute("CREATE TABLE bans (jail, ip, timeofban, bancount, data)")
conn.execute("CREATE INDEX idx_jail_timeofban_ip ON bans(jail, timeofban)")
conn.execute("CREATE INDEX idx_jail_ip ON bans(jail, ip)")
conn.execute("CREATE INDEX idx_ip ON bans(ip)")
conn.commit()
conn.close()
return db_path
@pytest.mark.asyncio
async def test_ensure_fail2ban_indexes_creates_missing_index(tmp_bans_table: str) -> None:
"""Index is created when idx_bans_timeofban_desc does not exist."""
await ensure_fail2ban_indexes(tmp_bans_table)
conn = sqlite3.connect(tmp_bans_table)
conn.row_factory = sqlite3.Row
cur = conn.execute(
"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='bans'"
)
index_names = [str(r["name"]) for r in cur.fetchall()]
conn.close()
assert "idx_bans_timeofban_desc" in index_names
@pytest.mark.asyncio
async def test_ensure_fail2ban_indexes_idempotent(tmp_bans_table: str) -> None:
"""Calling twice does not raise or duplicate the index."""
await ensure_fail2ban_indexes(tmp_bans_table)
await ensure_fail2ban_indexes(tmp_bans_table)
conn = sqlite3.connect(tmp_bans_table)
cur = conn.execute(
"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='bans' AND name='idx_bans_timeofban_desc'"
)
count = len(cur.fetchall())
conn.close()
assert count == 1
def test_escape_like_percent_sign() -> None: