39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
import pytest
|
|
|
|
from handlers.mqtt_handler import TalkingQMQTTService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alarm_event_response_includes_v2_audio_url(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
|
|
coro.close()
|
|
|
|
monkeypatch.setattr(service, "_publish", fake_publish)
|
|
monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence)
|
|
monkeypatch.setattr("handlers.mqtt_handler.settings.server_host", "101.35.224.118")
|
|
monkeypatch.setattr("handlers.mqtt_handler.settings.server_port", 8080)
|
|
|
|
await service._handle_alarm_report("TalkingQ_XQSN00001005", {"msg_id": "010"})
|
|
|
|
assert published == [
|
|
(
|
|
"device/TalkingQ_XQSN00001005/event_resp",
|
|
{
|
|
"msg_id": "010",
|
|
"status": "success",
|
|
"type": 0,
|
|
"params": {
|
|
"url": "http://101.35.224.118:8080/assets/audio/message_ok.mp3",
|
|
},
|
|
},
|
|
)
|
|
]
|