banban/talkingq-url/tests/test_ota_channel_selection.py

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