318 lines
12 KiB
Python
318 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import sys
|
||
import types
|
||
import unittest
|
||
from types import SimpleNamespace
|
||
from unittest import mock
|
||
|
||
|
||
if "truststore" not in sys.modules:
|
||
try:
|
||
import truststore # noqa: F401
|
||
except ModuleNotFoundError:
|
||
truststore_stub = types.ModuleType("truststore")
|
||
truststore_stub.inject_into_ssl = lambda: None
|
||
sys.modules["truststore"] = truststore_stub
|
||
|
||
if "lark_oapi" not in sys.modules:
|
||
try:
|
||
import lark_oapi # noqa: F401
|
||
except ModuleNotFoundError:
|
||
lark_stub = types.ModuleType("lark_oapi")
|
||
lark_stub.im = SimpleNamespace(v1=SimpleNamespace(P2ImMessageReceiveV1=object))
|
||
lark_stub.Client = object
|
||
sys.modules["lark_oapi"] = lark_stub
|
||
|
||
|
||
from shared.runtime import feishu_bot_bridge as bridge
|
||
|
||
|
||
def successful_payload() -> dict[str, object]:
|
||
return {
|
||
"data": {
|
||
"时间戳": "2026-08-31 12:00:00",
|
||
"交期汇总": {
|
||
"AM516预测交期": "2026-09-01",
|
||
"AM516总交期": 1,
|
||
"AM516主导因素": "现货",
|
||
},
|
||
"库存信息": {"可用库存(扣除了待产数量)": 100},
|
||
"BOM信息": {"是否有BOM(False则需提醒无BOM)": True},
|
||
"采购信息": {"缺料明细": []},
|
||
}
|
||
}
|
||
|
||
|
||
def message_event(message_id: str, chat_id: str, open_id: str) -> SimpleNamespace:
|
||
return SimpleNamespace(
|
||
event=SimpleNamespace(
|
||
message=SimpleNamespace(message_id=message_id, chat_id=chat_id),
|
||
sender=SimpleNamespace(sender_id=SimpleNamespace(open_id=open_id)),
|
||
)
|
||
)
|
||
|
||
|
||
class TriggerGateTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
with bridge._candidate_continuation_lock:
|
||
bridge._candidate_continuations.clear()
|
||
|
||
def test_complete_model_and_quantity_trigger(self) -> None:
|
||
request = bridge.parse_trigger_request("eRob70H50I-BHM-18CTC[V5] 10台")
|
||
|
||
self.assertEqual(
|
||
request,
|
||
bridge.RequestInput(model="eRob70H50I-BHM-18CTC[V5]", quantity=10),
|
||
)
|
||
|
||
def test_natural_language_model_and_quantity_trigger(self) -> None:
|
||
request = bridge.parse_trigger_request(
|
||
"请查询 eRob142H100I-BHM-18ET[V4]，这个数量20台的交期"
|
||
)
|
||
|
||
self.assertEqual(
|
||
request,
|
||
bridge.RequestInput(model="eRob142H100I-BHM-18ET[V4]", quantity=20),
|
||
)
|
||
|
||
def test_unrelated_group_message_is_silent(self) -> None:
|
||
self.assertIsNone(bridge.parse_trigger_request("这个机器人头像有点丑"))
|
||
|
||
def test_quantity_without_complete_model_is_silent(self) -> None:
|
||
self.assertIsNone(bridge.parse_trigger_request("关节模组要10台"))
|
||
|
||
def test_model_without_quantity_is_silent(self) -> None:
|
||
self.assertIsNone(bridge.parse_trigger_request("eRob70H50I-BHM-18CTC[V5]"))
|
||
|
||
def test_ambiguous_quantity_is_silent(self) -> None:
|
||
self.assertIsNone(
|
||
bridge.parse_trigger_request(
|
||
"eRob70H50I-BHM-18CTC[V5] 可能要10台，也可能要20台"
|
||
)
|
||
)
|
||
|
||
def test_invalid_series_version_is_silent(self) -> None:
|
||
self.assertIsNone(bridge.parse_trigger_request("eRob70H50I-BHM-18CTC[V6] 10台"))
|
||
|
||
def test_ignored_message_does_not_build_or_reply(self) -> None:
|
||
data = SimpleNamespace(
|
||
event=SimpleNamespace(message=SimpleNamespace(message_id="om_ignored"))
|
||
)
|
||
|
||
with (
|
||
mock.patch.object(bridge, "message_text", return_value="他怎么什么消息都回复啊"),
|
||
mock.patch.object(bridge, "build_report") as build_report,
|
||
mock.patch.object(bridge, "reply") as reply,
|
||
):
|
||
bridge.handle_message(object(), data)
|
||
|
||
build_report.assert_not_called()
|
||
reply.assert_not_called()
|
||
|
||
def test_valid_message_builds_and_replies(self) -> None:
|
||
data = SimpleNamespace(
|
||
event=SimpleNamespace(message=SimpleNamespace(message_id="om_valid"))
|
||
)
|
||
|
||
with (
|
||
mock.patch.object(
|
||
bridge,
|
||
"message_text",
|
||
return_value="eRob70H50I-BHM-18CTC[V5] 10台",
|
||
),
|
||
mock.patch.object(bridge, "build_report", return_value="report") as build_report,
|
||
mock.patch.object(bridge, "reply") as reply,
|
||
):
|
||
bridge.handle_message("client", data)
|
||
|
||
build_report.assert_called_once_with(
|
||
bridge.RequestInput(model="eRob70H50I-BHM-18CTC[V5]", quantity=10)
|
||
)
|
||
reply.assert_called_once_with("client", "om_valid", "report")
|
||
|
||
def test_candidate_continuation_phrase_is_recognized(self) -> None:
|
||
valid_phrases = (
|
||
"还有没有其他候选型号？",
|
||
"还有吗？",
|
||
"还有别的吗",
|
||
"其他呢",
|
||
"还有类似的吗",
|
||
"还有什么推荐",
|
||
"有没有其他选择",
|
||
"还有其他方案吗",
|
||
"再推荐几个",
|
||
"可以再给我几个吗",
|
||
"继续吧",
|
||
"接着来",
|
||
"下一批",
|
||
"换一批",
|
||
"多看几个候选",
|
||
"more candidates",
|
||
"any other options?",
|
||
)
|
||
for phrase in valid_phrases:
|
||
with self.subTest(phrase=phrase):
|
||
self.assertTrue(bridge.is_candidate_continuation_request(phrase))
|
||
|
||
invalid_phrases = (
|
||
"这个机器人头像有点丑",
|
||
"more interesting",
|
||
"继续查询英语例句",
|
||
"下一课",
|
||
"再来一首歌",
|
||
"有其他事情吗",
|
||
)
|
||
for phrase in invalid_phrases:
|
||
with self.subTest(phrase=phrase):
|
||
self.assertFalse(bridge.is_candidate_continuation_request(phrase))
|
||
|
||
def test_continuation_without_same_conversation_context_is_silent(self) -> None:
|
||
data = message_event("om_more", "oc_other", "ou_other")
|
||
|
||
with (
|
||
mock.patch.object(
|
||
bridge,
|
||
"message_text",
|
||
return_value="还有没有其他候选型号？",
|
||
),
|
||
mock.patch.object(bridge, "build_report") as build_report,
|
||
mock.patch.object(bridge, "reply") as reply,
|
||
):
|
||
bridge.handle_message("client", data)
|
||
|
||
build_report.assert_not_called()
|
||
reply.assert_not_called()
|
||
|
||
def test_follow_up_continues_from_fifth_candidate_in_same_conversation(self) -> None:
|
||
request_text = "eRob70H50I-FS-18CN[V5] 10台"
|
||
request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
|
||
initial_data = message_event("om_initial", "oc_same", "ou_same")
|
||
follow_up_data = message_event("om_follow_up", "oc_same", "ou_same")
|
||
|
||
with (
|
||
mock.patch.object(
|
||
bridge,
|
||
"message_text",
|
||
side_effect=[request_text, "还有吗？"],
|
||
),
|
||
mock.patch.object(
|
||
bridge,
|
||
"build_report",
|
||
side_effect=["initial report", "continued report"],
|
||
) as build_report,
|
||
mock.patch.object(bridge, "reply") as reply,
|
||
):
|
||
bridge.handle_message("client", initial_data)
|
||
bridge.handle_message("client", follow_up_data)
|
||
|
||
self.assertEqual(
|
||
build_report.call_args_list,
|
||
[mock.call(request), mock.call(request, candidate_offset=4)],
|
||
)
|
||
self.assertEqual(
|
||
reply.call_args_list,
|
||
[
|
||
mock.call("client", "om_initial", "initial report"),
|
||
mock.call("client", "om_follow_up", "continued report"),
|
||
],
|
||
)
|
||
|
||
def test_context_is_cleared_when_all_candidates_fit_in_first_batch(self) -> None:
|
||
request_text = "eRob70H50I-BHM-18CTC[V5] 10台"
|
||
initial_data = message_event("om_small", "oc_small", "ou_small")
|
||
follow_up_data = message_event("om_small_more", "oc_small", "ou_small")
|
||
|
||
with (
|
||
mock.patch.object(
|
||
bridge,
|
||
"message_text",
|
||
side_effect=[request_text, "还有没有其他候选型号？"],
|
||
),
|
||
mock.patch.object(bridge, "build_report", return_value="report") as build_report,
|
||
mock.patch.object(bridge, "reply") as reply,
|
||
):
|
||
bridge.handle_message("client", initial_data)
|
||
bridge.handle_message("client", follow_up_data)
|
||
|
||
build_report.assert_called_once_with(
|
||
bridge.RequestInput(model="eRob70H50I-BHM-18CTC[V5]", quantity=10)
|
||
)
|
||
reply.assert_called_once_with("client", "om_small", "report")
|
||
|
||
def test_expired_candidate_context_is_removed(self) -> None:
|
||
key = "oc_expired:ou_expired"
|
||
request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
|
||
bridge._candidate_continuations[key] = bridge.CandidateContinuation(
|
||
request=request,
|
||
next_offset=4,
|
||
total_candidates=10,
|
||
expires_at=0,
|
||
)
|
||
|
||
self.assertIsNone(bridge.get_candidate_continuation(key))
|
||
self.assertNotIn(key, bridge._candidate_continuations)
|
||
|
||
|
||
class CandidateBatchTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
|
||
self.specs = bridge.generate_candidate_specs(self.request.model)
|
||
self.assertGreater(len(self.specs), bridge.CANDIDATE_BATCH_SIZE * 2)
|
||
|
||
def test_initial_report_queries_baseline_and_only_four_candidates(self) -> None:
|
||
with (
|
||
mock.patch.object(
|
||
bridge,
|
||
"run_controlled_query",
|
||
return_value=successful_payload(),
|
||
) as run_query,
|
||
mock.patch.object(bridge, "write_record"),
|
||
):
|
||
report = bridge.build_report(self.request)
|
||
|
||
queried_models = [call.args[0].model for call in run_query.call_args_list]
|
||
self.assertEqual(len(queried_models), 1 + bridge.CANDIDATE_BATCH_SIZE)
|
||
self.assertEqual(queried_models[0], self.request.model)
|
||
self.assertCountEqual(
|
||
queried_models[1:],
|
||
[spec["model"] for spec in self.specs[: bridge.CANDIDATE_BATCH_SIZE]],
|
||
)
|
||
self.assertIn("本次先推荐4个候选型号（原型号另作基准）", report)
|
||
self.assertIn("可说“还有吗”“再推荐几个”“继续”等类似表达", report)
|
||
self.assertNotIn(self.specs[bridge.CANDIDATE_BATCH_SIZE]["model"], report)
|
||
|
||
def test_follow_up_report_queries_next_four_without_requerying_baseline(self) -> None:
|
||
with (
|
||
mock.patch.object(
|
||
bridge,
|
||
"run_controlled_query",
|
||
return_value=successful_payload(),
|
||
) as run_query,
|
||
mock.patch.object(bridge, "write_record"),
|
||
):
|
||
report = bridge.build_report(
|
||
self.request,
|
||
candidate_offset=bridge.CANDIDATE_BATCH_SIZE,
|
||
)
|
||
|
||
queried_models = [call.args[0].model for call in run_query.call_args_list]
|
||
expected_models = [
|
||
spec["model"]
|
||
for spec in self.specs[
|
||
bridge.CANDIDATE_BATCH_SIZE : bridge.CANDIDATE_BATCH_SIZE * 2
|
||
]
|
||
]
|
||
self.assertEqual(len(queried_models), bridge.CANDIDATE_BATCH_SIZE)
|
||
self.assertNotIn(self.request.model, queried_models)
|
||
self.assertCountEqual(queried_models, expected_models)
|
||
self.assertIn("本次继续推荐第5-8个候选型号", report)
|
||
for model in expected_models:
|
||
self.assertIn(model, report)
|
||
for model in [spec["model"] for spec in self.specs[:4]]:
|
||
self.assertNotIn(model, report)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|