Complete history archive support for dashboard/map data and mark task finished
Add source=archive option for dashboard endpoints and history service; update Docs/Tasks.md; include archive branch for list_bans, bans_by_country, ban_trend, bans_by_jail; tests for archive paths.
This commit is contained in:
@@ -11,6 +11,7 @@ from unittest.mock import AsyncMock, patch
|
||||
import aiosqlite
|
||||
import pytest
|
||||
|
||||
from app.db import init_db
|
||||
from app.services import history_service
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -264,6 +265,31 @@ class TestListHistory:
|
||||
assert result.page == 1
|
||||
assert result.page_size == 2
|
||||
|
||||
async def test_source_archive_reads_from_archive(self, f2b_db_path: str, tmp_path: Path) -> None:
|
||||
"""Using source='archive' reads from the BanGUI archive table."""
|
||||
app_db_path = str(tmp_path / "app_archive.db")
|
||||
async with aiosqlite.connect(app_db_path) as db:
|
||||
db.row_factory = aiosqlite.Row
|
||||
await init_db(db)
|
||||
await db.execute(
|
||||
"INSERT INTO history_archive (jail, ip, timeofban, bancount, data, action) VALUES (?, ?, ?, ?, ?, ?)",
|
||||
("sshd", "10.0.0.1", _ONE_HOUR_AGO, 1, '{"matches": [], "failures": 0}', "ban"),
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
with patch(
|
||||
"app.services.history_service.get_fail2ban_db_path",
|
||||
new=AsyncMock(return_value=f2b_db_path),
|
||||
):
|
||||
result = await history_service.list_history(
|
||||
"fake_socket",
|
||||
source="archive",
|
||||
db=db,
|
||||
)
|
||||
|
||||
assert result.total == 1
|
||||
assert result.items[0].ip == "10.0.0.1"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_ip_detail tests
|
||||
|
||||
Reference in New Issue
Block a user