diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..e2a3dbb --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,17 @@ +name: Tests + +on: + pull_request: + push: + branches: [main] + +jobs: + unit-tests: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + - name: Run unit tests + run: python -m unittest discover -s tests -v diff --git a/tests/test_notif_audit.py b/tests/test_notif_audit.py new file mode 100644 index 0000000..5bbbb85 --- /dev/null +++ b/tests/test_notif_audit.py @@ -0,0 +1,106 @@ +import sqlite3 +import tempfile +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from types import SimpleNamespace + + +ARGUS_PATH = Path(__file__).resolve().parents[1] / "bin" / "argus" +argus = SourceFileLoader("argus_cli", str(ARGUS_PATH)).load_module() + + +def init_db(db_path: str): + conn = sqlite3.connect(db_path) + conn.execute( + """ + CREATE TABLE notifications ( + id TEXT PRIMARY KEY, + title TEXT NOT NULL, + url TEXT, + dismissed INTEGER DEFAULT 0, + acted_at TEXT, + seen_at TEXT + ) + """ + ) + conn.execute( + """ + CREATE TABLE escalations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + category TEXT, + title TEXT, + detail TEXT + ) + """ + ) + conn.commit() + conn.close() + + +class TestNotifAudit(unittest.TestCase): + def test_audit_handles_sqlite_row_without_get(self): + with tempfile.TemporaryDirectory() as td: + db_path = str(Path(td) / "argus.db") + init_db(db_path) + + conn = sqlite3.connect(db_path) + conn.execute( + """ + INSERT INTO notifications (id, title, url, dismissed, acted_at, seen_at) + VALUES (?, ?, ?, 0, NULL, '2000-01-01T00:00:00') + """, + ("n1", "old stale notification", None), + ) + conn.commit() + conn.close() + + def fake_get_db(): + c = sqlite3.connect(db_path) + c.row_factory = sqlite3.Row + return c + + argus.get_db = fake_get_db + argus.cmd_notif_audit(SimpleNamespace(stale_hours=4)) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT category, detail FROM escalations").fetchone() + conn.close() + + self.assertIsNotNone(row) + self.assertEqual(row[0], "stale-notification") + self.assertIn("Notification n1 has been pending", row[1]) + + def test_audit_no_stale_notifications_creates_no_escalation(self): + with tempfile.TemporaryDirectory() as td: + db_path = str(Path(td) / "argus.db") + init_db(db_path) + + conn = sqlite3.connect(db_path) + conn.execute( + """ + INSERT INTO notifications (id, title, url, dismissed, acted_at, seen_at) + VALUES (?, ?, ?, 0, NULL, '2999-01-01T00:00:00') + """, + ("n2", "fresh notification", "https://example.com"), + ) + conn.commit() + conn.close() + + def fake_get_db(): + c = sqlite3.connect(db_path) + c.row_factory = sqlite3.Row + return c + + argus.get_db = fake_get_db + argus.cmd_notif_audit(SimpleNamespace(stale_hours=4)) + + conn = sqlite3.connect(db_path) + count = conn.execute("SELECT COUNT(*) FROM escalations").fetchone()[0] + conn.close() + + self.assertEqual(count, 0) + + +if __name__ == "__main__": + unittest.main()