ngxstat/tests/test_reports.py

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
  query: |
    SELECT {bucket} AS bucket, COUNT(*) AS value
    FROM logs
    GROUP BY bucket
    ORDER BY bucket
- name: error_rate
  query: |
    SELECT {bucket} 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"}


def test_generate_interval_domain_filter(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", "example.com")

    hits = json.loads(
        (tmp_path / "output" / "domains" / "example.com" / "hourly" / "hits.json").read_text()
    )
    assert hits[0]["value"] == 2


def test_generate_root_index(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")
    gr._generate_interval("daily")

    # create dummy domain directories
    (tmp_path / "output" / "domains" / "foo.com").mkdir(parents=True)
    (tmp_path / "output" / "domains" / "bar.com").mkdir(parents=True)

    gr._generate_root_index()

    index_file = tmp_path / "output" / "index.html"
    assert index_file.exists()
    content = index_file.read_text()

    # check for interval links
    assert 'data-interval="hourly"' in content
    assert 'data-interval="daily"' in content

    # check for domain links
    assert 'data-domain="foo.com"' in content
    assert 'data-domain="bar.com"' in content