8314a61815
Task 1 — Presence: ping family phones, derive household state (EMPTY/KIDS_HOME/ADULTS_HOME/ALL_HOME), configurable departure delay, per-member roles, auto-arm actions via MQTT. Task 2 — Detection: MobileNet-SSD v2 via OpenCV DNN for person/vehicle classification. Vehicle color/size fingerprinting for known car matching. Zone-based filtering per camera. Model download script. Task 3 — Health: periodic disk/MQTT/subsystem checks, auto-prune oldest non-starred recordings on disk pressure, daily digest builder. 126 tests passing. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
"""Tests for health monitoring."""
|
|
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from vigilar.health.pruner import find_prunable_recordings, calculate_disk_usage_pct
|
|
from vigilar.health.monitor import HealthCheck, HealthStatus, check_disk, check_mqtt_port
|
|
|
|
|
|
class TestDiskCheck:
|
|
@patch("vigilar.health.monitor.shutil.disk_usage")
|
|
def test_healthy(self, mock_usage):
|
|
mock_usage.return_value = type("U", (), {"total": 100_000_000_000, "used": 50_000_000_000, "free": 50_000_000_000})()
|
|
result = check_disk("/var/vigilar", warn_pct=85, critical_pct=95)
|
|
assert result.status == HealthStatus.HEALTHY
|
|
|
|
@patch("vigilar.health.monitor.shutil.disk_usage")
|
|
def test_warning(self, mock_usage):
|
|
mock_usage.return_value = type("U", (), {"total": 100_000_000_000, "used": 90_000_000_000, "free": 10_000_000_000})()
|
|
result = check_disk("/var/vigilar", warn_pct=85, critical_pct=95)
|
|
assert result.status == HealthStatus.WARNING
|
|
|
|
@patch("vigilar.health.monitor.shutil.disk_usage")
|
|
def test_critical(self, mock_usage):
|
|
mock_usage.return_value = type("U", (), {"total": 100_000_000_000, "used": 97_000_000_000, "free": 3_000_000_000})()
|
|
result = check_disk("/var/vigilar", warn_pct=85, critical_pct=95)
|
|
assert result.status == HealthStatus.CRITICAL
|
|
|
|
|
|
class TestPruner:
|
|
def test_calculate_disk_pct(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
pct = calculate_disk_usage_pct(d)
|
|
assert 0 <= pct <= 100
|
|
|
|
def test_find_prunable_empty(self, test_db):
|
|
result = find_prunable_recordings(test_db, limit=10)
|
|
assert result == []
|
|
|
|
|
|
class TestMQTTCheck:
|
|
@patch("vigilar.health.monitor.socket.create_connection")
|
|
def test_mqtt_reachable(self, mock_conn):
|
|
mock_conn.return_value.__enter__ = lambda s: s
|
|
mock_conn.return_value.__exit__ = lambda s, *a: None
|
|
result = check_mqtt_port("127.0.0.1", 1883)
|
|
assert result.status == HealthStatus.HEALTHY
|
|
|
|
@patch("vigilar.health.monitor.socket.create_connection")
|
|
def test_mqtt_unreachable(self, mock_conn):
|
|
mock_conn.side_effect = ConnectionRefusedError()
|
|
result = check_mqtt_port("127.0.0.1", 1883)
|
|
assert result.status == HealthStatus.CRITICAL
|