61 lines
1.4 KiB
Python
61 lines
1.4 KiB
Python
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
|