137 lines
4.9 KiB
Python
137 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""对话 API 测试（issue #77）。
|
||
|
||
覆盖：
|
||
1. dispatch 场景分发：alarm_explain / shift_handover / nl_query / default；
|
||
2. 请求校验：空 question、非法场景、异常处理 → error JSON；
|
||
3. HTTP 端点冒烟（mock runner）：GET /、/api/health、POST /api/chat、404。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from chat_api import ChatHandler, dispatch # noqa: E402
|
||
|
||
|
||
class _FakeResult:
|
||
def __init__(self, answer="ok", route_target="local", answer_id="a1",
|
||
needs_human=False):
|
||
self.answer = answer
|
||
self.route = mock.MagicMock()
|
||
self.route.target = route_target
|
||
self.answer_id = answer_id
|
||
self.needs_human = needs_human
|
||
|
||
|
||
class _FakeRunner:
|
||
"""模拟 TiScenarioRunner 三场景方法。"""
|
||
|
||
def explain_alarm(self, q, confidence=1.0):
|
||
return _FakeResult(answer=f"报警解释: {q}", route_target="local")
|
||
|
||
def generate_handover(self, q, confidence=1.0):
|
||
return _FakeResult(answer=f"交接班: {q}", route_target="local")
|
||
|
||
def query_cockpit(self, q, confidence=1.0):
|
||
return _FakeResult(answer=f"查询: {q}", route_target="local")
|
||
|
||
|
||
class TestDispatch(unittest.TestCase):
|
||
"""场景分发与统一响应。"""
|
||
|
||
def setUp(self):
|
||
self.runner = _FakeRunner()
|
||
|
||
def test_nl_query_default(self):
|
||
resp = dispatch(self.runner, {"question": "氯气流量趋势"})
|
||
self.assertEqual(resp["scenario"], "default")
|
||
self.assertTrue(resp["answer"].startswith("查询"))
|
||
|
||
def test_alarm_explain(self):
|
||
resp = dispatch(self.runner, {"question": "炉温报警",
|
||
"scenario": "alarm_explain"})
|
||
self.assertTrue(resp["answer"].startswith("报警解释"))
|
||
|
||
def test_shift_handover(self):
|
||
resp = dispatch(self.runner, {"question": "甲班交接",
|
||
"scenario": "shift_handover"})
|
||
self.assertTrue(resp["answer"].startswith("交接班"))
|
||
|
||
def test_empty_question(self):
|
||
resp = dispatch(self.runner, {"question": " "})
|
||
self.assertIn("error", resp)
|
||
|
||
def test_unknown_scenario_falls_back(self):
|
||
resp = dispatch(self.runner, {"question": "x",
|
||
"scenario": "no_such"})
|
||
self.assertEqual(resp["scenario"], "default")
|
||
|
||
def test_runner_exception_to_error(self):
|
||
bad = mock.MagicMock()
|
||
bad.query_cockpit.side_effect = RuntimeError("boom")
|
||
resp = dispatch(bad, {"question": "x"})
|
||
self.assertIn("处理失败", resp["error"])
|
||
|
||
|
||
class TestHTTPEndpoints(unittest.TestCase):
|
||
"""HTTP 端点冒烟（直接用 handler 方法 + mock runner）。"""
|
||
|
||
def setUp(self):
|
||
self.handler = ChatHandler.__new__(ChatHandler)
|
||
self.handler.runner = _FakeRunner()
|
||
self.handler.wfile = mock.MagicMock()
|
||
self.handler.send_response = mock.MagicMock()
|
||
self.handler.send_header = mock.MagicMock()
|
||
self.handler.end_headers = mock.MagicMock()
|
||
|
||
def _json_response(self, code, payload):
|
||
return json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
|
||
def test_get_health(self):
|
||
self.handler.path = "/api/health"
|
||
self.handler.do_GET()
|
||
sent = self.handler.wfile.write.call_args[0][0]
|
||
self.assertIn(b'"status": "ok"', sent)
|
||
|
||
def test_get_widget(self):
|
||
self.handler.path = "/"
|
||
with mock.patch("chat_api.WIDGET_PATH",
|
||
os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||
"..", "chat_widget.html")):
|
||
self.handler.do_GET()
|
||
sent = self.handler.wfile.write.call_args[0][0]
|
||
self.assertIn(b"iAOP", sent)
|
||
|
||
def test_post_chat(self):
|
||
self.handler.path = "/api/chat"
|
||
self.handler.headers = {"Content-Length": str(
|
||
len('{"question":"炉温报警","scenario":"alarm_explain"}'))}
|
||
self.handler.rfile = mock.MagicMock()
|
||
self.handler.rfile.read.return_value = (
|
||
'{"question":"炉温报警","scenario":"alarm_explain"}'.encode())
|
||
self.handler.do_POST()
|
||
sent = self.handler.wfile.write.call_args[0][0]
|
||
self.assertIn("报警解释", json.loads(sent.decode())["answer"])
|
||
|
||
def test_post_bad_json(self):
|
||
self.handler.path = "/api/chat"
|
||
self.handler.headers = {"Content-Length": "3"}
|
||
self.handler.rfile = mock.MagicMock()
|
||
self.handler.rfile.read.return_value = b"not json"
|
||
self.handler.do_POST()
|
||
sent = self.handler.wfile.write.call_args[0][0]
|
||
self.assertIn("合法 JSON", sent.decode())
|
||
|
||
def test_404(self):
|
||
self.handler.path = "/nope"
|
||
self.handler.do_GET()
|
||
sent = self.handler.wfile.write.call_args[0][0]
|
||
self.assertIn(b"not found", sent)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|