Miayui
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SERVER_PATH = ROOT / "apps" / "ir-toy-console" / "server.py"
|
|
SPEC = importlib.util.spec_from_file_location("ir_toy_console_server_timing_test", SERVER_PATH)
|
|
assert SPEC and SPEC.loader
|
|
SERVER = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = SERVER
|
|
SPEC.loader.exec_module(SERVER)
|
|
|
|
|
|
class IRToyConsoleTimingTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.catalog = [
|
|
{
|
|
"id": "cap-power",
|
|
"status": "learned",
|
|
"device_id": "cactus",
|
|
"device_name": "跳舞仙人掌玩具",
|
|
"action": "电源开关",
|
|
"risk": "low",
|
|
},
|
|
{
|
|
"id": "cap-song",
|
|
"status": "learned",
|
|
"device_id": "cactus",
|
|
"device_name": "跳舞仙人掌玩具",
|
|
"action": "儿歌",
|
|
"risk": "low",
|
|
},
|
|
]
|
|
|
|
@patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False)
|
|
def test_delay_is_required_and_preserved_in_normalized_plan(self) -> None:
|
|
proposed = {
|
|
"summary": "先开机，5 秒后播放儿歌",
|
|
"needs_confirmation": False,
|
|
"question": "",
|
|
"steps": [
|
|
{
|
|
"kind": "peripheral",
|
|
"capability_id": "cap-power",
|
|
"command": "",
|
|
"label": "仙人掌·电源开关",
|
|
"delay_before_ms": 0,
|
|
},
|
|
{
|
|
"kind": "peripheral",
|
|
"capability_id": "cap-song",
|
|
"command": "",
|
|
"label": "仙人掌·儿歌",
|
|
"delay_before_ms": 5000,
|
|
},
|
|
],
|
|
}
|
|
with patch.object(
|
|
SERVER, "request_gpt_stream", return_value=json.dumps(proposed, ensure_ascii=False)
|
|
) as request_gpt:
|
|
result = SERVER.plan_with_gpt(
|
|
{"utterance": "开启仙人掌，5秒后播放儿歌", "peripherals": self.catalog}
|
|
)
|
|
|
|
self.assertEqual([0, 5000], [step["delay_before_ms"] for step in result["steps"]])
|
|
request_payload = request_gpt.call_args.args[0]
|
|
step_schema = request_payload["text"]["format"]["schema"]["properties"]["steps"][
|
|
"items"
|
|
]
|
|
self.assertIn("delay_before_ms", step_schema["required"])
|
|
self.assertEqual(60_000, step_schema["properties"]["delay_before_ms"]["maximum"])
|
|
|
|
@patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False)
|
|
def test_out_of_range_delay_is_rejected(self) -> None:
|
|
proposed = {
|
|
"summary": "等待时间超限",
|
|
"needs_confirmation": False,
|
|
"question": "",
|
|
"steps": [
|
|
{
|
|
"kind": "peripheral",
|
|
"capability_id": "cap-power",
|
|
"command": "",
|
|
"label": "仙人掌·电源开关",
|
|
"delay_before_ms": 60_001,
|
|
}
|
|
],
|
|
}
|
|
with patch.object(
|
|
SERVER, "request_gpt_stream", return_value=json.dumps(proposed, ensure_ascii=False)
|
|
):
|
|
with self.assertRaisesRegex(SERVER.IntegrationError, "单步等待时间"):
|
|
SERVER.plan_with_gpt({"utterance": "稍后开机", "peripherals": self.catalog})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|