|
|
|
@@ -211,6 +211,10 @@ class RoutingTests(unittest.TestCase):
|
|
|
|
|
MODULE.classify_question("/steam 在线"),
|
|
|
|
|
("local", "local_steam_friends"),
|
|
|
|
|
)
|
|
|
|
|
self.assertEqual(
|
|
|
|
|
MODULE.classify_question("/知识问答"),
|
|
|
|
|
("local", "local_trivia_start"),
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
def test_reply_has_one_suffix_within_delivery_limit(self):
|
|
|
|
|
self.assertEqual(MODULE.clean_reply("收到。"), "收到喵")
|
|
|
|
@@ -308,6 +312,24 @@ class LedgerAndApplicationTests(unittest.TestCase):
|
|
|
|
|
"第二条回复喵",
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
def test_local_game_command_result_is_available_to_followup_chat(self):
|
|
|
|
|
observed_context = []
|
|
|
|
|
|
|
|
|
|
def complete(**kwargs):
|
|
|
|
|
observed_context.extend(kwargs["context"])
|
|
|
|
|
return "我看到了喵"
|
|
|
|
|
|
|
|
|
|
self.application.deepseek.complete = complete
|
|
|
|
|
self.application.accept(event("version-context", "@JWeBot /version"))
|
|
|
|
|
local = self.application.accept(event("coin-context", "@JWeBot /翻硬币"))
|
|
|
|
|
self.application.accept(event("chat-context", "@JWeBot 刚才是什么结果"))
|
|
|
|
|
self.application.process_direct_chat_once()
|
|
|
|
|
|
|
|
|
|
self.assertIn(("/翻硬币", local["text"]), observed_context)
|
|
|
|
|
self.assertTrue(
|
|
|
|
|
all(question != "/version" for question, _reply in observed_context)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
def test_route_and_ack_telemetry_is_privacy_safe(self):
|
|
|
|
|
observed = []
|
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
|
|
@@ -450,6 +472,12 @@ class LedgerAndApplicationTests(unittest.TestCase):
|
|
|
|
|
os.environ,
|
|
|
|
|
{"JWEBOT_RUNTIME_PYTHON": str(runtime_python)},
|
|
|
|
|
), \
|
|
|
|
|
mock.patch.object(
|
|
|
|
|
MODULE.random,
|
|
|
|
|
"uniform",
|
|
|
|
|
return_value=0.75,
|
|
|
|
|
) as uniform, \
|
|
|
|
|
mock.patch.object(MODULE.time, "sleep") as sleep, \
|
|
|
|
|
mock.patch.object(
|
|
|
|
|
MODULE.subprocess,
|
|
|
|
|
"run",
|
|
|
|
@@ -482,6 +510,11 @@ class LedgerAndApplicationTests(unittest.TestCase):
|
|
|
|
|
},
|
|
|
|
|
],
|
|
|
|
|
)
|
|
|
|
|
uniform.assert_called_once_with(
|
|
|
|
|
MODULE.MULTI_REPLY_DELAY_MIN_SECONDS,
|
|
|
|
|
MODULE.MULTI_REPLY_DELAY_MAX_SECONDS,
|
|
|
|
|
)
|
|
|
|
|
sleep.assert_called_once_with(0.75)
|
|
|
|
|
|
|
|
|
|
def test_two_direct_generations_run_in_parallel(self):
|
|
|
|
|
entered = []
|
|
|
|
@@ -996,9 +1029,9 @@ class LedgerAndApplicationTests(unittest.TestCase):
|
|
|
|
|
self.assertEqual(version["route"], "local")
|
|
|
|
|
self.assertIn("JWeBot 版本", version["text"])
|
|
|
|
|
self.assertIn("kernel", version["text"])
|
|
|
|
|
self.assertIn("最新版本:beta 1.2.1.1019", version["text"])
|
|
|
|
|
self.assertIn("更新时间:2026-10-19 17:23:58", version["text"])
|
|
|
|
|
self.assertIn("更新摘要:修复 Taffy 短句拆分只换行不分条", version["text"])
|
|
|
|
|
self.assertIn("最新版本:beta 1.2.2.1019", version["text"])
|
|
|
|
|
self.assertIn("更新时间:2026-10-19 17:31:56", version["text"])
|
|
|
|
|
self.assertIn("更新摘要:修复本地命令后的闲聊断上下文", version["text"])
|
|
|
|
|
self.assertEqual(debug["route"], "local")
|
|
|
|
|
self.assertIn("调试埋点", debug["text"])
|
|
|
|
|
self.assertEqual(balance["route"], "local")
|
|
|
|
@@ -1342,6 +1375,33 @@ class LedgerAndApplicationTests(unittest.TestCase):
|
|
|
|
|
account = ledger._load_player_account("group-a", str(person["person_key"]))
|
|
|
|
|
self.assertEqual(account["balance"], 1000 + reward)
|
|
|
|
|
|
|
|
|
|
def test_trivia_quiz_can_be_started_by_local_command(self):
|
|
|
|
|
self.application.deepseek.complete = mock.Mock(side_effect=AssertionError)
|
|
|
|
|
|
|
|
|
|
first = self.application.accept(
|
|
|
|
|
event("trivia-force", "@JWeBot /知识问答")
|
|
|
|
|
)
|
|
|
|
|
second = self.application.accept(
|
|
|
|
|
event("trivia-force-existing", "@JWeBot /知识问答")
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
self.assertEqual(first["route"], "local")
|
|
|
|
|
self.assertEqual(first["execution_path"], "local-command")
|
|
|
|
|
self.assertIn("冷门知识问答", first["text"])
|
|
|
|
|
self.assertIn("回答格式:/回答 A", first["text"])
|
|
|
|
|
self.assertIn("已经有一题在进行中", second["text"])
|
|
|
|
|
self.assertEqual(
|
|
|
|
|
self.application.ledger.row("trivia-force")["status"],
|
|
|
|
|
"local_completed",
|
|
|
|
|
)
|
|
|
|
|
self.assertIsNotNone(
|
|
|
|
|
self.application.ledger.local_game(
|
|
|
|
|
"group-a",
|
|
|
|
|
MODULE.TRIVIA_GAME_KEY,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
self.application.deepseek.complete.assert_not_called()
|
|
|
|
|
|
|
|
|
|
def test_ambient_chat_cooldown_keeps_next_hour_scan_broadcast(self):
|
|
|
|
|
ledger = self.application.ledger
|
|
|
|
|
observed_at = MODULE.datetime(
|
|
|
|
|