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()