完善设备端体验与微信校验支持

talkingq-url/tests/test_alarm_gps_location_refresh.py

@@ -1,8 +1,18 @@
 import pytest
 
+from banban.routers.devices import _row_to_current_location_response
+from banban.service.device_alarm import DeviceAlarmService
+from banban.service.location import LocationService, is_valid_coordinate_pair
 from handlers.mqtt_handler import TalkingQMQTTService
 
 
+def test_zero_zero_coordinate_is_invalid():
+    assert is_valid_coordinate_pair(23.2413923, 113.6051866) is True
+    assert is_valid_coordinate_pair(0, 0) is False
+    assert is_valid_coordinate_pair("0.0000000", "0.0000000") is False
+    assert is_valid_coordinate_pair(None, 113.6051866) is False
+
+
 @pytest.mark.asyncio
 async def test_alarm_requests_fresh_gps_before_sms(monkeypatch):
     service = TalkingQMQTTService({"broker": "127.0.0.1"})
@@ -58,7 +68,7 @@ async def test_alarm_requests_fresh_gps_before_sms(monkeypatch):
                 "status": "success",
                 "type": 0,
                 "params": {
-                    "url": "http://192.168.101.78:8080/assets/audio/message_ok.mp3",
+                    "url": "http://192.168.101.78:8080/assets/roles/banban/warn_parent_zh.mp3",
                 },
             },
         ),
@@ -94,6 +104,125 @@ async def test_alarm_requests_fresh_gps_before_sms(monkeypatch):
     ]
 
 
+@pytest.mark.asyncio
+async def test_alarm_ignores_invalid_zero_zero_location(monkeypatch):
+    service = DeviceAlarmService()
+    calls = []
+
+    async def fake_update_alarm_location(**kwargs):
+        calls.append(("update_alarm_location", kwargs))
+        return True
+
+    async def fake_resolve_alarm_address(**kwargs):
+        calls.append(("resolve_alarm_address", kwargs))
+
+    monkeypatch.setattr(service, "update_alarm_location", fake_update_alarm_location)
+    monkeypatch.setattr(service, "resolve_alarm_address", fake_resolve_alarm_address)
+
+    updated = await service.apply_alarm_location(
+        alarm_id=7,
+        device_id="TalkingQ_device001",
+        location={
+            "coord_type": "gcj02",
+            "lat": 0,
+            "lng": 0,
+            "updated_at": "2026-06-22 10:00:00",
+        },
+    )
+
+    assert updated is False
+    assert calls == []
+
+
+@pytest.mark.asyncio
+async def test_alarm_current_location_reuses_existing_address(monkeypatch):
+    service = DeviceAlarmService()
+    calls = []
+
+    async def fake_get_alarm_event(**kwargs):
+        calls.append(("get_alarm_event", kwargs))
+        return {
+            "alarm_id": 7,
+            "device_id": "TalkingQ_device001",
+            "lat": 23.2413923,
+            "lng": 113.6051866,
+            "address": "广东省广州市增城区中新镇叶岭新村街",
+        }
+
+    async def fake_resolve_alarm_address(**kwargs):
+        calls.append(("resolve_alarm_address", kwargs))
+
+    monkeypatch.setattr(service, "get_alarm_event", fake_get_alarm_event)
+    monkeypatch.setattr(service, "resolve_alarm_address", fake_resolve_alarm_address)
+
+    resolved = await service.resolve_alarm_current_location(
+        alarm_id=7,
+        device_id="TalkingQ_device001",
+    )
+
+    assert resolved is True
+    assert calls == [
+        ("get_alarm_event", {"alarm_id": 7, "device_id": "TalkingQ_device001"}),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_alarm_record_ignores_invalid_current_location_snapshot(monkeypatch):
+    service = DeviceAlarmService()
+    calls = []
+
+    class FakeDAO:
+        def __init__(self, session):
+            del session
+
+        async def get_active_binding_context(self, *, device_id):
+            calls.append(("binding", device_id))
+            return {"owner_user_id": 3, "child_id": 9}
+
+        async def get_current_location_snapshot(self, *, child_id):
+            calls.append(("snapshot", child_id))
+            return {
+                "coord_type": "gcj02",
+                "lat": 0,
+                "lng": 0,
+                "updated_at": "2026-06-22 10:00:00",
+                "address": None,
+            }
+
+        async def create(self, **kwargs):
+            calls.append(("create", kwargs))
+            return 7
+
+    class FakeSession:
+        async def close(self):
+            calls.append(("close",))
+
+    async def fake_get_session():
+        return FakeSession()
+
+    monkeypatch.setattr(service, "get_session", fake_get_session)
+    monkeypatch.setattr("banban.service.device_alarm.DeviceAlarmDAO", FakeDAO)
+
+    alarm_id = await service.record_alarm_event(device_id="TalkingQ_device001")
+
+    assert alarm_id == 7
+    assert calls == [
+        ("binding", "TalkingQ_device001"),
+        ("snapshot", 9),
+        (
+            "create",
+            {
+                "device_id": "TalkingQ_device001",
+                "owner_user_id": 3,
+                "child_id": 9,
+                "source_msg_id": "010",
+                "location": None,
+            },
+        ),
+        ("close",),
+    ]
+
+
 @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"})
@@ -148,6 +277,27 @@ async def test_alarm_falls_back_to_existing_alarm_location_when_gps_times_out(mo
     ]
 
 
+@pytest.mark.asyncio
+async def test_mqtt_location_report_ignores_zero_zero(monkeypatch):
+    service = LocationService()
+    sessions = []
+
+    async def fail_get_session():
+        sessions.append("called")
+        raise AssertionError("invalid location should not open a database session")
+
+    monkeypatch.setattr(service, "get_session", fail_get_session)
+
+    row = await service.report_mqtt_device_location(
+        device_id="TalkingQ_device001",
+        latitude=0,
+        longitude=0,
+    )
+
+    assert row is None
+    assert sessions == []
+
+
 @pytest.mark.asyncio
 async def test_gps_waiter_returns_persisted_location(monkeypatch):
     service = TalkingQMQTTService({"broker": "127.0.0.1"})
@@ -198,3 +348,31 @@ async def test_gps_waiter_returns_persisted_location(monkeypatch):
     ]
     assert location["lat"] == 30.245
     assert location["lng"] == 120.215
+
+
+def test_current_location_without_address_displays_pending_text():
+    row = {
+        "child_id": 9,
+        "child_name": "测试孩子",
+        "device_id": "TalkingQ_device001",
+        "coord_type": "gcj02",
+        "lat": 23.2413923,
+        "lng": 113.6051866,
+        "accuracy_m": None,
+        "altitude_m": None,
+        "speed_mps": None,
+        "heading_deg": None,
+        "source": 1,
+        "battery_pct": 80,
+        "device_time": "2026-06-22 10:00:00",
+        "server_time": "2026-06-22 10:00:01",
+        "updated_at": "2026-06-22 10:00:01",
+        "address": None,
+        "address_resolved_at": None,
+        "address_resolve_status": 2,
+    }
+
+    response = _row_to_current_location_response(row)
+
+    assert response.address == "地址解析中"
+    assert response.address_resolve_status == 2