You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
92 lines
3.5 KiB
92 lines
3.5 KiB
from datetime import datetime, timedelta
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from health import HealthStatus
|
|
from health.checks import check_embedding_coverage, check_llm_coverage, check_motion_freshness
|
|
|
|
|
|
class TestCheckMotionFreshness:
|
|
def test_recent_motions_ok(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.return_value = [150]
|
|
result = check_motion_freshness(conn, max_age_days=7, min_motions=100)
|
|
assert result.status == HealthStatus.OK
|
|
assert result.details["count"] == 150
|
|
|
|
def test_no_motions_critical(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.return_value = [0]
|
|
result = check_motion_freshness(conn, max_age_days=7, min_motions=100)
|
|
assert result.status == HealthStatus.CRITICAL
|
|
|
|
def test_low_count_warning(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.return_value = [50]
|
|
result = check_motion_freshness(conn, max_age_days=7, min_motions=100)
|
|
assert result.status == HealthStatus.WARNING
|
|
|
|
def test_query_error_critical(self):
|
|
conn = MagicMock()
|
|
conn.execute.side_effect = Exception("db down")
|
|
result = check_motion_freshness(conn)
|
|
assert result.status == HealthStatus.CRITICAL
|
|
assert "db down" in result.message
|
|
|
|
|
|
class TestCheckEmbeddingCoverage:
|
|
def test_full_coverage_ok(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.side_effect = [[100], [100]]
|
|
result = check_embedding_coverage(conn, min_coverage=0.95)
|
|
assert result.status == HealthStatus.OK
|
|
assert result.details["coverage"] == 1.0
|
|
|
|
def test_low_coverage_warning(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.side_effect = [[100], [80]]
|
|
result = check_embedding_coverage(conn, min_coverage=0.95)
|
|
assert result.status == HealthStatus.WARNING
|
|
assert result.details["coverage"] == 0.8
|
|
|
|
def test_empty_db_critical(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.side_effect = [[0], [0]]
|
|
result = check_embedding_coverage(conn)
|
|
assert result.status == HealthStatus.CRITICAL
|
|
|
|
def test_query_error_critical(self):
|
|
conn = MagicMock()
|
|
conn.execute.side_effect = Exception("db down")
|
|
result = check_embedding_coverage(conn)
|
|
assert result.status == HealthStatus.CRITICAL
|
|
|
|
|
|
class TestCheckLLMCoverage:
|
|
def test_full_coverage_ok(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.side_effect = [[100], [0]]
|
|
result = check_llm_coverage(conn)
|
|
assert result.status == HealthStatus.OK
|
|
assert result.details["missing_ratio"] == 0.0
|
|
|
|
def test_some_missing_warning(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.side_effect = [[100], [10]]
|
|
result = check_llm_coverage(conn)
|
|
assert result.status == HealthStatus.WARNING
|
|
assert result.details["missing_ratio"] == 0.1
|
|
|
|
def test_too_many_missing_critical(self):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.side_effect = [[100], [20]]
|
|
result = check_llm_coverage(conn, max_missing_ratio=0.15)
|
|
assert result.status == HealthStatus.CRITICAL
|
|
assert result.details["missing_ratio"] == 0.2
|
|
|
|
def test_query_error_critical(self):
|
|
conn = MagicMock()
|
|
conn.execute.side_effect = Exception("db down")
|
|
result = check_llm_coverage(conn)
|
|
assert result.status == HealthStatus.CRITICAL
|
|
|