500 lines
26 KiB
Python
500 lines
26 KiB
Python
from __future__ import annotations
|
||
|
||
import contextlib
|
||
import importlib.util
|
||
import io
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
|
||
SCRIPT_PATH = (
|
||
Path(__file__).resolve().parents[1]
|
||
/ "skills/send-complex-message/scripts/send_complex_message.py"
|
||
)
|
||
NOW = 1_789_272_000 # 2026-09-13 12:00:00 +08:00
|
||
|
||
|
||
class HistoryCursor:
|
||
"""在内存数据库执行实际查询，仅转换 MySQL 的占位符和转义字符串语法。"""
|
||
|
||
def __init__(self, database) -> None:
|
||
self.cursor = database.cursor()
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *_):
|
||
self.cursor.close()
|
||
|
||
def execute(self, sql, params) -> None:
|
||
sql = sql.replace("%s", "?").replace("ESCAPE '\\\\'", "ESCAPE '\\'")
|
||
self.cursor.execute(sql, params)
|
||
|
||
def fetchall(self):
|
||
return [dict(row) for row in self.cursor.fetchall()]
|
||
|
||
|
||
class HistoryConnection:
|
||
def __init__(self, messages: list[dict]) -> None:
|
||
self.database = sqlite3.connect(":memory:")
|
||
self.database.row_factory = sqlite3.Row
|
||
self.closed = False
|
||
self.database.execute("""
|
||
CREATE TABLE messages (
|
||
id INTEGER PRIMARY KEY, from_wxid TEXT, sender_wxid TEXT, to_wxid TEXT,
|
||
is_chat_room INTEGER, type INTEGER, app_msg_type INTEGER,
|
||
content TEXT, display_full_content TEXT, is_recalled INTEGER, created_at INTEGER
|
||
)
|
||
""")
|
||
for message in messages:
|
||
row = {
|
||
"from_wxid": "room@chatroom", "sender_wxid": "wxid_alice",
|
||
"to_wxid": "wxid_robot", "is_chat_room": 1, "type": 1,
|
||
"app_msg_type": 0, "content": "安排会议", "display_full_content": "",
|
||
"is_recalled": 0, "created_at": NOW - 60,
|
||
**message,
|
||
}
|
||
self.database.execute(
|
||
f"INSERT INTO messages ({', '.join(row)}) VALUES ({', '.join(['?'] * len(row))})",
|
||
tuple(row.values()),
|
||
)
|
||
|
||
def cursor(self):
|
||
return HistoryCursor(self.database)
|
||
|
||
def close(self) -> None:
|
||
self.closed = True
|
||
self.database.close()
|
||
|
||
|
||
class MemberConnection(HistoryConnection):
|
||
def __init__(self, members: list[dict]) -> None:
|
||
super().__init__([])
|
||
self.database.execute("""
|
||
CREATE TABLE chat_room_members (
|
||
id INTEGER PRIMARY KEY, chat_room_id TEXT, wechat_id TEXT,
|
||
remark TEXT, nickname TEXT, is_leaved INTEGER
|
||
)
|
||
""")
|
||
for member in members:
|
||
row = {
|
||
"chat_room_id": "room@chatroom", "remark": "", "nickname": "",
|
||
"is_leaved": 0, **member,
|
||
}
|
||
self.database.execute(
|
||
f"INSERT INTO chat_room_members ({', '.join(row)}) VALUES ({', '.join(['?'] * len(row))})",
|
||
tuple(row.values()),
|
||
)
|
||
|
||
|
||
class SendComplexMessageTests(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
spec = importlib.util.spec_from_file_location("send_complex_message", SCRIPT_PATH)
|
||
if spec is None or spec.loader is None:
|
||
raise RuntimeError(f"无法加载测试脚本：{SCRIPT_PATH}")
|
||
cls.module = importlib.util.module_from_spec(spec)
|
||
with mock.patch.object(sys, "stderr", sys.stderr):
|
||
spec.loader.exec_module(cls.module)
|
||
|
||
def setUp(self) -> None:
|
||
env = mock.patch.dict(os.environ, {
|
||
"ROBOT_FROM_WX_ID": "room@chatroom",
|
||
"ROBOT_WECHAT_CLIENT_PORT": "9000",
|
||
# 已有上下文消息也不能让不引用的请求自动带上引用 ID。
|
||
"ROBOT_MESSAGE_ID": "123",
|
||
"ROBOT_REF_MESSAGE_ID": "456",
|
||
}, clear=True)
|
||
env.start()
|
||
self.addCleanup(env.stop)
|
||
|
||
def test_sending_modes_and_optional_reference(self) -> None:
|
||
cases = [
|
||
("mention only", ["--mention", "张三"], "", ["wxid_zhangsan"], None),
|
||
("text only", ["--content", "收到"], "收到", [], None),
|
||
("quote only", ["--refer-message-id", "12", "--content", "收到"], "收到", [], 12),
|
||
("quote and mention", ["--refer-message-id", "12", "--mention", "张三", "--content", "收到"], "收到", ["wxid_zhangsan"], 12),
|
||
("mention all only", ["--all"], "", ["notify@all"], None),
|
||
("text and mention", ["--mention", "张三", "--content", "收到"], "收到", ["wxid_zhangsan"], None),
|
||
("large primary ID", ["--refer-message-id", "9007199254740993", "--content", "收到"], "收到", [], 9007199254740993),
|
||
]
|
||
for name, args, content, at, reference in cases:
|
||
with self.subTest(name=name), contextlib.ExitStack() as stack:
|
||
stack.enter_context(mock.patch.object(sys, "argv", [str(SCRIPT_PATH), *args, "--ended"]))
|
||
stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
||
connect = stack.enter_context(mock.patch.object(self.module, "_mysql_connect"))
|
||
resolve = stack.enter_context(mock.patch.object(self.module, "_resolve_mentions", return_value=(["wxid_zhangsan"], [])))
|
||
post = stack.enter_context(mock.patch.object(self.module, "_http_post_json", return_value={"code": 200, "data": None}))
|
||
|
||
self.assertEqual(self.module.main(), 0)
|
||
body = {"to_wxid": "room@chatroom", "content": content, "at": at}
|
||
if reference is not None:
|
||
body["refer_message_id"] = reference
|
||
post.assert_called_once_with(
|
||
"http://127.0.0.1:9000/api/v1/robot/message/send/refermessage",
|
||
body,
|
||
)
|
||
if "--mention" in args:
|
||
resolve.assert_called_once_with(connect.return_value, "room@chatroom", ["张三"], [])
|
||
connect.return_value.close.assert_called_once()
|
||
else:
|
||
connect.assert_not_called()
|
||
resolve.assert_not_called()
|
||
|
||
def test_invalid_combinations_do_not_send(self) -> None:
|
||
cases = [
|
||
[],
|
||
["--content", " \t\n"],
|
||
["--mentions", "[]"],
|
||
["--refer-message-id", "12"],
|
||
["--refer-message-id", "12", "--mention", "张三"],
|
||
["--refer-message-id", "12", "--content", " \t\n", "--all"],
|
||
["--refer-message-id", "0", "--content", "收到"],
|
||
["--refer-message-id", "-1", "--content", "收到"],
|
||
["--refer-message-id", "9223372036854775808", "--content", "收到"],
|
||
["--mention-wxid", " \t", "--content", "收到"],
|
||
["--mention-wxid", "wxid_alice", "--all"],
|
||
["--mention-wxid", "wxid_alice", "--refer-message-id", "12"],
|
||
]
|
||
for args in cases:
|
||
with self.subTest(args=args), contextlib.ExitStack() as stack:
|
||
output = stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
||
stack.enter_context(mock.patch.object(sys, "argv", [str(SCRIPT_PATH), *args, "--ended"]))
|
||
connect = stack.enter_context(mock.patch.object(self.module, "_mysql_connect"))
|
||
post = stack.enter_context(mock.patch.object(self.module, "_http_post_json"))
|
||
|
||
self.assertEqual(self.module.main(), 1)
|
||
connect.assert_not_called()
|
||
post.assert_not_called()
|
||
self.assertFalse(output.getvalue().endswith("ended"))
|
||
|
||
def run_member_send(self, members, args):
|
||
connection = MemberConnection(members)
|
||
self.addCleanup(connection.close)
|
||
with mock.patch.object(sys, "argv", [str(SCRIPT_PATH), *args, "--ended"]), \
|
||
mock.patch.object(self.module, "_mysql_connect", return_value=connection), \
|
||
mock.patch.object(self.module, "_http_post_json", return_value={"code": 200}) as post, \
|
||
contextlib.redirect_stdout(io.StringIO()) as output:
|
||
result = self.module.main()
|
||
self.assertTrue(connection.closed)
|
||
return result, output.getvalue(), post
|
||
|
||
def test_memory_resolved_id_targets_renamed_member_despite_duplicate_names(self) -> None:
|
||
result, output, post = self.run_member_send([
|
||
{"id": 1, "wechat_id": "wxid_other", "nickname": "现在的名字"},
|
||
{"id": 2, "wechat_id": "wxid_renamed", "nickname": "现在的名字", "is_leaved": None},
|
||
], ["--mention-wxid", " wxid_renamed "])
|
||
self.assertEqual(result, 0)
|
||
post.assert_called_once_with(
|
||
"http://127.0.0.1:9000/api/v1/robot/message/send/refermessage",
|
||
{"to_wxid": "room@chatroom", "content": "", "at": ["wxid_renamed"]},
|
||
)
|
||
self.assertTrue(output.endswith("ended"))
|
||
|
||
def test_explicit_ids_require_exact_current_room_membership(self) -> None:
|
||
cases = [
|
||
{"wechat_id": "wxid_target", "is_leaved": 1},
|
||
{"wechat_id": "wxid_target", "chat_room_id": "other@chatroom"},
|
||
{"wechat_id": "wxid_target_suffix"},
|
||
]
|
||
for member in cases:
|
||
with self.subTest(member=member):
|
||
result, output, post = self.run_member_send([
|
||
{"id": 1, **member},
|
||
# 即使别人的当前昵称或备注恰好等于目标 ID，也不能回退匹配。
|
||
{"id": 2, "wechat_id": "wxid_other", "nickname": "wxid_target", "remark": "wxid_target"},
|
||
], ["--mention-wxid", "wxid_target", "--content", "请看公告"])
|
||
self.assertEqual(result, 1)
|
||
self.assertIn("未找到当前群内未退群成员: wxid_target", output)
|
||
self.assertFalse(output.endswith("ended"))
|
||
post.assert_not_called()
|
||
|
||
def test_missing_old_name_stops_entire_send_then_accepts_resolved_id(self) -> None:
|
||
members = [
|
||
{"id": 1, "wechat_id": "wxid_zhangsan", "nickname": "张三"},
|
||
{"id": 2, "wechat_id": "wxid_renamed", "nickname": "新名字"},
|
||
]
|
||
common_args = ["--mention", "张三", "--content", "请看公告", "--refer-message-id", "12"]
|
||
result, output, post = self.run_member_send(members, [*common_args, "--mention", "旧名字"])
|
||
self.assertEqual(result, 1)
|
||
self.assertIn("旧名字", output)
|
||
self.assertIn("本次未发送消息", output)
|
||
self.assertIn("search_chat_room_memory", output)
|
||
self.assertIn("--mention-wxid", output)
|
||
self.assertFalse(output.endswith("ended"))
|
||
post.assert_not_called()
|
||
|
||
result, _, post = self.run_member_send(members, [*common_args, "--mention-wxid", "wxid_renamed"])
|
||
self.assertEqual(result, 0)
|
||
post.assert_called_once_with(
|
||
"http://127.0.0.1:9000/api/v1/robot/message/send/refermessage",
|
||
{"to_wxid": "room@chatroom", "content": "请看公告",
|
||
"at": ["wxid_zhangsan", "wxid_renamed"], "refer_message_id": 12},
|
||
)
|
||
|
||
def test_mixed_names_and_repeated_ids_mention_each_person_once(self) -> None:
|
||
result, _, post = self.run_member_send([
|
||
{"id": 1, "wechat_id": "wxid_zhangsan", "nickname": "张三"},
|
||
{"id": 2, "wechat_id": "wxid_lisi", "nickname": "李四"},
|
||
], ["--mention", "张三", "--mention-wxid", "wxid_zhangsan",
|
||
"--mention-wxid", "wxid_lisi", "--mention-wxid", "wxid_lisi"])
|
||
self.assertEqual(result, 0)
|
||
self.assertEqual(post.call_args.args[1]["at"], ["wxid_zhangsan", "wxid_lisi"])
|
||
|
||
def test_current_name_fallback_rejects_exact_and_partial_ambiguity(self) -> None:
|
||
cases = [
|
||
[{"remark": "张三"}, {"nickname": "张三"}],
|
||
[{"nickname": "张三"}, {"nickname": "张三"}],
|
||
[{"remark": "张三甲"}, {"nickname": "张三乙"}],
|
||
]
|
||
for names in cases:
|
||
with self.subTest(names=names):
|
||
members = [
|
||
{"id": index, "wechat_id": f"wxid_{index}", **name}
|
||
for index, name in enumerate(names, start=1)
|
||
]
|
||
result, output, post = self.run_member_send(members, ["--mention", "张三"])
|
||
self.assertEqual(result, 1)
|
||
self.assertIn("成员名称有歧义", output)
|
||
self.assertFalse(output.endswith("ended"))
|
||
post.assert_not_called()
|
||
|
||
def test_current_name_fallback_finds_exact_match_beyond_many_partial_matches(self) -> None:
|
||
members = [
|
||
{"id": index, "wechat_id": f"wxid_{index}", "nickname": f"张三{index}"}
|
||
for index in range(1, 61)
|
||
]
|
||
members.extend([
|
||
{"id": 61, "wechat_id": "wxid_target", "nickname": " 张三 "},
|
||
{"id": 62, "wechat_id": "wxid_left", "nickname": "张三", "is_leaved": 1},
|
||
{"id": 63, "wechat_id": "wxid_other_room", "nickname": "张三", "chat_room_id": "other@chatroom"},
|
||
])
|
||
result, _, post = self.run_member_send(members, ["--mention", "张三"])
|
||
self.assertEqual(result, 0)
|
||
self.assertEqual(post.call_args.args[1]["at"], ["wxid_target"])
|
||
|
||
def test_current_name_partial_matching_treats_wildcards_as_literal(self) -> None:
|
||
result, _, post = self.run_member_send([
|
||
{"id": 1, "wechat_id": "wxid_target", "nickname": "前缀100%_\\后缀"},
|
||
{"id": 2, "wechat_id": "wxid_other", "nickname": "前缀100AB\\后缀"},
|
||
], ["--mention", "100%_\\"])
|
||
self.assertEqual(result, 0)
|
||
self.assertEqual(post.call_args.args[1]["at"], ["wxid_target"])
|
||
|
||
def test_unresolved_id_does_not_send_other_resolved_members(self) -> None:
|
||
result, output, post = self.run_member_send([
|
||
{"id": 1, "wechat_id": "wxid_zhangsan", "nickname": "张三"},
|
||
], ["--mention", "张三", "--mention-wxid", "wxid_missing", "--content", "请看公告"])
|
||
self.assertEqual(result, 1)
|
||
self.assertFalse(output.endswith("ended"))
|
||
post.assert_not_called()
|
||
|
||
def test_mention_ids_are_rejected_in_private_conversations(self) -> None:
|
||
with mock.patch.dict(os.environ, {"ROBOT_FROM_WX_ID": "wxid_friend"}), \
|
||
mock.patch.object(sys, "argv", [str(SCRIPT_PATH), "--mention-wxid", "wxid_alice"]), \
|
||
mock.patch.object(self.module, "_mysql_connect") as connect, \
|
||
mock.patch.object(self.module, "_http_post_json") as post, \
|
||
contextlib.redirect_stdout(io.StringIO()) as output:
|
||
self.assertEqual(self.module.main(), 1)
|
||
self.assertIn("当前会话不是群聊", output.getvalue())
|
||
connect.assert_not_called()
|
||
post.assert_not_called()
|
||
|
||
def run_history(self, rows, args=(), conversation="room@chatroom") -> dict:
|
||
connection = HistoryConnection(rows)
|
||
self.addCleanup(connection.close)
|
||
with contextlib.ExitStack() as stack:
|
||
# 查询无须客户端端口；也不应该调用任何发送接口。
|
||
stack.enter_context(mock.patch.dict(os.environ, {"ROBOT_FROM_WX_ID": conversation}, clear=True))
|
||
stack.enter_context(mock.patch.object(sys, "argv", [str(SCRIPT_PATH), "--query-history", *args]))
|
||
stack.enter_context(mock.patch.object(self.module.time, "time", return_value=NOW))
|
||
stack.enter_context(mock.patch.object(self.module, "_mysql_connect", return_value=connection))
|
||
send = stack.enter_context(mock.patch.object(self.module, "_http_post_json"))
|
||
output = stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
||
|
||
self.assertEqual(self.module.main(), 0, output.getvalue())
|
||
self.assertTrue(connection.closed)
|
||
send.assert_not_called()
|
||
result = json.loads(output.getvalue())
|
||
self.assertEqual(result["conversation_id"], conversation)
|
||
return result
|
||
|
||
def test_history_group_isolation_and_24_hour_boundaries(self) -> None:
|
||
result = self.run_history([
|
||
{"id": 1, "created_at": NOW - 86400},
|
||
{"id": 2, "created_at": NOW},
|
||
{"id": 3, "created_at": NOW - 86401},
|
||
{"id": 4, "created_at": NOW + 1},
|
||
{"id": 5, "from_wxid": "other@chatroom"},
|
||
{"id": 6, "from_wxid": "wxid_alice", "is_chat_room": 0},
|
||
{"id": 7, "is_chat_room": 0},
|
||
])
|
||
self.assertEqual([row["id"] for row in result["messages"]], [2, 1])
|
||
self.assertTrue(result["is_chat_room"])
|
||
self.assertEqual((result["start_time"], result["end_time"]), (NOW - 86400, NOW))
|
||
|
||
def test_history_private_chat_includes_both_directions_only_for_current_friend(self) -> None:
|
||
result = self.run_history([
|
||
{"id": 1, "from_wxid": "wxid_alice", "is_chat_room": 0},
|
||
{"id": 2, "from_wxid": "wxid_alice", "is_chat_room": 0, "sender_wxid": "wxid_robot"},
|
||
{"id": 3, "from_wxid": "wxid_bob", "is_chat_room": 0},
|
||
{"id": 4}, # 同一个人的群消息不能混入私聊。
|
||
{"id": 5, "from_wxid": "wxid_alice", "is_chat_room": 1},
|
||
{"id": 6, "from_wxid": "wxid_alice", "is_chat_room": 0, "created_at": NOW - 86401},
|
||
], conversation="wxid_alice")
|
||
self.assertEqual([row["id"] for row in result["messages"]], [2, 1])
|
||
self.assertFalse(result["is_chat_room"])
|
||
|
||
def test_history_accepts_shorter_relative_and_absolute_ranges(self) -> None:
|
||
rows = [
|
||
{"id": 1, "created_at": NOW - 1801},
|
||
{"id": 2, "created_at": NOW - 1800},
|
||
{"id": 3, "created_at": NOW - 900},
|
||
{"id": 4, "created_at": NOW - 899},
|
||
{"id": 5, "created_at": NOW},
|
||
]
|
||
cases = [
|
||
(["--hours", "0.5"], [5, 4, 3, 2], NOW - 1800, NOW),
|
||
(["--hours", "24"], [5, 4, 3, 2, 1], NOW - 86400, NOW),
|
||
(["--start-time", str(NOW - 1800), "--end-time", str(NOW - 900)], [3, 2], NOW - 1800, NOW - 900),
|
||
(["--start-time", "2026-09-13 11:30", "--end-time", "2026-09-13 11:45:00"], [3, 2], NOW - 1800, NOW - 900),
|
||
(["--start-time", str(NOW - 900)], [5, 4, 3], NOW - 900, NOW),
|
||
(["--end-time", str(NOW - 900)], [3, 2, 1], NOW - 86400, NOW - 900),
|
||
]
|
||
for args, ids, start, end in cases:
|
||
with self.subTest(args=args):
|
||
result = self.run_history(rows, args)
|
||
self.assertEqual([row["id"] for row in result["messages"]], ids)
|
||
self.assertEqual((result["start_time"], result["end_time"]), (start, end))
|
||
|
||
def test_history_combines_keywords_types_and_sender(self) -> None:
|
||
result = self.run_history([
|
||
{"id": 1, "content": "安排", "display_full_content": "会议通知", "type": 49, "app_msg_type": 57},
|
||
{"id": 2, "content": "安排会议", "type": 49, "app_msg_type": 6},
|
||
{"id": 3, "content": "安排会议", "type": 49, "app_msg_type": 5},
|
||
{"id": 4, "content": "安排会议", "type": 1, "app_msg_type": 57},
|
||
{"id": 5, "content": "安排会议", "type": 49, "app_msg_type": 57, "sender_wxid": "wxid_bob"},
|
||
{"id": 6, "content": "安排", "type": 49, "app_msg_type": 57},
|
||
{"id": 7, "content": "安排会议", "type": 49, "app_msg_type": 57, "from_wxid": "other@chatroom"},
|
||
{"id": 8, "content": "安排会议", "type": 49, "app_msg_type": 57, "created_at": NOW - 3601},
|
||
], ["--hours", "1", "--keyword", "安排", "--keyword", "会议", "--message-type", "app",
|
||
"--app-msg-type", "57", "--app-msg-type", "6", "--sender-wxid", "wxid_alice"])
|
||
self.assertEqual([row["id"] for row in result["messages"]], [2, 1])
|
||
|
||
def test_history_accepts_multiple_message_types_and_app_subtype_alone(self) -> None:
|
||
rows = [
|
||
{"id": 1, "type": 3}, {"id": 2, "type": 43}, {"id": 3, "type": 34},
|
||
{"id": 4, "type": 49, "app_msg_type": 57}, {"id": 5, "type": 1, "app_msg_type": 57},
|
||
]
|
||
result = self.run_history(rows, ["--message-type", "image", "--message-type", "43"])
|
||
self.assertEqual([row["id"] for row in result["messages"]], [2, 1])
|
||
result = self.run_history(rows, ["--app-msg-type", "57"])
|
||
self.assertEqual([row["id"] for row in result["messages"]], [4])
|
||
|
||
def test_history_keywords_are_literal_and_cannot_bypass_scope(self) -> None:
|
||
for keyword in ["100%_\\", "' OR 1=1 --"]:
|
||
with self.subTest(keyword=keyword):
|
||
result = self.run_history([
|
||
{"id": 1, "content": f"前缀{keyword}后缀"},
|
||
{"id": 2, "content": "100AB\\"},
|
||
{"id": 3, "content": keyword, "from_wxid": "other@chatroom"},
|
||
], ["--keyword", keyword])
|
||
self.assertEqual([row["id"] for row in result["messages"]], [1])
|
||
|
||
def test_history_pagination_preserves_order_and_primary_ids(self) -> None:
|
||
rows = [{"id": 1}, {"id": 2}, {"id": 9007199254740993}]
|
||
first = self.run_history(rows, ["--limit", "2"])
|
||
self.assertEqual([row["id"] for row in first["messages"]], [9007199254740993, 2])
|
||
self.assertEqual(first["count"], 2)
|
||
self.assertTrue(first["has_more"])
|
||
self.assertEqual(first["next_offset"], 2)
|
||
last = self.run_history(rows, ["--limit", "2", "--offset", str(first["next_offset"])])
|
||
self.assertEqual([row["id"] for row in last["messages"]], [1])
|
||
self.assertFalse(last["has_more"])
|
||
self.assertIsNone(last["next_offset"])
|
||
empty = self.run_history(rows, ["--offset", "3"])
|
||
self.assertEqual(empty["messages"], [])
|
||
self.assertEqual(empty["count"], 0)
|
||
self.assertFalse(empty["has_more"])
|
||
|
||
def test_invalid_history_requests_neither_query_nor_send(self) -> None:
|
||
cases = [
|
||
["--hours", "0"], ["--hours", "-1"], ["--hours", "24.01"],
|
||
["--hours", "nan"], ["--hours", "inf"], ["--hours", "0.00001"], ["--hours", "bad"],
|
||
["--start-time", str(NOW - 86401)],
|
||
["--start-time", str(NOW - 172800), "--end-time", str(NOW - 86400)],
|
||
["--start-time", str(NOW - 172800), "--end-time", str(NOW - 169200)],
|
||
["--end-time", str(NOW + 1)], ["--end-time", str(NOW - 86401)],
|
||
["--start-time", str(NOW)], ["--start-time", str(NOW + 1)],
|
||
["--start-time", "invalid"], ["--start-time", ""], ["--end-time", "2026-09-31 09:00"],
|
||
["--start-time", str(NOW - 60), "--end-time", str(NOW - 120)],
|
||
["--hours", "1", "--start-time", str(NOW - 60)],
|
||
["--hours", "1", "--end-time", str(NOW - 60)],
|
||
["--limit", "0"], ["--limit", "201"], ["--offset", "-1"],
|
||
["--keyword", " "], ["--sender-wxid", " "],
|
||
["--message-type", "bad"], ["--message-type", "0"], ["--app-msg-type", "-1"],
|
||
["--message-type", "text", "--app-msg-type", "57"],
|
||
["--content", "你好"], ["--content", ""], ["--mention", "张三"], ["--mentions", "[]"],
|
||
["--mention-wxid", "wxid_alice"],
|
||
["--all"], ["--refer-message-id", "1"], ["--ended"],
|
||
["--conversation-id", "other@chatroom"], ["--from-wxid", "other@chatroom"],
|
||
]
|
||
for args in cases:
|
||
with self.subTest(args=args), contextlib.ExitStack() as stack:
|
||
stack.enter_context(mock.patch.object(sys, "argv", [str(SCRIPT_PATH), "--query-history", *args]))
|
||
stack.enter_context(mock.patch.object(self.module.time, "time", return_value=NOW))
|
||
connect = stack.enter_context(mock.patch.object(self.module, "_mysql_connect"))
|
||
send = stack.enter_context(mock.patch.object(self.module, "_http_post_json"))
|
||
output = stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
||
self.assertEqual(self.module.main(), 1)
|
||
connect.assert_not_called()
|
||
send.assert_not_called()
|
||
self.assertFalse(output.getvalue().endswith("ended"))
|
||
|
||
def test_history_filters_require_explicit_query_mode(self) -> None:
|
||
for flag, value in [("--hours", "1"), ("--limit", "50"), ("--keyword", "会议")]:
|
||
with self.subTest(flag=flag):
|
||
with self.assertRaises(ValueError):
|
||
self.module._parse_cli_params(["--content", "你好", flag, value])
|
||
|
||
def test_history_requires_current_conversation(self) -> None:
|
||
with mock.patch.dict(os.environ, {}, clear=True), \
|
||
mock.patch.object(sys, "argv", [str(SCRIPT_PATH), "--query-history"]), \
|
||
mock.patch.object(self.module, "_mysql_connect") as connect, \
|
||
contextlib.redirect_stdout(io.StringIO()) as output:
|
||
self.assertEqual(self.module.main(), 1)
|
||
self.assertIn("ROBOT_FROM_WX_ID", output.getvalue())
|
||
connect.assert_not_called()
|
||
|
||
def test_history_recomputes_window_after_connecting(self) -> None:
|
||
connection = HistoryConnection([{"id": 1, "created_at": NOW - 86400}])
|
||
with mock.patch.object(sys, "argv", [str(SCRIPT_PATH), "--query-history"]), \
|
||
mock.patch.object(self.module.time, "time", side_effect=[NOW, NOW + 10]), \
|
||
mock.patch.object(self.module, "_mysql_connect", return_value=connection), \
|
||
contextlib.redirect_stdout(io.StringIO()) as output:
|
||
self.assertEqual(self.module.main(), 0)
|
||
self.assertEqual(json.loads(output.getvalue())["messages"], [])
|
||
self.assertTrue(connection.closed)
|
||
|
||
def test_history_query_failure_closes_connection(self) -> None:
|
||
connection = mock.MagicMock()
|
||
connection.cursor.side_effect = RuntimeError("查询失败")
|
||
with mock.patch.object(sys, "argv", [str(SCRIPT_PATH), "--query-history"]), \
|
||
mock.patch.object(self.module, "_mysql_connect", return_value=connection), \
|
||
mock.patch.object(self.module, "_http_post_json") as send, \
|
||
contextlib.redirect_stdout(io.StringIO()) as output:
|
||
self.assertEqual(self.module.main(), 1)
|
||
self.assertIn("查询失败", output.getvalue())
|
||
self.assertFalse(output.getvalue().endswith("ended"))
|
||
connection.close.assert_called_once()
|
||
send.assert_not_called()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|