bambuddy/backend/tests/unit/test_power_alerts.py

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