257 lines
8.8 KiB
Python
257 lines
8.8 KiB
Python
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app import create_app, db
|
|
from app.services.admin_peripheral_monitor import (
|
|
append_peripheral_event,
|
|
build_hardware_status,
|
|
clear_peripheral_events,
|
|
diagnostics_hardware_block,
|
|
read_peripheral_events,
|
|
reset_debounce_state_for_tests,
|
|
)
|
|
from app.services.admin_system_metrics import _detect_root_disk_media
|
|
from config import TestingConfig
|
|
|
|
|
|
class PeripheralMonitorConfig(TestingConfig):
|
|
AUTH_LOGIN = "periph-admin"
|
|
AUTH_PASSWORD = "periph-secret"
|
|
WESP_PERIPHERAL_EVENT_DEBOUNCE_SEC = 60
|
|
SIMULATION_MODE = True
|
|
|
|
|
|
class AdminPeripheralMonitorTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
reset_debounce_state_for_tests()
|
|
self._tmp_dir = tempfile.mkdtemp(prefix="wesp-periph-monitor-")
|
|
log_path = os.path.join(self._tmp_dir, "peripherals.jsonl")
|
|
|
|
class _Cfg(PeripheralMonitorConfig):
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(self._tmp_dir, 'recipes_test.db')}"
|
|
SQLALCHEMY_BINDS = {
|
|
"reports": f"sqlite:///{os.path.join(self._tmp_dir, 'reports_test.db')}"
|
|
}
|
|
WESP_ADMIN_PERIPHERAL_LOG_PATH = log_path
|
|
|
|
self.app = create_app(_Cfg)
|
|
self.client = self.app.test_client()
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
|
|
def tearDown(self) -> None:
|
|
reset_debounce_state_for_tests()
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def test_append_and_read_events(self) -> None:
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="hx711",
|
|
code="hx711_read_error",
|
|
level="err",
|
|
message="test error",
|
|
force=True,
|
|
)
|
|
events, path = read_peripheral_events(self.app, limit=10)
|
|
self.assertTrue(path)
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["code"], "hx711_read_error")
|
|
|
|
def test_clear_peripheral_events(self) -> None:
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="hx711",
|
|
code="hx711_read_error",
|
|
message="to clear",
|
|
force=True,
|
|
)
|
|
events, _ = read_peripheral_events(self.app, limit=10)
|
|
self.assertEqual(len(events), 1)
|
|
ok, path = clear_peripheral_events(self.app)
|
|
self.assertTrue(ok)
|
|
self.assertTrue(path)
|
|
events, _ = read_peripheral_events(self.app, limit=10)
|
|
self.assertEqual(events, [])
|
|
|
|
def test_admin_peripheral_events_clear_api(self) -> None:
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="gpio",
|
|
code="gpio_blink",
|
|
message="blink",
|
|
force=True,
|
|
)
|
|
self.client = self.app.test_client()
|
|
login = self.client.post(
|
|
"/api/auth/login",
|
|
json={"login": "periph-admin", "password": "periph-secret"},
|
|
)
|
|
self.assertEqual(login.status_code, 200)
|
|
r = self.client.delete("/api/admin/peripheral-events")
|
|
self.assertEqual(r.status_code, 200, r.get_data(as_text=True))
|
|
self.assertEqual(r.get_json().get("status"), "success")
|
|
events, _ = read_peripheral_events(self.app, limit=10)
|
|
self.assertEqual(events, [])
|
|
|
|
def test_read_excludes_network_component(self) -> None:
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="network",
|
|
code="mdns_started",
|
|
message="net",
|
|
force=True,
|
|
)
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="hx711",
|
|
code="hx711_read_error",
|
|
message="hw",
|
|
force=True,
|
|
)
|
|
events, _ = read_peripheral_events(self.app, limit=10, exclude_component="network")
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0].get("component"), "hx711")
|
|
|
|
def test_clear_excludes_network_component(self) -> None:
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="network",
|
|
code="mdns_started",
|
|
message="net",
|
|
force=True,
|
|
)
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="gpio",
|
|
code="gpio_blink",
|
|
message="blink",
|
|
force=True,
|
|
)
|
|
ok, _ = clear_peripheral_events(self.app, exclude_component="network")
|
|
self.assertTrue(ok)
|
|
net_events, _ = read_peripheral_events(self.app, limit=10, component="network")
|
|
hw_events, _ = read_peripheral_events(self.app, limit=10, exclude_component="network")
|
|
self.assertEqual(len(net_events), 1)
|
|
self.assertEqual(net_events[0].get("code"), "mdns_started")
|
|
self.assertEqual(hw_events, [])
|
|
|
|
def test_debounce_skips_duplicate(self) -> None:
|
|
self.app.config["WESP_PERIPHERAL_EVENT_DEBOUNCE_SEC"] = 120
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="hx711",
|
|
code="hx711_read_error",
|
|
message="first",
|
|
force=True,
|
|
)
|
|
append_peripheral_event(
|
|
self.app,
|
|
component="hx711",
|
|
code="hx711_read_error",
|
|
message="second",
|
|
)
|
|
events, _ = read_peripheral_events(self.app, limit=10)
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["message"], "first")
|
|
|
|
def test_diagnostics_hardware_block_simulation(self) -> None:
|
|
from app.models import HardwareSetting
|
|
from app.routes import scales as scales_module
|
|
|
|
if scales_module._scales_reader is not None:
|
|
scales_module._scales_reader.stop()
|
|
scales_module._scales_reader = None
|
|
db.session.add(
|
|
HardwareSetting(
|
|
id=1,
|
|
counts_per_kg=1.0,
|
|
tare_raw=0.0,
|
|
simulation_mode=True,
|
|
simulation_weight_kg=0.0,
|
|
)
|
|
)
|
|
db.session.commit()
|
|
block = diagnostics_hardware_block(self.app)
|
|
self.assertTrue(block["simulation_mode"])
|
|
self.assertTrue(block["hx711_driver_ready"] or block["drivers_absent"] is False)
|
|
|
|
def test_build_hardware_status_structure(self) -> None:
|
|
mock_reader = MagicMock()
|
|
mock_reader.get_scale_health.return_value = {
|
|
"simulation_mode": True,
|
|
"hx711_ok": True,
|
|
"hx711_error": None,
|
|
}
|
|
mock_reader.get_scale_debug_snapshot.return_value = {
|
|
"counts_per_kg": 1.0,
|
|
"tare_raw": 0.0,
|
|
"raw_history_len": 5,
|
|
"last_raw": 10.0,
|
|
"raw_min_recent": 8.0,
|
|
"raw_max_recent": 12.0,
|
|
"error_streak": 0,
|
|
"last_success_at": time.time(),
|
|
}
|
|
mock_reader.get_current_weight.return_value = 3
|
|
mock_reader.get_last_hx711_error.return_value = None
|
|
mock_reader.get_simulation_state.return_value = {
|
|
"simulation_mode": True,
|
|
"simulation_weight_kg": 0.0,
|
|
}
|
|
|
|
with patch("app.routes.scales._get_reader", return_value=mock_reader):
|
|
payload = build_hardware_status(self.app)
|
|
self.assertIn("machine", payload)
|
|
self.assertIn("host_alerts", payload)
|
|
self.assertIn("peripherals", payload)
|
|
self.assertTrue(payload["scales"]["available"])
|
|
self.assertEqual(payload["scales"]["weight_kg"], 3)
|
|
|
|
def test_detect_root_disk_media_no_name_error(self) -> None:
|
|
try:
|
|
label = _detect_root_disk_media()
|
|
except NameError as exc:
|
|
self.fail(f"_detect_root_disk_media raised NameError: {exc}")
|
|
self.assertIsInstance(label, str)
|
|
|
|
def test_admin_hardware_status_api(self) -> None:
|
|
from app.routes import scales as scales_module
|
|
|
|
reader = getattr(scales_module, "_scales_reader", None)
|
|
if reader is not None:
|
|
reader.stop()
|
|
scales_module._scales_reader = None
|
|
|
|
self.client = self.app.test_client()
|
|
login = self.client.post(
|
|
"/api/auth/login",
|
|
json={"login": "periph-admin", "password": "periph-secret"},
|
|
)
|
|
self.assertEqual(login.status_code, 200)
|
|
|
|
r = self.client.get("/api/admin/hardware-status")
|
|
self.assertEqual(r.status_code, 200)
|
|
body = r.get_json()
|
|
self.assertEqual(body.get("status"), "success")
|
|
self.assertIn("scales", body)
|
|
self.assertIn("peripherals", body)
|
|
|
|
ev = self.client.get("/api/admin/peripheral-events?limit=5")
|
|
self.assertEqual(ev.status_code, 200)
|
|
self.assertIn("events", ev.get_json())
|
|
|
|
summary = self.client.get("/api/admin/summary")
|
|
self.assertEqual(summary.status_code, 200)
|
|
hw = summary.get_json().get("hardware") or {}
|
|
self.assertIn("simulation_mode", hw)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|