90 lines
2.3 KiB
Python
90 lines
2.3 KiB
Python
import sys
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.append(str(REPO_ROOT))
|
|
from scripts import analyze
|
|
from scripts import generate_reports as gr
|
|
|
|
|
|
def setup_db(path: Path) -> None:
|
|
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",
|
|
"missing.com",
|
|
"2024-01-01 11:00:00",
|
|
"GET / HTTP/1.1",
|
|
200,
|
|
100,
|
|
"-",
|
|
"curl",
|
|
"MISS",
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def test_check_missing_domains(tmp_path, monkeypatch, capsys):
|
|
db_path = tmp_path / "database" / "ngxstat.db"
|
|
setup_db(db_path)
|
|
|
|
conf = tmp_path / "nginx.conf"
|
|
conf.write_text(
|
|
"""
|
|
server {
|
|
listen 80;
|
|
server_name example.com;
|
|
}
|
|
"""
|
|
)
|
|
|
|
monkeypatch.setattr(analyze, "DB_PATH", db_path)
|
|
monkeypatch.setattr(gr, "DB_PATH", db_path)
|
|
monkeypatch.setattr(analyze.nginx_config, "DEFAULT_PATHS", [str(conf)])
|
|
|
|
analyze.check_missing_domains(json_output=False)
|
|
out = capsys.readouterr().out.strip().splitlines()
|
|
assert out == ["missing.com"]
|
|
|
|
analyze.check_missing_domains(json_output=True)
|
|
out_json = json.loads(capsys.readouterr().out.strip())
|
|
assert out_json == ["missing.com"]
|