monitormysolar/tests/test_ota_snapshot_suppression.py at 62a9869450d52d95ce4f07331895a9bf80a80bbe · RAR/monitormysolar · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
"""Tests for snapshot suppression during an MQTT OTA update.

While a dongle is in OTA mode it has no snapshot queue allocated, so a
{"what":"all"} request crashes it into a reboot loop. The coordinator must
suppress every snapshot path (bootstrap, forced, and recovery) while the
update entity has marked the dongle as mid-OTA, and must not burn the
recovery debounce window while suppressed so the post-OTA refresh still
goes through.
"""
import asyncio
from unittest.mock import MagicMock


def _run(coro):
    return asyncio.run(coro)


def test_request_snapshot_suppressed_during_ota(coordinator, monkeypatch):
    from custom_components.monitormysolar import coordinator as coord_mod

    publishes = []

    async def fake_publish(hass, topic, payload, **kwargs):
        publishes.append((topic, payload))

    monkeypatch.setattr(coord_mod.mqtt, "async_publish", fake_publish)
    coordinator._snapshot_requested = set()

    coordinator.set_ota_in_progress("dongle-A", True)
    _run(coordinator.request_snapshot("dongle-A", "4.3.0.111S3", force=True))
    assert publishes == []
    # Suppression must not mark the snapshot as already-requested.
    assert "dongle-A" not in coordinator._snapshot_requested

    coordinator.set_ota_in_progress("dongle-A", False)
    _run(coordinator.request_snapshot("dongle-A", "4.3.0.111S3", force=True))
    assert publishes == [("dongle-A/snapshot/request", '{"what":"all"}')]


def test_recovery_snapshot_suppressed_and_debounce_not_burned(coordinator, monkeypatch):
    calls = []

    async def fake_request_snapshot(dongle_id, version="", force=False):
        calls.append((dongle_id, force))

    monkeypatch.setattr(coordinator, "request_snapshot", fake_request_snapshot)
    coordinator.current_fw_versions = {}
    coordinator._last_recovery_snapshot = {}
    coordinator._recovery_snapshot_debounce = 30.0

    coordinator.set_ota_in_progress("dongle-A", True)
    # Reconnect trigger fires mid-OTA (e.g. availability 'online').
    _run(coordinator.request_recovery_snapshot("dongle-A", "availability online"))
    assert calls == []
    # The debounce window must not have been consumed by the suppressed call.
    assert coordinator._last_recovery_snapshot == {}

    coordinator.set_ota_in_progress("dongle-A", False)
    _run(coordinator.request_recovery_snapshot("dongle-A", "post-ota"))
    assert calls == [("dongle-A", True)]


def test_ota_flag_is_per_dongle(coordinator):
    coordinator.set_ota_in_progress("dongle-A", True)
    assert coordinator.is_ota_in_progress("dongle-A")
    assert not coordinator.is_ota_in_progress("dongle-B")
    # Clearing an unset dongle is a no-op, not an error.
    coordinator.set_ota_in_progress("dongle-B", False)
    coordinator.set_ota_in_progress("dongle-A", False)
    assert not coordinator.is_ota_in_progress("dongle-A")