fix: ignore non-query Feishu messages

This commit is contained in:
ar51
2026-08-08 18:09:05 +08:00
parent 0aaedb8b66
commit d97b788e0e
7 changed files with 187 additions and 19 deletions

View File

@@ -0,0 +1,107 @@
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
class TriggerGateTests(unittest.TestCase):
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")
if __name__ == "__main__":
unittest.main()