banban/talkingq-url/tests/test_alarm_gps_location_refresh.py

import pytest

from handlers.mqtt_handler import TalkingQMQTTService


@pytest.mark.asyncio
async def test_alarm_requests_fresh_gps_before_sms(monkeypatch):
    service = TalkingQMQTTService({"broker": "127.0.0.1"})
    published = []
    calls = []

    async def fake_publish(topic, payload):
        published.append((topic, payload))
        return True

    async def fake_schedule_persistence(device_id, label, coro):
        calls.append(("schedule", label))
        await coro

    async def fake_record_alarm_event(**kwargs):
        calls.append(("record_alarm", kwargs))
        return 123

    async def fake_query_gps_and_wait_for_location(device_id, *, timeout_seconds):
        calls.append(("query_gps", device_id, timeout_seconds))
        return {
            "child_id": 9,
            "device_id": device_id,
            "coord_type": "gcj02",
            "lat": 30.245,
            "lng": 120.215,
            "updated_at": "2026-06-08 12:00:00",
            "address": "浙江省杭州市萧山区扬帆路附近",
        }

    async def fake_apply_alarm_location(**kwargs):
        calls.append(("apply_alarm_location", kwargs))
        return True

    def fake_schedule_alarm_notification(**kwargs):
        calls.append(("sms", kwargs))

    monkeypatch.setattr(service, "_publish", fake_publish)
    monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence)
    monkeypatch.setattr(service, "query_gps_and_wait_for_location", fake_query_gps_and_wait_for_location)
    monkeypatch.setattr("handlers.mqtt_handler.settings.alarm_gps_query_timeout_seconds", 2.5)
    monkeypatch.setattr("handlers.mqtt_handler.device_alarm_service.record_alarm_event", fake_record_alarm_event)
    monkeypatch.setattr("handlers.mqtt_handler.device_alarm_service.apply_alarm_location", fake_apply_alarm_location)
    monkeypatch.setattr("handlers.mqtt_handler.sms_notification_service.schedule_alarm_notification", fake_schedule_alarm_notification)

    await service._handle_alarm_report("TalkingQ_device001", {"msg_id": "010"})

    assert published == [
        (
            "device/TalkingQ_device001/event_resp",
            {
                "msg_id": "010",
                "status": "success",
                "type": 0,
                "params": {
                    "url": "http://192.168.101.78:8080/assets/audio/message_ok.mp3",
                },
            },
        ),
    ]
    assert calls == [
        ("schedule", "alarm_event"),
        (
            "record_alarm",
            {
                "device_id": "TalkingQ_device001",
                "source_msg_id": "010",
                "resolve_address": False,
            },
        ),
        ("query_gps", "TalkingQ_device001", 2.5),
        (
            "apply_alarm_location",
            {
                "alarm_id": 123,
                "device_id": "TalkingQ_device001",
                "location": {
                    "child_id": 9,
                    "device_id": "TalkingQ_device001",
                    "coord_type": "gcj02",
                    "lat": 30.245,
                    "lng": 120.215,
                    "updated_at": "2026-06-08 12:00:00",
                    "address": "浙江省杭州市萧山区扬帆路附近",
                },
            },
        ),
        ("sms", {"device_id": "TalkingQ_device001", "alarm_id": 123}),
    ]


@pytest.mark.asyncio
async def test_alarm_falls_back_to_existing_alarm_location_when_gps_times_out(monkeypatch):
    service = TalkingQMQTTService({"broker": "127.0.0.1"})
    calls = []

    async def fake_publish(topic, payload):
        del topic, payload
        return True

    async def fake_schedule_persistence(device_id, label, coro):
        calls.append(("schedule", label))
        await coro

    async def fake_record_alarm_event(**kwargs):
        calls.append(("record_alarm", kwargs))
        return 456

    async def fake_query_gps_and_wait_for_location(device_id, *, timeout_seconds):
        calls.append(("query_gps", device_id, timeout_seconds))
        return None

    async def fake_resolve_alarm_current_location(**kwargs):
        calls.append(("resolve_existing", kwargs))
        return True

    def fake_schedule_alarm_notification(**kwargs):
        calls.append(("sms", kwargs))

    monkeypatch.setattr(service, "_publish", fake_publish)
    monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence)
    monkeypatch.setattr(service, "query_gps_and_wait_for_location", fake_query_gps_and_wait_for_location)
    monkeypatch.setattr("handlers.mqtt_handler.settings.alarm_gps_query_timeout_seconds", 1.0)
    monkeypatch.setattr("handlers.mqtt_handler.device_alarm_service.record_alarm_event", fake_record_alarm_event)
    monkeypatch.setattr("handlers.mqtt_handler.device_alarm_service.resolve_alarm_current_location", fake_resolve_alarm_current_location)
    monkeypatch.setattr("handlers.mqtt_handler.sms_notification_service.schedule_alarm_notification", fake_schedule_alarm_notification)

    await service._handle_alarm_report("TalkingQ_device001", {"msg_id": "010"})

    assert calls == [
        ("schedule", "alarm_event"),
        (
            "record_alarm",
            {
                "device_id": "TalkingQ_device001",
                "source_msg_id": "010",
                "resolve_address": False,
            },
        ),
        ("query_gps", "TalkingQ_device001", 1.0),
        ("resolve_existing", {"alarm_id": 456, "device_id": "TalkingQ_device001"}),
        ("sms", {"device_id": "TalkingQ_device001", "alarm_id": 456}),
    ]


@pytest.mark.asyncio
async def test_gps_waiter_returns_persisted_location(monkeypatch):
    service = TalkingQMQTTService({"broker": "127.0.0.1"})
    published = []

    async def fake_publish(topic, payload):
        published.append((topic, payload))
        return True

    async def fake_schedule_persistence(device_id, label, coro):
        del device_id, label
        await coro

    async def fake_report_mqtt_device_location(**kwargs):
        return {
            "child_id": 9,
            "device_id": kwargs["device_id"],
            "lat": kwargs["latitude"],
            "lng": kwargs["longitude"],
            "updated_at": "2026-06-08 12:00:00",
        }

    monkeypatch.setattr(service, "_publish", fake_publish)
    monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence)
    monkeypatch.setattr("handlers.mqtt_handler.location_service.report_mqtt_device_location", fake_report_mqtt_device_location)

    wait_task = __import__("asyncio").create_task(
        service.query_gps_and_wait_for_location("TalkingQ_device001", timeout_seconds=1.0)
    )
    await __import__("asyncio").sleep(0)

    await service._handle_gps_response(
        "TalkingQ_device001",
        {
            "msg_id": "001",
            "status": "success",
            "data": {
                "latitude": 30.245,
                "longitude": 120.215,
            },
        },
    )

    location = await wait_task

    assert published == [
        ("device/TalkingQ_device001/command", {"msg_id": "001"}),
    ]
    assert location["lat"] == 30.245
    assert location["lng"] == 120.215