Add idempotent import test

tests/test_importer.py

@@ -0,0 +1,62 @@
+import os
+import sqlite3
+from pathlib import Path
+import runpy
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SCRIPT_PATH = REPO_ROOT / "scripts" / "init_db.py"
+
+
+@pytest.fixture()
+def sample_logs(tmp_path):
+    log_dir = Path("/var/log/nginx")
+    log_dir.mkdir(parents=True, exist_ok=True)
+
+    (log_dir / "access.log.1").write_text(
+        "127.0.0.1 - example.com [01/Jan/2024:10:00:00 +0000] \"GET / HTTP/1.1\" 200 123 \"-\" \"curl\" MISS\n"
+    )
+    (log_dir / "access.log").write_text(
+        "127.0.0.1 - example.com [01/Jan/2024:10:05:00 +0000] \"GET /about HTTP/1.1\" 200 123 \"-\" \"curl\" MISS\n"
+    )
+
+    yield log_dir
+
+    for f in log_dir.glob("access.log*"):
+        f.unlink()
+    try:
+        log_dir.rmdir()
+    except OSError:
+        pass
+
+
+def run_import(temp_dir):
+    db_path = temp_dir / "database" / "ngxstat.db"
+    cwd = os.getcwd()
+    os.chdir(temp_dir)
+    try:
+        runpy.run_path(str(SCRIPT_PATH), run_name="__main__")
+    finally:
+        os.chdir(cwd)
+    return db_path
+
+
+def count_rows(db_path):
+    conn = sqlite3.connect(db_path)
+    cur = conn.cursor()
+    cur.execute("SELECT COUNT(*) FROM logs")
+    count = cur.fetchone()[0]
+    conn.close()
+    return count
+
+
+def test_idempotent_import(sample_logs, tmp_path):
+    db_path = run_import(tmp_path)
+    first_count = count_rows(db_path)
+    db_path = run_import(tmp_path)
+    second_count = count_rows(db_path)
+
+    assert first_count == 2
+    assert second_count == first_count
+