87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
import sqlite3
|
|
from pathlib import Path
|
|
import json
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.append(str(REPO_ROOT))
|
|
from scripts import generate_reports as gr
|
|
|
|
|
|
def setup_db(path: Path):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(path)
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE logs (
|
|
id INTEGER PRIMARY KEY,
|
|
ip TEXT,
|
|
host TEXT,
|
|
time TEXT,
|
|
request TEXT,
|
|
status INTEGER,
|
|
bytes_sent INTEGER,
|
|
referer TEXT,
|
|
user_agent TEXT,
|
|
cache_status TEXT
|
|
)
|
|
"""
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO logs (ip, host, time, request, status, bytes_sent, referer, user_agent, cache_status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
("127.0.0.1", "example.com", "2024-01-01 10:00:00", "GET / HTTP/1.1", 200, 100, "-", "curl", "MISS"),
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO logs (ip, host, time, request, status, bytes_sent, referer, user_agent, cache_status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
("127.0.0.1", "example.com", "2024-01-01 10:05:00", "GET /err HTTP/1.1", 500, 100, "-", "curl", "MISS"),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def sample_reports(tmp_path):
|
|
cfg = tmp_path / "reports.yml"
|
|
cfg.write_text(
|
|
"""
|
|
- name: hits
|
|
interval: hourly
|
|
query: |
|
|
SELECT strftime('%Y-%m-%d %H:00:00', datetime(time)) AS bucket, COUNT(*) AS value
|
|
FROM logs
|
|
GROUP BY bucket
|
|
ORDER BY bucket
|
|
- name: error_rate
|
|
interval: hourly
|
|
query: |
|
|
SELECT strftime('%Y-%m-%d %H:00:00', datetime(time)) AS bucket,
|
|
SUM(CASE WHEN status >= 400 THEN 1 ELSE 0 END) * 100.0 / COUNT(*) AS value
|
|
FROM logs
|
|
GROUP BY bucket
|
|
ORDER BY bucket
|
|
"""
|
|
)
|
|
return cfg
|
|
|
|
|
|
def test_generate_interval(tmp_path, sample_reports, monkeypatch):
|
|
db_path = tmp_path / "database" / "ngxstat.db"
|
|
setup_db(db_path)
|
|
|
|
monkeypatch.setattr(gr, "DB_PATH", db_path)
|
|
monkeypatch.setattr(gr, "OUTPUT_DIR", tmp_path / "output")
|
|
monkeypatch.setattr(gr, "REPORT_CONFIG", sample_reports)
|
|
monkeypatch.setattr(gr, "TEMPLATE_DIR", Path(__file__).resolve().parents[1] / "templates")
|
|
|
|
gr._generate_interval("hourly")
|
|
|
|
hits = json.loads((tmp_path / "output" / "hourly" / "hits.json").read_text())
|
|
assert hits[0]["value"] == 2
|
|
error_rate = json.loads((tmp_path / "output" / "hourly" / "error_rate.json").read_text())
|
|
assert error_rate[0]["value"] == pytest.approx(50.0)
|
|
reports = json.loads((tmp_path / "output" / "hourly" / "reports.json").read_text())
|
|
assert {r["name"] for r in reports} == {"hits", "error_rate"}
|
|
|