bambuddy/backend/tests/unit/test_location_ha_sensor_manager_2824.py

import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest

from backend.app.services.location_ha_sensor_manager import LocationHASensorManager


def _sensor(**overrides):
    base = {
        "id": 1,
        "location_id": 7,
        "name": "Drybox Humidity",
        "entity_id": "sensor.drybox_humidity",
        "kind": "numeric",
        "device_class": "humidity",
        "unit": "%",
        "alert_state": None,
        "alert_above": 60,
        "alert_below": None,
        "notify_on_alert": False,
        "last_state": None,
    }
    base.update(overrides)
    return SimpleNamespace(**base)


class TestNotificationEdge:
    async def _apply(self, manager, sensor, states, notify):
        db = AsyncMock()
        db.get.return_value = SimpleNamespace(name="Drybox 1")
        with patch("backend.app.services.notification_service.notification_service", notify):
            await manager._apply(db, [sensor], states)

    async def test_fires_once_on_the_way_in(self):
        manager = LocationHASensorManager()
        sensor = _sensor(notify_on_alert=True)
        notify = AsyncMock()

        await self._apply(manager, sensor, {sensor.entity_id: {"state": "40"}}, notify)
        assert notify.on_location_ha_sensor_alert.await_count == 0

        await self._apply(manager, sensor, {sensor.entity_id: {"state": "65"}}, notify)
        assert notify.on_location_ha_sensor_alert.await_count == 1

        await self._apply(manager, sensor, {sensor.entity_id: {"state": "66"}}, notify)
        assert notify.on_location_ha_sensor_alert.await_count == 1

    async def test_silent_on_the_first_poll_after_a_restart(self):
        manager = LocationHASensorManager()
        sensor = _sensor(notify_on_alert=True)
        notify = AsyncMock()

        await self._apply(manager, sensor, {sensor.entity_id: {"state": "65"}}, notify)

        assert notify.on_location_ha_sensor_alert.await_count == 0
        assert manager.get_reading(sensor.id).alerting is True

    async def test_silent_when_the_sensor_opts_out(self):
        manager = LocationHASensorManager()
        sensor = _sensor(notify_on_alert=False)
        notify = AsyncMock()

        await self._apply(manager, sensor, {sensor.entity_id: {"state": "40"}}, notify)
        await self._apply(manager, sensor, {sensor.entity_id: {"state": "65"}}, notify)

        assert notify.on_location_ha_sensor_alert.await_count == 0

    async def test_passes_the_location_name_not_a_printer_name(self):
        manager = LocationHASensorManager()
        sensor = _sensor(notify_on_alert=True)
        notify = AsyncMock()

        await self._apply(manager, sensor, {sensor.entity_id: {"state": "40"}}, notify)
        await self._apply(manager, sensor, {sensor.entity_id: {"state": "65"}}, notify)

        notify.on_location_ha_sensor_alert.assert_awaited_once()
        _, kwargs = notify.on_location_ha_sensor_alert.await_args
        assert kwargs["location_name"] == "Drybox 1"
        assert kwargs["sensor_name"] == "Drybox Humidity"

    async def test_a_dropout_does_not_count_as_the_alert_clearing(self):
        manager = LocationHASensorManager()
        sensor = _sensor(notify_on_alert=True)
        notify = AsyncMock()

        await self._apply(manager, sensor, {sensor.entity_id: {"state": "40"}}, notify)
        await self._apply(manager, sensor, {sensor.entity_id: {"state": "65"}}, notify)
        await self._apply(manager, sensor, {sensor.entity_id: None}, notify)
        await self._apply(manager, sensor, {sensor.entity_id: {"state": "66"}}, notify)

        assert notify.on_location_ha_sensor_alert.await_count == 1


class TestLastStatePersistence:
    """What goes into last_state must fit its String(64) column.

    A numeric entity can start reporting free text longer than the column.
    SQLite stores it anyway, but PostgreSQL rejects the row — and since
    _apply commits the whole pass at once, one such sensor would sink every
    sensor's update on every tick.
    """

    async def _apply(self, manager, sensor, state):
        db = AsyncMock()
        with patch("backend.app.services.notification_service.notification_service", AsyncMock()):
            await manager._apply(db, [sensor], {sensor.entity_id: {"state": state}})

    async def test_a_long_text_state_is_cut_to_the_column_width(self):
        manager = LocationHASensorManager()
        sensor = _sensor(last_changed=None, last_checked=None)
        long_state = "x" * 500

        await self._apply(manager, sensor, long_state)

        assert sensor.last_state == "x" * 64
        # The cache keeps the full state — only what is persisted is cut.
        assert manager.get_reading(sensor.id).state == long_state

    async def test_an_unchanged_long_state_is_not_a_change_on_every_poll(self):
        manager = LocationHASensorManager()
        sensor = _sensor(last_changed=None, last_checked=None)
        long_state = "x" * 500

        await self._apply(manager, sensor, long_state)
        first_changed = sensor.last_changed
        await self._apply(manager, sensor, long_state)

        # Comparing the stored (cut) value against the raw state would see a
        # difference on every poll and churn last_changed forever.
        assert sensor.last_changed == first_changed


class TestPollLoopSurvival:
    """The loop must outlive a transient database error (#2824 review).

    _get_poll_interval() reads Settings, so the sleep leg of _poll_loop does
    I/O and can raise on pool exhaustion or a restarting database. Letting
    that escape ends the task permanently: stop() is what clears _task, so a
    self-terminated loop leaves it set and start() refuses to restart it.
    """

    async def test_survives_a_failing_poll_interval_lookup(self):
        manager = LocationHASensorManager()
        polls = 0

        async def counting_poll():
            nonlocal polls
            polls += 1
            if polls >= 2:
                raise asyncio.CancelledError  # end the loop once we've proven it came back

        async def failing_interval():
            raise RuntimeError("QueuePool limit reached")

        with (
            patch.object(manager, "poll_once", counting_poll),
            patch.object(manager, "_get_poll_interval", failing_interval),
            patch("backend.app.services.location_ha_sensor_manager.POLL_INTERVAL", 0),
        ):
            await manager._poll_loop()

        # Without the guard the first lookup failure escapes _poll_loop and
        # poll_once never runs a second time.
        assert polls == 2

    async def test_a_cancel_during_the_fallback_sleep_still_stops_the_loop(self):
        manager = LocationHASensorManager()

        async def failing_interval():
            raise RuntimeError("database is locked")

        with (
            patch.object(manager, "poll_once", AsyncMock()),
            patch.object(manager, "_get_poll_interval", failing_interval),
            patch("backend.app.services.location_ha_sensor_manager.POLL_INTERVAL", 3600),
        ):
            task = asyncio.create_task(manager._poll_loop())
            await asyncio.sleep(0)  # let it reach the fallback sleep
            task.cancel()
            with pytest.raises(asyncio.CancelledError):
                await task