支持设备定时定位上报与OTA通道选择

talkingq-url/tests/test_ota_channel_selection.py

@@ -0,0 +1,201 @@
+import pytest
+from fastapi import HTTPException
+
+from banban.service.device import DeviceService
+from banban.service.ota_firmware import OtaFirmwareSelection
+from banban.service.device_setting import DeviceSettingService
+
+
+@pytest.mark.asyncio
+async def test_firmware_status_selects_gps_channel(monkeypatch):
+    service = DeviceService()
+
+    async def fake_get_device_status(**kwargs):
+        del kwargs
+        return {
+            "device_id": "TalkingQ_XQSN00001005",
+            "version": "1.0.0",
+            "device_type": 2,
+            "ota_channel": None,
+        }
+
+    async def fake_select_latest_for_device(status_row):
+        assert status_row["device_type"] == 2
+        return OtaFirmwareSelection(
+            channel_key="gps",
+            version="1.1.0",
+            firmware_url="https://example.com/gps-1.1.0.bin",
+            firmware_id=10,
+        )
+
+    async def fake_get_firmware_update_dict(device_id):
+        del device_id
+        return None
+
+    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)
+
+    result = await service.get_firmware_status(device_id="TalkingQ_XQSN00001005", user_id=1)
+
+    assert result["ota_channel"] == "gps"
+    assert result["firmware_id"] == 10
+    assert result["latest_version"] == "1.1.0"
+    assert result["firmware_url"] == "https://example.com/gps-1.1.0.bin"
+    assert result["update_available"] is True
+    assert result["can_update"] is True
+
+
+@pytest.mark.asyncio
+async def test_firmware_status_unknown_channel_cannot_update(monkeypatch):
+    service = DeviceService()
+
+    async def fake_get_device_status(**kwargs):
+        del kwargs
+        return {
+            "device_id": "TalkingQ_XQSN00001005",
+            "version": "1.0.0",
+            "device_type": None,
+            "ota_channel": None,
+        }
+
+    async def fake_select_latest_for_device(status_row):
+        assert status_row["device_type"] is None
+        return None
+
+    async def fake_get_firmware_update_dict(device_id):
+        del device_id
+        return None
+
+    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)
+
+    result = await service.get_firmware_status(device_id="TalkingQ_XQSN00001005", user_id=1)
+
+    assert result["ota_channel"] is None
+    assert result["latest_version"] is None
+    assert result["can_update"] is False
+
+
+@pytest.mark.asyncio
+async def test_start_firmware_update_records_channel_and_keeps_mqtt_payload(monkeypatch):
+    service = DeviceService()
+    calls = []
+
+    async def fake_get_firmware_status(**kwargs):
+        del kwargs
+        return {
+            "device_id": "TalkingQ_XQSN00001005",
+            "current_version": "1.0.0",
+            "latest_version": "1.1.0",
+            "ota_channel": "no_gps",
+            "firmware_id": 22,
+            "firmware_url": "https://example.com/no-gps-1.1.0.bin",
+            "update_available": True,
+            "can_update": True,
+            "update_status": "idle",
+            "progress": 0.0,
+            "target_version": None,
+            "updated_at": None,
+        }
+
+    async def fake_update_firmware_update(**kwargs):
+        calls.append(("record", kwargs))
+        return True
+
+    class FakeMqttService:
+        async def send_ota_command(self, device_id, url, version):
+            calls.append(("mqtt", {"device_id": device_id, "url": url, "version": version}))
+            return "003"
+
+    async def fake_get_instance():
+        return FakeMqttService()
+
+    monkeypatch.setattr(service, "get_firmware_status", fake_get_firmware_status)
+    monkeypatch.setattr("banban.service.device.device_firmware_update_manager.update_firmware_update", fake_update_firmware_update)
+    monkeypatch.setattr("handlers.mqtt_handler.TalkingQMQTTService.get_instance", fake_get_instance)
+
+    result = await service.start_firmware_update(device_id="TalkingQ_XQSN00001005", user_id=1)
+
+    assert calls == [
+        (
+            "record",
+            {
+                "device_id": "TalkingQ_XQSN00001005",
+                "firmware_version": "1.1.0",
+                "update_status": "sent",
+                "progress": 0.0,
+                "ota_channel": "no_gps",
+                "target_version": "1.1.0",
+                "firmware_url": "https://example.com/no-gps-1.1.0.bin",
+                "firmware_id": 22,
+                "source": "manual",
+            },
+        ),
+        (
+            "mqtt",
+            {
+                "device_id": "TalkingQ_XQSN00001005",
+                "url": "https://example.com/no-gps-1.1.0.bin",
+                "version": "1.1.0",
+            },
+        ),
+    ]
+    assert result["msg_id"] == "003"
+    assert result["ota_channel"] == "no_gps"
+    assert result["target_version"] == "1.1.0"
+
+
+@pytest.mark.asyncio
+async def test_start_firmware_update_without_known_channel_is_blocked(monkeypatch):
+    service = DeviceService()
+
+    async def fake_get_firmware_status(**kwargs):
+        del kwargs
+        return {
+            "device_id": "TalkingQ_XQSN00001005",
+            "current_version": "1.0.0",
+            "latest_version": None,
+            "update_available": False,
+            "can_update": False,
+            "update_status": "idle",
+            "progress": 0.0,
+            "target_version": None,
+            "updated_at": None,
+        }
+
+    monkeypatch.setattr(service, "get_firmware_status", fake_get_firmware_status)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await service.start_firmware_update(device_id="TalkingQ_XQSN00001005", user_id=1)
+
+    assert exc_info.value.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_device_type_update_preserves_existing_manual_ota_channel(monkeypatch):
+    service = DeviceSettingService()
+    calls = []
+
+    async def fake_get_setting_by_device_id(device_id):
+        assert device_id == "TalkingQ_XQSN00001005"
+        return {"device_id": device_id, "ota_channel": "factory_gps"}
+
+    async def fake_update_setting(**kwargs):
+        calls.append(kwargs)
+
+    monkeypatch.setattr(service, "get_setting_by_device_id", fake_get_setting_by_device_id)
+    monkeypatch.setattr(service, "update_setting", fake_update_setting)
+
+    await service.insert_or_update(
+        device_id="TalkingQ_XQSN00001005",
+        power=80,
+        volume=60,
+        signal_strength=4,
+        version_str="1.0.0",
+        device_type=1,
+    )
+
+    assert calls[0]["device_type"] == 1
+    assert calls[0]["ota_channel"] is None