68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for dual-writing generated reports to SQLite."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import app.scan
|
|
from app.analyzer import Analyzer
|
|
from app.db import connect
|
|
from app.db import repositories as repo
|
|
from app.utils.blocklist_history import write_blocklist_report
|
|
|
|
|
|
class ReportPersistenceTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp_dir.name)
|
|
self.conn = connect(str(self.root / "test.db"))
|
|
self.connection_patch = patch("app.db.get_connection",
|
|
return_value=self.conn)
|
|
self.connection_patch.start()
|
|
|
|
def tearDown(self):
|
|
self.connection_patch.stop()
|
|
self.conn.close()
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_analyzer_summary_is_stored_with_file_provenance(self):
|
|
report_dir = self.root / "copyparty"
|
|
report_dir.mkdir()
|
|
path = report_dir / "2026-11-06-11-30-10.txt"
|
|
path.write_text("summary body", encoding="utf-8")
|
|
analyzer = Analyzer("logs", str(report_dir), quiet=True)
|
|
|
|
analyzer.persist_report_artifact(str(path), scan_id=14)
|
|
|
|
item = repo.list_report_artifacts(
|
|
self.conn, source="ids")["items"][0]
|
|
detail = repo.get_report_artifact(self.conn, item["id"])
|
|
self.assertEqual(item["original_path"], "ids/2026-11-06-11-30-10.txt")
|
|
self.assertEqual(detail["scan_id"], 14)
|
|
self.assertEqual(detail["payload"]["content"], "summary body")
|
|
|
|
def test_blocklist_report_is_stored_as_structured_artifact(self):
|
|
path = write_blocklist_report(
|
|
str(self.root / "reports"), history_v4=set(), history_v6=set(),
|
|
prev_v4=set(), prev_v6=set(),
|
|
complete_v4={"192.0.2.1/32"}, complete_v6=set(),
|
|
firewall_changes={"added_ipv4": ["192.0.2.1/32"]},
|
|
dry_run=True, scan_id=22,
|
|
)
|
|
|
|
items = repo.list_report_artifacts(
|
|
self.conn, source="app", report_type="blocklist_change")["items"]
|
|
self.assertEqual(len(items), 1)
|
|
detail = repo.get_report_artifact(self.conn, items[0]["id"])
|
|
self.assertEqual(detail["scan_id"], 22)
|
|
self.assertEqual(detail["payload"]["blocked_ipv4"], ["192.0.2.1/32"])
|
|
self.assertTrue(Path(path).exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|