Jarian Cottingham
93e287f88b
- Add pytest test suite across all modules (config, database, file_processor, ai_processor, api, scheduler, monitoring, ftp_server) - 145 tests covering normal paths, error handling, edge cases - Mock external dependencies (AI endpoint, requests, pyftpdlib) - 95% code coverage with fail-under=90 threshold in CI - Update CI workflow to run pytest with coverage enforcement - Add pyproject.toml with pytest/coverage configuration
108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
"""Tests for monitoring module."""
|
|
|
|
import os
|
|
import pytest
|
|
import threading
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
from factsdb.monitoring import (
|
|
increment_fact_extraction,
|
|
increment_file_processing,
|
|
increment_error,
|
|
get_metrics,
|
|
get_metrics_json,
|
|
start_uptime_monitor,
|
|
_metrics,
|
|
_metrics_lock,
|
|
)
|
|
from factsdb.config import DatabaseConfig
|
|
|
|
|
|
def reset_metrics():
|
|
"""Reset global metrics to known state."""
|
|
with _metrics_lock:
|
|
_metrics["fact_extraction_count"] = 0
|
|
_metrics["file_processing_count"] = 0
|
|
_metrics["error_count"] = 0
|
|
_metrics["last_extraction_time"] = 0
|
|
_metrics["uptime_seconds"] = 0
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_metrics():
|
|
reset_metrics()
|
|
yield
|
|
reset_metrics()
|
|
|
|
|
|
class TestIncrementFactExtraction:
|
|
def test_increment(self):
|
|
increment_fact_extraction()
|
|
assert _metrics["fact_extraction_count"] == 1
|
|
assert _metrics["last_extraction_time"] > 0
|
|
|
|
def test_multiple(self):
|
|
for _ in range(3):
|
|
increment_fact_extraction()
|
|
assert _metrics["fact_extraction_count"] == 3
|
|
|
|
|
|
class TestIncrementFileProcessing:
|
|
def test_increment(self):
|
|
increment_file_processing()
|
|
assert _metrics["file_processing_count"] == 1
|
|
|
|
def test_multiple(self):
|
|
for _ in range(5):
|
|
increment_file_processing()
|
|
assert _metrics["file_processing_count"] == 5
|
|
|
|
|
|
class TestIncrementError:
|
|
def test_increment(self):
|
|
increment_error()
|
|
assert _metrics["error_count"] == 1
|
|
|
|
|
|
class TestGetMetrics:
|
|
def test_prometheus_format(self, tmp_path, monkeypatch):
|
|
db_path = str(tmp_path / "mon.db")
|
|
monkeypatch.setenv("DATABASE_PATH", db_path)
|
|
monkeypatch.setenv("AI_ENDPOINT_URL", "http://localhost:8000/v1")
|
|
monkeypatch.setenv("AI_ENDPOINT_TOKEN", "tok")
|
|
monkeypatch.setenv("FTP_USERNAME", "u")
|
|
monkeypatch.setenv("FTP_PASSWORD", "p")
|
|
text = get_metrics()
|
|
assert "factsdb_fact_extractions_total" in text
|
|
assert "factsdb_files_processed_total" in text
|
|
assert "factsdb_errors_total" in text
|
|
assert "factsdb_uptime_seconds" in text
|
|
|
|
|
|
class TestGetMetricsJson:
|
|
def test_json_format(self, tmp_path, monkeypatch):
|
|
db_path = str(tmp_path / "mon.db")
|
|
monkeypatch.setenv("DATABASE_PATH", db_path)
|
|
monkeypatch.setenv("AI_ENDPOINT_URL", "http://localhost:8000/v1")
|
|
monkeypatch.setenv("AI_ENDPOINT_TOKEN", "tok")
|
|
monkeypatch.setenv("FTP_USERNAME", "u")
|
|
monkeypatch.setenv("FTP_PASSWORD", "p")
|
|
data = get_metrics_json()
|
|
assert "metrics" in data
|
|
assert "timestamp" in data
|
|
assert "database_stats" in data
|
|
m = data["metrics"]
|
|
assert "fact_extractions" in m
|
|
assert "files_processed" in m
|
|
assert "errors" in m
|
|
assert "uptime_seconds" in m
|
|
|
|
|
|
class TestStartUptimeMonitor:
|
|
def test_returns_thread(self):
|
|
thread = start_uptime_monitor()
|
|
assert isinstance(thread, threading.Thread)
|
|
assert thread.daemon is True
|
|
assert thread.is_alive()
|