fix(ota): settle stale firmware update states

talkingq-url/tests/test_ota_channel_selection.py

@@ -1,3 +1,5 @@
+from datetime import datetime, timedelta
+
 import pytest
 from fastapi import HTTPException
 
@@ -78,6 +80,143 @@ async def test_firmware_status_unknown_channel_cannot_update(monkeypatch):
     assert result["can_update"] is False
 
 
+@pytest.mark.asyncio
+async def test_firmware_status_settles_active_update_when_target_is_installed(monkeypatch):
+    service = DeviceService()
+    calls = []
+    update_record = {
+        "device_id": "TalkingQ_ZHWK00003111",
+        "firmware_version": "1.8.3",
+        "target_version": "1.8.3",
+        "firmware_url": "https://example.com/gps-1.8.3.bin",
+        "firmware_id": 12,
+        "update_status": "accepted",
+        "progress": 0.0,
+        "updated_at": datetime.now(),
+    }
+
+    async def fake_get_device_status(**kwargs):
+        del kwargs
+        return {
+            "device_id": "TalkingQ_ZHWK00003111",
+            "version": "1.8.3",
+            "device_type": 2,
+            "ota_channel": "gps",
+        }
+
+    async def fake_select_latest_for_device(status_row):
+        del status_row
+        return OtaFirmwareSelection(
+            channel_key="gps",
+            version="1.8.4",
+            firmware_url="https://example.com/gps-1.8.4.bin",
+            firmware_id=13,
+        )
+
+    async def fake_get_firmware_update_dict(device_id):
+        assert device_id == "TalkingQ_ZHWK00003111"
+        return update_record
+
+    async def fake_update_firmware_update(**kwargs):
+        calls.append(kwargs)
+        return True
+
+    monkeypatch.setattr(service, "get_device_status", fake_get_device_status)
+    monkeypatch.setattr("banban.service.device.ota_firmware_service.select_latest_for_device", fake_select_latest_for_device)
+    monkeypatch.setattr("banban.service.device.device_firmware_update_manager.get_firmware_update_dict", fake_get_firmware_update_dict)
+    monkeypatch.setattr("banban.service.device.device_firmware_update_manager.update_firmware_update", fake_update_firmware_update)
+
+    result = await service.get_firmware_status(device_id="TalkingQ_ZHWK00003111", user_id=1)
+
+    assert result["update_status"] == "success"
+    assert result["progress"] == 100.0
+    assert result["update_available"] is True
+    assert calls == [{
+        "device_id": "TalkingQ_ZHWK00003111",
+        "firmware_version": "1.8.3",
+        "update_status": "success",
+        "progress": 100.0,
+    }]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("active_status", ["sent", "accepted", "updating"])
+async def test_firmware_status_times_out_stale_active_update(monkeypatch, active_status):
+    service = DeviceService()
+    calls = []
+    update_record = {
+        "device_id": "TalkingQ_ZHWK00003111",
+        "firmware_version": "1.8.3",
+        "target_version": "1.8.3",
+        "firmware_url": "https://example.com/gps-1.8.3.bin",
+        "firmware_id": 12,
+        "update_status": active_status,
+        "progress": 42.0,
+        "updated_at": datetime.now() - timedelta(seconds=301),
+    }
+
+    async def fake_get_device_status(**kwargs):
+        del kwargs
+        return {
+            "device_id": "TalkingQ_ZHWK00003111",
+            "version": "1.8.2",
+            "device_type": 2,
+            "ota_channel": "gps",
+        }
+
+    async def fake_select_latest_for_device(status_row):
+        del status_row
+        return OtaFirmwareSelection(
+            channel_key="gps",
+            version="1.8.4",
+            firmware_url="https://example.com/gps-1.8.4.bin",
+            firmware_id=13,
+        )
+
+    async def fake_get_firmware_update_dict(device_id):
+        assert device_id == "TalkingQ_ZHWK00003111"
+        return update_record
+
+    async def fake_update_firmware_update(**kwargs):
+        calls.append(kwargs)
+        return True
+
+    monkeypatch.setattr(service, "get_device_status", fake_get_device_status)
+    monkeypatch.setattr("banban.service.device.ota_firmware_service.select_latest_for_device", fake_select_latest_for_device)
+    monkeypatch.setattr("banban.service.device.device_firmware_update_manager.get_firmware_update_dict", fake_get_firmware_update_dict)
+    monkeypatch.setattr("banban.service.device.device_firmware_update_manager.update_firmware_update", fake_update_firmware_update)
+
+    result = await service.get_firmware_status(device_id="TalkingQ_ZHWK00003111", user_id=1)
+
+    assert result["update_status"] == "failed"
+    assert result["progress"] == 0.0
+    assert result["can_update"] is True
+    assert calls == [{
+        "device_id": "TalkingQ_ZHWK00003111",
+        "firmware_version": "1.8.3",
+        "update_status": "failed",
+        "progress": 0.0,
+    }]
+
+
+def test_firmware_update_not_complete_until_target_version_is_reported():
+    service = DeviceService()
+
+    assert not service._firmware_update_reached_target("1.8.2", "1.8.3")
+    assert not service._firmware_update_reached_target("unknown", "1.8.3")
+    assert not service._firmware_update_reached_target("1.8.4", "unknown")
+    assert service._firmware_update_reached_target("1.8.3", "1.8.3")
+    assert service._firmware_update_reached_target("1.8.4", "1.8.3")
+
+
+def test_firmware_update_timeout_does_not_fail_recent_progress(monkeypatch):
+    service = DeviceService()
+    monkeypatch.setattr("banban.service.device.settings.ota_sent_timeout_seconds", 300)
+
+    assert not service._firmware_update_timed_out(datetime.now())
+    assert service._firmware_update_timed_out(datetime.now() - timedelta(seconds=301))
+
+
 @pytest.mark.asyncio
 async def test_start_firmware_update_records_channel_and_keeps_mqtt_payload(monkeypatch):
     service = DeviceService()