iAOP/web/chat/tests/test_chat_api.py

# -*- 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()