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