|
|
@@ -0,0 +1,369 @@
|
|
|
+"""Power alerts: checked in the background, sent once per crossing.
|
|
|
+
|
|
|
+Pre-fix the check called a ``notification_service.send_notification`` that does
|
|
|
+not exist, so no power alert was ever sent and the plug status request failed
|
|
|
+with 500 once a threshold was crossed. It also only ran from that status
|
|
|
+request, i.e. while someone had the printer page open, and it fired every five
|
|
|
+minutes for as long as the power stayed outside the range.
|
|
|
+"""
|
|
|
+
|
|
|
+from datetime import timedelta
|
|
|
+from unittest.mock import AsyncMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+from sqlalchemy import select
|
|
|
+
|
|
|
+from backend.app.models.smart_plug import SmartPlug
|
|
|
+from backend.app.services.power_alerts import PowerAlertMonitor, check_all_plugs
|
|
|
+from backend.app.utils.local_time import utcnow_naive
|
|
|
+
|
|
|
+
|
|
|
+async def _plug(db_session, **kwargs) -> SmartPlug:
|
|
|
+ fields = {
|
|
|
+ "name": "Printer plug",
|
|
|
+ "ip_address": "192.168.1.50",
|
|
|
+ "power_alert_enabled": True,
|
|
|
+ "power_alert_high": 200.0,
|
|
|
+ "power_alert_low": None,
|
|
|
+ }
|
|
|
+ fields.update(kwargs)
|
|
|
+ plug = SmartPlug(**fields)
|
|
|
+ db_session.add(plug)
|
|
|
+ await db_session.commit()
|
|
|
+ return plug
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def sent():
|
|
|
+ with patch(
|
|
|
+ "backend.app.services.power_alerts.notification_service.on_printer_error", new_callable=AsyncMock
|
|
|
+ ) as on_printer_error:
|
|
|
+ yield on_printer_error
|
|
|
+
|
|
|
+
|
|
|
+class TestCrossings:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_alerts_once_when_power_rises_above_the_high_value(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session)
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+
|
|
|
+ for watts in (150.0, 250.0, 260.0, 255.0):
|
|
|
+ await monitor.evaluate(plug, "ON", watts, db_session)
|
|
|
+
|
|
|
+ sent.assert_awaited_once()
|
|
|
+ kwargs = sent.await_args.kwargs
|
|
|
+ assert kwargs["printer_name"] == "Printer plug"
|
|
|
+ assert kwargs["error_type"] == "Power High"
|
|
|
+ assert "250.0W" in kwargs["error_detail"] and "200.0W" in kwargs["error_detail"]
|
|
|
+ assert plug.power_alert_last_triggered is not None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_alerts_again_after_the_power_was_back_in_range(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session)
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+
|
|
|
+ await monitor.evaluate(plug, "ON", 250.0, db_session)
|
|
|
+ await monitor.evaluate(plug, "ON", 150.0, db_session)
|
|
|
+ plug.power_alert_last_triggered = utcnow_naive() - timedelta(minutes=10)
|
|
|
+ await monitor.evaluate(plug, "ON", 250.0, db_session)
|
|
|
+
|
|
|
+ assert sent.await_count == 2
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_alerts_when_power_falls_below_the_low_value(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session, power_alert_high=None, power_alert_low=20.0)
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+
|
|
|
+ await monitor.evaluate(plug, "ON", 80.0, db_session)
|
|
|
+ await monitor.evaluate(plug, "ON", 5.0, db_session)
|
|
|
+
|
|
|
+ sent.assert_awaited_once()
|
|
|
+ assert sent.await_args.kwargs["error_type"] == "Power Low"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_switched_off_plug_raises_no_low_alert(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session, power_alert_high=None, power_alert_low=20.0)
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+
|
|
|
+ await monitor.evaluate(plug, "ON", 80.0, db_session)
|
|
|
+ await monitor.evaluate(plug, "OFF", 0.0, db_session)
|
|
|
+
|
|
|
+ sent.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_crossing_inside_the_cooldown_is_sent_once_it_ends(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session, power_alert_last_triggered=utcnow_naive())
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+
|
|
|
+ await monitor.evaluate(plug, "ON", 250.0, db_session)
|
|
|
+ sent.assert_not_awaited()
|
|
|
+
|
|
|
+ plug.power_alert_last_triggered = utcnow_naive() - timedelta(minutes=10)
|
|
|
+ await monitor.evaluate(plug, "ON", 250.0, db_session)
|
|
|
+ sent.assert_awaited_once()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_disabled_alerts_send_nothing(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session, power_alert_enabled=False)
|
|
|
+
|
|
|
+ await PowerAlertMonitor().evaluate(plug, "ON", 999.0, db_session)
|
|
|
+
|
|
|
+ sent.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_linked_printer_is_passed_for_the_provider_filter(self, db_session, sent, printer_factory):
|
|
|
+ printer = await printer_factory()
|
|
|
+ plug = await _plug(db_session, printer_id=printer.id)
|
|
|
+
|
|
|
+ await PowerAlertMonitor().evaluate(plug, "ON", 250.0, db_session)
|
|
|
+
|
|
|
+ assert sent.await_args.kwargs["printer_id"] == printer.id
|
|
|
+
|
|
|
+
|
|
|
+class TestBackgroundPass:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_reads_only_plugs_with_alerts_on_and_survives_an_unreachable_one(self, db_session, sent):
|
|
|
+ down = await _plug(db_session, name="Down")
|
|
|
+ up = await _plug(db_session, name="Up", ip_address="192.168.1.51")
|
|
|
+ await _plug(db_session, name="Off", ip_address="192.168.1.52", power_alert_enabled=False)
|
|
|
+ down_id, up_id = down.id, up.id
|
|
|
+
|
|
|
+ read_names = []
|
|
|
+
|
|
|
+ async def read(plug, db):
|
|
|
+ read_names.append(plug.name)
|
|
|
+ if plug.id == down_id:
|
|
|
+ raise TimeoutError("unreachable")
|
|
|
+ return "ON", 250.0
|
|
|
+
|
|
|
+ with patch("backend.app.services.power_alerts.read_plug", side_effect=read):
|
|
|
+ await check_all_plugs(db_session, PowerAlertMonitor())
|
|
|
+
|
|
|
+ assert sorted(read_names) == ["Down", "Up"]
|
|
|
+ sent.assert_awaited_once()
|
|
|
+ assert sent.await_args.kwargs["printer_name"] == "Up"
|
|
|
+ refreshed = (await db_session.execute(select(SmartPlug).where(SmartPlug.id == up_id))).scalar_one()
|
|
|
+ assert refreshed.power_alert_last_triggered is not None
|
|
|
+
|
|
|
+
|
|
|
+class TestStatusRouteNoLongerChecks:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_status_answers_with_a_plug_above_its_threshold(self, async_client, db_session, sent):
|
|
|
+ plug = await _plug(db_session)
|
|
|
+ with (
|
|
|
+ patch(
|
|
|
+ "backend.app.api.routes.smart_plugs.tasmota_service.get_status",
|
|
|
+ AsyncMock(return_value={"state": "ON", "reachable": True, "device_name": None}),
|
|
|
+ ),
|
|
|
+ patch(
|
|
|
+ "backend.app.api.routes.smart_plugs.tasmota_service.get_energy",
|
|
|
+ AsyncMock(return_value={"power": 250.0, "today": 1.0, "total": 5.0}),
|
|
|
+ ),
|
|
|
+ ):
|
|
|
+ response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}/status")
|
|
|
+
|
|
|
+ assert response.status_code == 200, response.text
|
|
|
+ assert response.json()["energy"]["power"] == 250.0
|
|
|
+ sent.assert_not_awaited()
|
|
|
+
|
|
|
+
|
|
|
+class TestReadPlug:
|
|
|
+ """State and power for every plug type; an unreachable plug reads as nothing."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ @pytest.mark.parametrize("plug_type", ["tasmota", "rest", "homeassistant"])
|
|
|
+ async def test_polled_plugs(self, db_session, plug_type):
|
|
|
+ from backend.app.services import plug_energy
|
|
|
+
|
|
|
+ plug = await _plug(db_session, plug_type=plug_type)
|
|
|
+ service = {
|
|
|
+ "tasmota": plug_energy.tasmota_service,
|
|
|
+ "rest": plug_energy.rest_smart_plug_service,
|
|
|
+ "homeassistant": plug_energy.homeassistant_service,
|
|
|
+ }[plug_type]
|
|
|
+ with (
|
|
|
+ patch.object(service, "get_status", AsyncMock(return_value={"state": "ON", "reachable": True})),
|
|
|
+ patch.object(service, "get_energy", AsyncMock(return_value={"power": 42.5})),
|
|
|
+ patch(
|
|
|
+ "backend.app.api.routes.settings.get_homeassistant_settings",
|
|
|
+ AsyncMock(return_value={"ha_url": "http://ha", "ha_token": "t"}),
|
|
|
+ ),
|
|
|
+ ):
|
|
|
+ assert await plug_energy.read_plug(plug, db_session) == ("ON", 42.5)
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch.object(service, "get_status", AsyncMock(return_value={"state": None, "reachable": False})),
|
|
|
+ patch(
|
|
|
+ "backend.app.api.routes.settings.get_homeassistant_settings",
|
|
|
+ AsyncMock(return_value={"ha_url": "http://ha", "ha_token": "t"}),
|
|
|
+ ),
|
|
|
+ ):
|
|
|
+ assert await plug_energy.read_plug(plug, db_session) == (None, None)
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_mqtt_plug_answers_from_its_last_message(self, db_session):
|
|
|
+ from types import SimpleNamespace
|
|
|
+
|
|
|
+ from backend.app.services import plug_energy
|
|
|
+
|
|
|
+ plug = await _plug(db_session, plug_type="mqtt")
|
|
|
+ mqtt = plug_energy.mqtt_relay.smart_plug_service
|
|
|
+ with (
|
|
|
+ patch.object(mqtt, "get_plug_data", return_value=SimpleNamespace(state="OFF", power=0.0, energy=1.0)),
|
|
|
+ patch.object(mqtt, "is_reachable", return_value=True),
|
|
|
+ ):
|
|
|
+ assert await plug_energy.read_plug(plug, db_session) == ("OFF", 0.0)
|
|
|
+ with patch.object(mqtt, "get_plug_data", return_value=None):
|
|
|
+ assert await plug_energy.read_plug(plug, db_session) == (None, None)
|
|
|
+
|
|
|
+
|
|
|
+class TestReviewRound1:
|
|
|
+ """Alerts must not be lost on a failed write, one plug's error must not stop
|
|
|
+ the others, stale state must not swallow a crossing, a plug that can't say
|
|
|
+ whether it is on raises no low alert, and an unreachable plug is not polled
|
|
|
+ every minute."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_failed_write_does_not_lose_the_alert(self, db_session, sent):
|
|
|
+ from sqlalchemy.exc import OperationalError
|
|
|
+
|
|
|
+ plug = await _plug(db_session)
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+ real_commit = db_session.commit
|
|
|
+ failing = AsyncMock(side_effect=OperationalError("UPDATE", {}, Exception("database is locked")))
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.power_alerts.read_plug", AsyncMock(return_value=("ON", 250.0))),
|
|
|
+ patch.object(db_session, "commit", failing),
|
|
|
+ ):
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ sent.assert_not_awaited()
|
|
|
+
|
|
|
+ db_session.commit = real_commit
|
|
|
+ with patch("backend.app.services.power_alerts.read_plug", AsyncMock(return_value=("ON", 250.0))):
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ sent.assert_awaited_once()
|
|
|
+ assert sent.await_args.kwargs["printer_name"] == plug.name
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_failed_send_is_retried_after_the_cooldown(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session)
|
|
|
+ plug_id = plug.id
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+ sent.side_effect = [RuntimeError("provider down"), None]
|
|
|
+
|
|
|
+ with patch("backend.app.services.power_alerts.read_plug", AsyncMock(return_value=("ON", 250.0))):
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ row = (await db_session.execute(select(SmartPlug).where(SmartPlug.id == plug_id))).scalar_one()
|
|
|
+ row.power_alert_last_triggered = utcnow_naive() - timedelta(minutes=10)
|
|
|
+ await db_session.commit()
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+
|
|
|
+ assert sent.await_count == 2
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_one_plugs_db_error_does_not_stop_the_others(self, test_engine, db_session, sent):
|
|
|
+ """A real failed UPDATE leaves the session needing a rollback, as
|
|
|
+ "database is locked" does on SQLite."""
|
|
|
+ from sqlalchemy import event
|
|
|
+ from sqlalchemy.exc import OperationalError
|
|
|
+
|
|
|
+ await _plug(db_session, name="First")
|
|
|
+ await _plug(db_session, name="Second", ip_address="192.168.1.51")
|
|
|
+ failed = {"done": False}
|
|
|
+
|
|
|
+ def fail_first_update(conn, cursor, statement, parameters, context, executemany):
|
|
|
+ if statement.lstrip().upper().startswith("UPDATE SMART_PLUGS") and not failed["done"]:
|
|
|
+ failed["done"] = True
|
|
|
+ raise OperationalError(statement, parameters, Exception("database is locked"))
|
|
|
+
|
|
|
+ event.listen(test_engine.sync_engine, "before_cursor_execute", fail_first_update)
|
|
|
+ try:
|
|
|
+ with patch("backend.app.services.power_alerts.read_plug", AsyncMock(return_value=("ON", 250.0))):
|
|
|
+ await check_all_plugs(db_session, PowerAlertMonitor())
|
|
|
+ finally:
|
|
|
+ event.remove(test_engine.sync_engine, "before_cursor_execute", fail_first_update)
|
|
|
+
|
|
|
+ assert failed["done"]
|
|
|
+ assert [c.kwargs["printer_name"] for c in sent.await_args_list] == ["Second"]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_switching_alerts_off_and_on_forgets_the_old_crossing(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session, power_alert_high=None, power_alert_low=10.0)
|
|
|
+ plug_id = plug.id
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+ reader = AsyncMock(return_value=("ON", 3.0))
|
|
|
+
|
|
|
+ async def set_enabled(enabled: bool):
|
|
|
+ row = (await db_session.execute(select(SmartPlug).where(SmartPlug.id == plug_id))).scalar_one()
|
|
|
+ row.power_alert_enabled = enabled
|
|
|
+ row.power_alert_last_triggered = utcnow_naive() - timedelta(minutes=10)
|
|
|
+ await db_session.commit()
|
|
|
+
|
|
|
+ with patch("backend.app.services.power_alerts.read_plug", reader):
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ await set_enabled(False)
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ await set_enabled(True)
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+
|
|
|
+ assert sent.await_count == 2
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unknown_switch_state_raises_no_low_alert_but_still_high(self, db_session, sent):
|
|
|
+ plug = await _plug(db_session, power_alert_high=200.0, power_alert_low=10.0)
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+
|
|
|
+ await monitor.evaluate(plug, None, 0.0, db_session)
|
|
|
+ sent.assert_not_awaited()
|
|
|
+
|
|
|
+ await monitor.evaluate(plug, None, 250.0, db_session)
|
|
|
+ sent.assert_awaited_once()
|
|
|
+ assert sent.await_args.kwargs["error_type"] == "Power High"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_an_unreachable_plug_is_polled_again_only_after_a_pause(self, db_session, sent):
|
|
|
+ await _plug(db_session)
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+ reader = AsyncMock(return_value=(None, None))
|
|
|
+ clock = {"now": 1000.0}
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.power_alerts.read_plug", reader),
|
|
|
+ patch("backend.app.services.power_alerts.time.monotonic", side_effect=lambda: clock["now"]),
|
|
|
+ ):
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ clock["now"] += 60
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ assert reader.await_count == 1
|
|
|
+
|
|
|
+ clock["now"] += 600
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ assert reader.await_count == 2
|
|
|
+
|
|
|
+
|
|
|
+class TestReviewRound2:
|
|
|
+ def test_a_fresh_mqtt_subscription_can_be_asked_if_reachable(self):
|
|
|
+ """Its default last_seen had no time zone, so the check raised TypeError
|
|
|
+ until the first message arrived."""
|
|
|
+ from backend.app.services.mqtt_smart_plug import MQTTSmartPlugService, SmartPlugMQTTData
|
|
|
+
|
|
|
+ service = MQTTSmartPlugService()
|
|
|
+ service.plug_data[7] = SmartPlugMQTTData(plug_id=7)
|
|
|
+
|
|
|
+ assert service.is_reachable(7) is True
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_an_mqtt_plug_without_data_is_not_paused(self, db_session, sent):
|
|
|
+ """MQTT is read from memory, so there is no network call to spare; a plug
|
|
|
+ that hasn't published yet must be checked again on the next pass."""
|
|
|
+ await _plug(db_session, plug_type="mqtt")
|
|
|
+ monitor = PowerAlertMonitor()
|
|
|
+ reader = AsyncMock(side_effect=[(None, None), ("ON", 250.0)])
|
|
|
+
|
|
|
+ with patch("backend.app.services.power_alerts.read_plug", reader):
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+ await check_all_plugs(db_session, monitor)
|
|
|
+
|
|
|
+ assert reader.await_count == 2
|
|
|
+ sent.assert_awaited_once()
|