history archive purge uses current age and test uses dynamic timestamps

backend/tests/test_repositories/test_history_archive_repo.py

@@ -0,0 +1,60 @@
+"""Tests for history_archive_repo."""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import aiosqlite
+import pytest
+
+from app.db import init_db
+from app.repositories.history_archive_repo import archive_ban_event, get_archived_history, purge_archived_history
+
+
+@pytest.fixture
+async def app_db(tmp_path: Path) -> str:
+    path = str(tmp_path / "app.db")
+    async with aiosqlite.connect(path) as db:
+        db.row_factory = aiosqlite.Row
+        await init_db(db)
+    return path
+
+
+@pytest.mark.asyncio
+async def test_archive_ban_event_deduplication(app_db: str) -> None:
+    async with aiosqlite.connect(app_db) as db:
+        # first insert should add
+        inserted = await archive_ban_event(db, "sshd", "1.1.1.1", 1000, 1, "{}", "ban")
+        assert inserted
+
+        # duplicate event is ignored
+        inserted = await archive_ban_event(db, "sshd", "1.1.1.1", 1000, 1, "{}", "ban")
+        assert not inserted
+
+
+@pytest.mark.asyncio
+async def test_get_archived_history_filtering_and_pagination(app_db: str) -> None:
+    async with aiosqlite.connect(app_db) as db:
+        await archive_ban_event(db, "sshd", "1.1.1.1", 1000, 1, "{}", "ban")
+        await archive_ban_event(db, "nginx", "2.2.2.2", 2000, 1, "{}", "ban")
+
+        rows, total = await get_archived_history(db, jail="sshd")
+        assert total == 1
+        assert rows[0]["ip"] == "1.1.1.1"
+
+        rows, total = await get_archived_history(db, page=1, page_size=1)
+        assert total == 2
+        assert len(rows) == 1
+
+
+@pytest.mark.asyncio
+async def test_purge_archived_history(app_db: str) -> None:
+    now = int(time.time())
+    async with aiosqlite.connect(app_db) as db:
+        await archive_ban_event(db, "sshd", "1.1.1.1", now - 3000, 1, "{}", "ban")
+        await archive_ban_event(db, "sshd", "1.1.1.2", now - 1000, 1, "{}", "ban")
+        deleted = await purge_archived_history(db, age_seconds=2000)
+        assert deleted == 1
+        rows, total = await get_archived_history(db)
+        assert total == 1