114 lines
3.0 KiB
Python
114 lines
3.0 KiB
Python
|
import asyncio
|
||
|
|
import json
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from hass.typing import XiaoAiConfig
|
||
|
|
from hass.xiaoai_bridge import XiaoAiBridge
|
||
|
|
|
||
|
|
|
||
|
|
class FakeServer:
|
||
|
|
def __init__(self):
|
||
|
|
self._fns = {}
|
||
|
|
self.shell_calls = []
|
||
|
|
|
||
|
|
def register_fn(self, name, fn):
|
||
|
|
self._fns[name] = fn
|
||
|
|
|
||
|
|
async def run_shell(self, script, timeout):
|
||
|
|
self.shell_calls.append((script, timeout))
|
||
|
|
return '{"stdout":"ok","stderr":"","exit_code":0}'
|
||
|
|
|
||
|
|
async def start_server(self):
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bridge_interrupts_before_dispatch(monkeypatch):
|
||
|
|
fake_server = FakeServer()
|
||
|
|
|
||
|
|
import hass.xiaoai_bridge as module
|
||
|
|
|
||
|
|
monkeypatch.setattr(module, "_LOGGER", module._LOGGER)
|
||
|
|
|
||
|
|
cfg = XiaoAiConfig(interrupt_on_new_input=True, shell_timeout_ms=123)
|
||
|
|
bridge = XiaoAiBridge(cfg)
|
||
|
|
bridge._open_xiaoai_server = fake_server
|
||
|
|
bridge._loop = asyncio.get_running_loop()
|
||
|
|
|
||
|
|
called = []
|
||
|
|
|
||
|
|
async def on_text(input_data):
|
||
|
|
called.append(input_data)
|
||
|
|
|
||
|
|
bridge.set_on_text(on_text)
|
||
|
|
bridge._ready.set()
|
||
|
|
bridge._register_callbacks()
|
||
|
|
|
||
|
|
event = {
|
||
|
|
"event": "instruction",
|
||
|
|
"data": {
|
||
|
|
"ClientIP": "192.168.1.101",
|
||
|
|
"NewLine": json.dumps(
|
||
|
|
{
|
||
|
|
"header": {"namespace": "SpeechRecognizer", "name": "RecognizeResult"},
|
||
|
|
"payload": {
|
||
|
|
"is_final": True,
|
||
|
|
"results": [{"text": "打开客厅灯"}],
|
||
|
|
},
|
||
|
|
},
|
||
|
|
ensure_ascii=False,
|
||
|
|
),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
fake_server._fns["on_event"](json.dumps(event, ensure_ascii=False))
|
||
|
|
await asyncio.sleep(0.05)
|
||
|
|
|
||
|
|
assert len(fake_server.shell_calls) == 1
|
||
|
|
assert "ubus call mibrain ai_service" in fake_server.shell_calls[0][0]
|
||
|
|
assert len(called) == 1
|
||
|
|
assert called[0].text == "打开客厅灯"
|
||
|
|
assert called[0].client_ip == "192.168.1.101"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_bridge_extracts_missing_ip_as_none():
|
||
|
|
fake_server = FakeServer()
|
||
|
|
cfg = XiaoAiConfig(interrupt_on_new_input=False)
|
||
|
|
bridge = XiaoAiBridge(cfg)
|
||
|
|
bridge._open_xiaoai_server = fake_server
|
||
|
|
bridge._loop = asyncio.get_running_loop()
|
||
|
|
|
||
|
|
captured = []
|
||
|
|
|
||
|
|
async def on_text(input_data):
|
||
|
|
captured.append(input_data)
|
||
|
|
|
||
|
|
bridge.set_on_text(on_text)
|
||
|
|
bridge._ready.set()
|
||
|
|
bridge._register_callbacks()
|
||
|
|
|
||
|
|
event = {
|
||
|
|
"event": "instruction",
|
||
|
|
"data": {
|
||
|
|
"NewLine": json.dumps(
|
||
|
|
{
|
||
|
|
"header": {"namespace": "SpeechRecognizer", "name": "RecognizeResult"},
|
||
|
|
"payload": {
|
||
|
|
"is_final": True,
|
||
|
|
"results": [{"text": "查询天气"}],
|
||
|
|
},
|
||
|
|
},
|
||
|
|
ensure_ascii=False,
|
||
|
|
),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
fake_server._fns["on_event"](json.dumps(event, ensure_ascii=False))
|
||
|
|
await asyncio.sleep(0.05)
|
||
|
|
|
||
|
|
assert len(captured) == 1
|
||
|
|
assert captured[0].client_ip is None
|
||
|
|
assert fake_server.shell_calls == []
|