3328 lines
121 KiB
Python
3328 lines
121 KiB
Python
#!/usr/bin/env python3
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
os.environ["JWEBOT_CONFIG_ROOT"] = str(
|
||
PROJECT_ROOT / ".test-config-does-not-exist"
|
||
)
|
||
sys.path[:0] = [
|
||
str(PROJECT_ROOT / "packages" / "jwebot-config" / "src"),
|
||
str(PROJECT_ROOT / "packages" / "jwebot-sdk" / "src"),
|
||
str(PROJECT_ROOT / "packages" / "jwebot-official-extensions" / "src"),
|
||
str(PROJECT_ROOT / "packages" / "jwebot-kernel" / "src"),
|
||
]
|
||
|
||
from jwebot_kernel import edge_runtime as MODULE # noqa: E402
|
||
|
||
|
||
TEST_PRIVATE_CONTEXT = {
|
||
"bot": {
|
||
"mention": "JWeBot",
|
||
"identity_note": "You are JWeBot, an isolated test assistant.",
|
||
},
|
||
"people": [
|
||
{
|
||
"canonical_name": "Alice",
|
||
"aliases": ["alice", "a1", "alice-ocr"],
|
||
"prefixes": ["alice-"],
|
||
"identity_fact": "Alice is a configured test profile.",
|
||
"display_id": "alice",
|
||
},
|
||
{
|
||
"canonical_name": "Bob",
|
||
"aliases": ["bob", "b1", "bob-ocr"],
|
||
"prefixes": ["bob-"],
|
||
"identity_fact": "Bob is a configured test profile.",
|
||
"display_id": "bob",
|
||
},
|
||
{
|
||
"canonical_name": "Carol",
|
||
"aliases": ["carol", "c1", "carol-ocr"],
|
||
"prefixes": [],
|
||
"identity_fact": "Carol is a configured test profile.",
|
||
"display_id": "carol",
|
||
},
|
||
],
|
||
"clock_scan_targets": [
|
||
{"canonical_name": "Alice", "display_id": "alice"},
|
||
{"canonical_name": "Bob", "display_id": "bob"},
|
||
{"canonical_name": "Carol", "display_id": "carol"},
|
||
],
|
||
}
|
||
|
||
|
||
def config(
|
||
directory: Path,
|
||
*,
|
||
api_key: str = "",
|
||
model: str = "deepseek-flash",
|
||
steam_api_key: str = "",
|
||
steam_id64: str = "",
|
||
steam_roster: tuple[dict[str, object], ...] = (),
|
||
local_extensions_enabled: bool = True,
|
||
extension_mode: str = "in-process",
|
||
extension_transport: str = "unix",
|
||
extension_state_root: Path | None = None,
|
||
extension_allowed_signers: Path | None = None,
|
||
extension_publisher: str = "",
|
||
extension_fallback_enabled: bool = True,
|
||
):
|
||
MODULE.apply_private_context(TEST_PRIVATE_CONTEXT)
|
||
return MODULE.EdgeConfig(
|
||
socket_path=directory / "edge.sock",
|
||
database_path=directory / "edge.sqlite3",
|
||
cloud_host="cloud.example",
|
||
cloud_agent="wechat-responder",
|
||
deepseek_key=api_key,
|
||
deepseek_base_url="https://api.deepseek.com",
|
||
deepseek_model=model,
|
||
memory_remote_path="/home/node/.openclaw/workspace/MEMORY.md",
|
||
delivery_enabled=True,
|
||
steam_api_key=steam_api_key,
|
||
steam_id64=steam_id64,
|
||
steam_proxy_url=MODULE.DEFAULT_STEAM_MIHOMO_PROXY_URL,
|
||
steam_roster=steam_roster,
|
||
local_extensions_enabled=local_extensions_enabled,
|
||
extension_mode=extension_mode,
|
||
extension_transport=extension_transport,
|
||
extension_state_root=(
|
||
extension_state_root
|
||
if extension_state_root is not None
|
||
else directory / "extensions"
|
||
),
|
||
extension_allowed_signers=(
|
||
extension_allowed_signers
|
||
if extension_allowed_signers is not None
|
||
else directory / "allowed_signers"
|
||
),
|
||
extension_publisher=extension_publisher,
|
||
extension_fallback_enabled=extension_fallback_enabled,
|
||
)
|
||
|
||
|
||
def event(
|
||
event_id: str = "event-1",
|
||
text: str = "@JWeBot 你好",
|
||
*,
|
||
sender: str = "Alice",
|
||
group: str = "group-a",
|
||
) -> dict[str, str]:
|
||
return {
|
||
"event_id": event_id,
|
||
"group_fingerprint": group,
|
||
"sender": sender,
|
||
"text": text,
|
||
}
|
||
|
||
|
||
class RoutingTests(unittest.TestCase):
|
||
def test_hourly_scan_pool_covers_all_configured_group_ids(self):
|
||
MODULE.apply_private_context(TEST_PRIVATE_CONTEXT)
|
||
self.assertEqual(
|
||
{label for _, label in MODULE.CLOCK_SCAN_TARGETS},
|
||
{"alice", "bob", "carol"},
|
||
)
|
||
|
||
def test_private_steam_roster_does_not_change_alias_migration_version(self):
|
||
MODULE.apply_private_context(TEST_PRIVATE_CONTEXT)
|
||
alias_version = MODULE.PERSON_ALIAS_MIGRATION_VERSION
|
||
private_context = json.loads(json.dumps(TEST_PRIVATE_CONTEXT))
|
||
private_context["people"][0].update(
|
||
{
|
||
"steam_id64": "10000000000000001",
|
||
"steam_aliases": ["alice-steam"],
|
||
}
|
||
)
|
||
try:
|
||
MODULE.apply_private_context(private_context)
|
||
self.assertEqual(MODULE.PERSON_ALIAS_MIGRATION_VERSION, alias_version)
|
||
self.assertEqual(
|
||
MODULE.STEAM_GROUP_ROSTER,
|
||
(
|
||
{
|
||
"steam_id": "10000000000000001",
|
||
"label": "alice",
|
||
"selectors": (
|
||
"Alice",
|
||
"alice",
|
||
"a1",
|
||
"alice-ocr",
|
||
"alice-",
|
||
"alice-steam",
|
||
),
|
||
},
|
||
),
|
||
)
|
||
finally:
|
||
MODULE.apply_private_context(TEST_PRIVATE_CONTEXT)
|
||
|
||
def test_hash_requests_route_to_cloud(self):
|
||
self.assertEqual(MODULE.classify_question("#status"), ("cloud", "hash_command"))
|
||
|
||
def test_status_and_version_are_local_commands(self):
|
||
self.assertEqual(
|
||
MODULE.classify_question("/status"),
|
||
("local", "local_server_status"),
|
||
)
|
||
self.assertEqual(
|
||
MODULE.classify_question("/version"),
|
||
("local", "local_jwebot_version"),
|
||
)
|
||
self.assertEqual(
|
||
MODULE.classify_question("/版本"),
|
||
("local", "local_jwebot_version"),
|
||
)
|
||
|
||
def test_unknown_slash_command_stays_local(self):
|
||
self.assertEqual(
|
||
MODULE.classify_question("/unknown"),
|
||
("local", "local_unknown"),
|
||
)
|
||
|
||
def test_ordinary_chat_stays_direct(self):
|
||
self.assertEqual(MODULE.classify_question("今天怎么样"), ("chat", "direct_chat"))
|
||
|
||
def test_local_game_commands_stay_on_the_edge(self):
|
||
self.assertEqual(
|
||
MODULE.classify_question("/丢骰子 2d6"),
|
||
("local", "local_dice"),
|
||
)
|
||
self.assertEqual(
|
||
MODULE.classify_question("/功能列表"),
|
||
("local", "local_feature_list"),
|
||
)
|
||
self.assertEqual(
|
||
MODULE.classify_question("/丢骰子5D20"),
|
||
("local", "local_dice"),
|
||
)
|
||
self.assertEqual(
|
||
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("收到。"), "收到喵")
|
||
self.assertEqual(MODULE.clean_reply("收到喵～"), "收到喵")
|
||
self.assertEqual(MODULE.clean_reply("第一行\n第二行"), "第一行\n第二行喵")
|
||
self.assertEqual(len(MODULE.clean_reply("好" * 900)), 900)
|
||
self.assertEqual(MODULE.FAST_ACKNOWLEDGEMENT, "喵......")
|
||
|
||
def test_chat_reply_style_removes_ai_like_padding(self):
|
||
self.assertEqual(
|
||
MODULE.clean_chat_reply(
|
||
"李神是啥我都不清楚，怎么比，总不能凭空给自己打满分吧\n"
|
||
"反正我就是个会唠嗑的，聊聊天、记记小账还行，"
|
||
"别的功能想都别想喵"
|
||
),
|
||
"李神是啥喵",
|
||
)
|
||
self.assertEqual(
|
||
MODULE.clean_chat_reply("作为AI我无法比较。\n李神是谁？"),
|
||
"李神是谁喵",
|
||
)
|
||
self.assertEqual(
|
||
MODULE.clean_chat_reply("你说“一切”是啥意思喵？做题就 /回答 A喵"),
|
||
"你说“一切”是啥意思喵",
|
||
)
|
||
self.assertEqual(
|
||
MODULE.unpack_reply_parts(MODULE.clean_chat_reply(
|
||
"这波像是脑子知道要撤，但是手完全没同意，"
|
||
"而且还顺手把队友也拖下水了。"
|
||
)),
|
||
[
|
||
"这波像是脑子知道要撤喵",
|
||
"但是手完全没同意喵",
|
||
"而且还顺手把队友也拖下水了喵",
|
||
],
|
||
)
|
||
|
||
def test_profile_memory_rejects_bare_versions_but_keeps_audio_context(self):
|
||
self.assertFalse(MODULE.profile_fact_is_safe("5.1"))
|
||
self.assertTrue(MODULE.profile_fact_is_safe("最近在研究 5.1 音响配置"))
|
||
|
||
def test_local_extension_flag_accepts_legacy_environment_alias(self):
|
||
with mock.patch.object(
|
||
MODULE,
|
||
"load_environment_file",
|
||
), mock.patch.dict(
|
||
os.environ,
|
||
{"JWEBOT_OFFICIAL_EXTENSIONS_ENABLED": "0"},
|
||
clear=True,
|
||
):
|
||
edge_config = MODULE.EdgeConfig.from_environment()
|
||
self.assertFalse(edge_config.local_extensions_enabled)
|
||
|
||
def test_local_extension_flag_prefers_canonical_environment_key(self):
|
||
with mock.patch.object(
|
||
MODULE,
|
||
"load_environment_file",
|
||
), mock.patch.dict(
|
||
os.environ,
|
||
{
|
||
"JWEBOT_LOCAL_EXTENSIONS_ENABLED": "1",
|
||
"JWEBOT_OFFICIAL_EXTENSIONS_ENABLED": "0",
|
||
},
|
||
clear=True,
|
||
):
|
||
edge_config = MODULE.EdgeConfig.from_environment()
|
||
self.assertTrue(edge_config.local_extensions_enabled)
|
||
|
||
|
||
class LedgerAndApplicationTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.temporary = tempfile.TemporaryDirectory()
|
||
self.config = config(Path(self.temporary.name))
|
||
self.application = MODULE.EdgeApplication(self.config)
|
||
|
||
def tearDown(self):
|
||
self.application.close()
|
||
self.temporary.cleanup()
|
||
|
||
def test_chat_is_completed_locally_and_context_is_reused(self):
|
||
observed_context = []
|
||
|
||
def complete(**kwargs):
|
||
observed_context.extend(kwargs["context"])
|
||
return "第一条回复喵" if not observed_context else "第二条回复喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
first = self.application.accept(event())
|
||
self.application.process_direct_chat_once()
|
||
second = self.application.accept(event("event-2", "@JWeBot 接着说"))
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertEqual(first["text"], MODULE.FAST_ACKNOWLEDGEMENT)
|
||
self.assertEqual(second["text"], MODULE.FAST_ACKNOWLEDGEMENT)
|
||
self.assertEqual(len(observed_context), 1)
|
||
self.assertEqual(observed_context[0]["speaker"], "Alice")
|
||
self.assertEqual(observed_context[0]["user_text"], "你好")
|
||
self.assertEqual(observed_context[0]["taffy_text"], "第一条回复喵")
|
||
self.assertIn("same_sender", observed_context[0]["tags"])
|
||
self.assertEqual(
|
||
self.application.ledger.row("event-2")["response_text"],
|
||
"第二条回复喵",
|
||
)
|
||
|
||
def test_conversation_context_is_isolated_by_fingerprint(self):
|
||
observed = {}
|
||
|
||
def complete(**kwargs):
|
||
observed[kwargs["event_id"]] = list(kwargs["context"])
|
||
return f"{kwargs['event_id']} 回复喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
for event_id, group, text in (
|
||
("group-first", "group-a", "@JWeBot 群里的第一条"),
|
||
("direct-first", "direct-a", "私聊里的第一条"),
|
||
("group-second", "group-a", "@JWeBot 群里继续"),
|
||
("direct-second", "direct-a", "私聊继续"),
|
||
):
|
||
self.application.accept(event(event_id, text, group=group))
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertEqual(
|
||
[(item["speaker"], item["user_text"], item["taffy_text"])
|
||
for item in observed["group-second"]],
|
||
[("Alice", "群里的第一条", "group-first 回复喵")],
|
||
)
|
||
self.assertEqual(
|
||
[(item["speaker"], item["user_text"], item["taffy_text"])
|
||
for item in observed["direct-second"]],
|
||
[("Alice", "私聊里的第一条", "direct-first 回复喵")],
|
||
)
|
||
|
||
def test_shadow_event_is_rejected_before_routing_or_persistence(self):
|
||
response = self.application.accept(
|
||
{
|
||
**event("shadow-1", "/状态", group="direct-a"),
|
||
"shadow_only": True,
|
||
"dispatch_allowed": False,
|
||
}
|
||
)
|
||
|
||
self.assertEqual(
|
||
response,
|
||
{"status": "ignored", "reason": "shadow_only"},
|
||
)
|
||
self.assertIsNone(self.application.ledger.row("shadow-1"))
|
||
|
||
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.assertTrue(
|
||
any(
|
||
item["route"] == "local"
|
||
and item["route_reason"] == "local_coin"
|
||
and item["speaker"] == "Alice"
|
||
and item["user_text"] == "/翻硬币"
|
||
and item["taffy_text"] == local["text"]
|
||
for item in observed_context
|
||
)
|
||
)
|
||
self.assertTrue(
|
||
all(item["user_text"] != "/version" for item in observed_context)
|
||
)
|
||
|
||
def test_self_referential_context_does_not_cross_senders(self):
|
||
observed_context = []
|
||
|
||
def complete(**kwargs):
|
||
observed_context[:] = list(kwargs["context"])
|
||
return "我只看你的喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(event("alice-first", "@JWeBot Alice 的事", sender="Alice"))
|
||
self.application.process_direct_chat_once()
|
||
self.application.accept(event("bob-first", "@JWeBot Bob 的事", sender="Bob"))
|
||
self.application.process_direct_chat_once()
|
||
self.application.accept(event("bob-follow", "@JWeBot 我刚才说啥", sender="Bob"))
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertTrue(observed_context)
|
||
self.assertEqual(
|
||
{item["speaker"] for item in observed_context},
|
||
{"Bob"},
|
||
)
|
||
self.assertTrue(
|
||
all("Alice 的事" not in item["user_text"] for item in observed_context)
|
||
)
|
||
|
||
def test_mentioned_person_context_can_cross_senders(self):
|
||
observed_context = []
|
||
|
||
def complete(**kwargs):
|
||
observed_context[:] = list(kwargs["context"])
|
||
return "看 Bob 那句喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(event("bob-mentioned", "@JWeBot Bob 的事", sender="Bob"))
|
||
self.application.process_direct_chat_once()
|
||
self.application.accept(
|
||
event("alice-asks-bob", "@JWeBot Bob 刚才说啥", sender="Alice")
|
||
)
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertTrue(
|
||
any(
|
||
item["speaker"] == "Bob"
|
||
and item["user_text"] == "Bob 的事"
|
||
and "mentioned_person" in item["tags"]
|
||
for item in observed_context
|
||
)
|
||
)
|
||
|
||
def test_low_confidence_sender_does_not_receive_identity_memory(self):
|
||
captured = {}
|
||
|
||
def complete(**kwargs):
|
||
captured.update(kwargs)
|
||
return "先确认是谁喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(
|
||
{
|
||
**event("low-conf-chat", "@JWeBot 我是谁", sender="alice"),
|
||
"sender_confidence": 0.42,
|
||
"body_confidence": 0.98,
|
||
"source": "rapidocr-local",
|
||
}
|
||
)
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertEqual(captured["person_memory"], ())
|
||
self.assertIsNone(
|
||
self.application.ledger.person_profile(
|
||
"group-a",
|
||
MODULE.person_scope_key("group-a", "Alice"),
|
||
)
|
||
)
|
||
|
||
def test_low_confidence_identity_bound_local_command_is_rejected(self):
|
||
response = self.application.accept(
|
||
{
|
||
**event("low-conf-draw", "@JWeBot /抽奖 10", sender="alice"),
|
||
"sender_confidence": 0.42,
|
||
"body_confidence": 0.98,
|
||
"source": "rapidocr-local",
|
||
}
|
||
)
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertIn("没认稳", response["text"])
|
||
self.assertEqual(
|
||
self.application.ledger.row("low-conf-draw")["status"],
|
||
"local_rejected",
|
||
)
|
||
|
||
def test_delivered_taffy_broadcast_is_available_to_context(self):
|
||
self.application.ledger.reserve_clock_broadcast(
|
||
group_fingerprint="group-a",
|
||
slot_key="hourly:2026-10-21T17",
|
||
local_date="2026-10-21",
|
||
kind="ambient_chat",
|
||
message="我刚刚路过说了一句喵",
|
||
)
|
||
self.assertTrue(
|
||
self.application.ledger.complete_clock_broadcast(
|
||
"group-a",
|
||
"hourly:2026-10-21T17",
|
||
)
|
||
)
|
||
|
||
context = self.application.ledger.conversation_context(
|
||
"group-a",
|
||
exclude_event_id="current",
|
||
sender="Alice",
|
||
question="Taffy 刚刚说了什么",
|
||
)
|
||
|
||
self.assertTrue(
|
||
any(
|
||
item["speaker"] == "Taffy"
|
||
and item["taffy_text"] == "我刚刚路过说了一句喵"
|
||
and "taffy_broadcast" in item["tags"]
|
||
for item in context
|
||
)
|
||
)
|
||
|
||
def test_local_game_status_question_answers_without_direct_chat(self):
|
||
person = self.application.ledger.resolve_person("group-a", "Bob")
|
||
account = MODULE.normalize_player_account({})
|
||
account["game"] = {
|
||
"bet": 50,
|
||
"deck": [1, 2, 3],
|
||
"player": [5, 10],
|
||
"dealer": [8, 2],
|
||
"updated_at": time.time(),
|
||
"index": 1,
|
||
}
|
||
with self.application.ledger.lock, self.application.ledger.connection:
|
||
self.application.ledger._save_player_account(
|
||
"group-a",
|
||
str(person["person_key"]),
|
||
account,
|
||
MODULE.utc_now(),
|
||
)
|
||
self.application.deepseek.complete = mock.Mock(
|
||
side_effect=AssertionError("direct chat should not run")
|
||
)
|
||
|
||
response = self.application.accept(
|
||
event("bob-local-status", "@JWeBot Bob 在玩什么吗", sender="Alice")
|
||
)
|
||
|
||
self.assertEqual(response["text"].count("喵"), 1)
|
||
self.assertIn("bob 在玩单人21点 #1", response["text"])
|
||
self.assertIn("16点", response["text"])
|
||
self.assertEqual(response["execution_path"], "local-command")
|
||
self.assertEqual(
|
||
self.application.ledger.row("bob-local-status")["status"],
|
||
"chat_completed",
|
||
)
|
||
self.assertEqual(
|
||
self.application.ledger.status()["direct_chat_pending"],
|
||
0,
|
||
)
|
||
self.application.deepseek.complete.assert_not_called()
|
||
|
||
def test_short_evaluation_uses_active_taffy_thread(self):
|
||
observed_context = []
|
||
|
||
def complete(**kwargs):
|
||
observed_context[:] = list(kwargs["context"])
|
||
return "他那手确实挺敢喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
self.application.ledger.reserve_clock_broadcast(
|
||
group_fingerprint="group-a",
|
||
slot_key="hourly:2026-10-21T18",
|
||
local_date="2026-10-21",
|
||
kind="ambient_chat",
|
||
message="21点那把16点还敢停牌喵",
|
||
)
|
||
self.assertTrue(
|
||
self.application.ledger.complete_clock_broadcast(
|
||
"group-a",
|
||
"hourly:2026-10-21T18",
|
||
)
|
||
)
|
||
|
||
self.application.accept(
|
||
event("short-active-thread", "@JWeBot 还真是 胆子大", sender="Alice")
|
||
)
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertTrue(
|
||
any(
|
||
"active_thread" in item["tags"]
|
||
and "16点还敢停牌" in item["taffy_text"]
|
||
for item in observed_context
|
||
)
|
||
)
|
||
|
||
def test_route_and_ack_telemetry_is_privacy_safe(self):
|
||
observed = []
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
telemetry = MODULE.RuntimeTelemetry(
|
||
component="edge",
|
||
observer=observed.append,
|
||
logger=mock.Mock(),
|
||
)
|
||
application = MODULE.EdgeApplication(
|
||
config(Path(directory)),
|
||
telemetry=telemetry,
|
||
)
|
||
try:
|
||
incoming = event(
|
||
"private-event-id",
|
||
"@JWeBot 这是私密消息",
|
||
group="private-group",
|
||
)
|
||
response = application.accept(incoming)
|
||
application.confirm_fast_acknowledgement(
|
||
response["event_id"],
|
||
delivered=True,
|
||
)
|
||
finally:
|
||
application.close()
|
||
|
||
names = [value["name"] for value in observed]
|
||
self.assertIn("edge.route.classified", names)
|
||
self.assertIn("edge.ack.prepared", names)
|
||
self.assertIn("edge.ack.confirmed", names)
|
||
self.assertIn("edge.fast_ack.prepared", names)
|
||
self.assertIn("edge.fast_ack.confirmed", names)
|
||
serialized = json.dumps(
|
||
[dict(value) for value in observed],
|
||
ensure_ascii=False,
|
||
)
|
||
self.assertNotIn("private-event-id", serialized)
|
||
self.assertNotIn("private-group", serialized)
|
||
self.assertNotIn("这是私密消息", serialized)
|
||
classified = next(
|
||
value
|
||
for value in observed
|
||
if value["name"] == "edge.route.classified"
|
||
)
|
||
self.assertEqual(classified["execution_path"], "deepseek")
|
||
|
||
def test_queue_wait_and_end_to_end_delivery_are_measured(self):
|
||
observed = []
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
telemetry = MODULE.RuntimeTelemetry(
|
||
component="edge",
|
||
observer=observed.append,
|
||
logger=mock.Mock(),
|
||
)
|
||
application = MODULE.EdgeApplication(
|
||
config(Path(directory)),
|
||
telemetry=telemetry,
|
||
)
|
||
application.deepseek.complete = mock.Mock(
|
||
return_value="最终回复喵"
|
||
)
|
||
application.deliver = mock.Mock()
|
||
try:
|
||
response = application.accept(event())
|
||
application.confirm_fast_acknowledgement(
|
||
response["event_id"],
|
||
delivered=True,
|
||
)
|
||
self.assertTrue(
|
||
application.process_direct_chat_generation_once()
|
||
)
|
||
application.deliver.assert_not_called()
|
||
self.assertTrue(
|
||
application.process_direct_chat_delivery_once()
|
||
)
|
||
finally:
|
||
application.close()
|
||
|
||
by_name = {
|
||
value["name"]: value
|
||
for value in observed
|
||
if value["name"]
|
||
in {"edge.queue.wait", "edge.delivery.end_to_end"}
|
||
}
|
||
self.assertIn("edge.queue.wait", by_name)
|
||
self.assertIn("edge.delivery.end_to_end", by_name)
|
||
self.assertGreaterEqual(by_name["edge.queue.wait"]["elapsed_ms"], 0)
|
||
self.assertGreaterEqual(
|
||
by_name["edge.delivery.end_to_end"]["elapsed_ms"],
|
||
by_name["edge.queue.wait"]["elapsed_ms"],
|
||
)
|
||
|
||
def test_missing_direct_key_acknowledges_then_retries_without_losing_event(self):
|
||
response = self.application.accept(event())
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertEqual(response["text"], MODULE.FAST_ACKNOWLEDGEMENT)
|
||
stored = self.application.ledger.row("event-1")
|
||
self.assertEqual(stored["status"], "chat_retrying")
|
||
|
||
def test_direct_chat_starts_before_ack_but_final_delivery_waits_for_ack(self):
|
||
self.application.deepseek.complete = mock.Mock(return_value="最终回复喵")
|
||
self.application.deliver = mock.Mock()
|
||
|
||
response = self.application.accept(event())
|
||
self.assertTrue(response["fast_ack"])
|
||
self.assertTrue(self.application.process_direct_chat_once())
|
||
|
||
with self.application.ledger.lock:
|
||
state = self.application.ledger.connection.execute(
|
||
"SELECT state FROM direct_chat_outbox WHERE event_id = ?",
|
||
("event-1",),
|
||
).fetchone()["state"]
|
||
self.assertEqual(state, "result_waiting_ack")
|
||
self.application.deepseek.complete.assert_called_once()
|
||
self.application.deliver.assert_not_called()
|
||
|
||
self.application.ledger.mark_direct_chat_acknowledged(
|
||
"event-1",
|
||
delivered=True,
|
||
)
|
||
self.assertTrue(self.application.process_direct_chat_once())
|
||
|
||
self.application.deliver.assert_called_once()
|
||
self.assertEqual(
|
||
self.application.ledger.row("event-1")["status"],
|
||
"chat_delivered",
|
||
)
|
||
|
||
def test_delivery_sends_multi_reply_as_separate_helper_calls(self):
|
||
runtime_python = Path(self.temporary.name) / "python"
|
||
delivery_helper = Path(self.temporary.name) / "jwebot_edge_delivery.py"
|
||
runtime_python.write_text("# runtime\n", encoding="utf-8")
|
||
delivery_helper.write_text("# helper\n", encoding="utf-8")
|
||
result = mock.Mock(returncode=0, stdout="", stderr="")
|
||
reply = MODULE.pack_reply_parts(["第一段喵", "第二段喵"])
|
||
|
||
with mock.patch.object(MODULE, "DELIVERY_HELPER", delivery_helper), \
|
||
mock.patch.dict(
|
||
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",
|
||
return_value=result,
|
||
) as run:
|
||
self.application.deliver(
|
||
{
|
||
"event_id": "event-1",
|
||
"group_fingerprint": "group-a",
|
||
"result_text": reply,
|
||
}
|
||
)
|
||
|
||
payloads = [
|
||
json.loads(call.kwargs["input"])
|
||
for call in run.call_args_list
|
||
]
|
||
self.assertEqual(
|
||
payloads,
|
||
[
|
||
{
|
||
"event_id": "event-1:part-1",
|
||
"group_fingerprint": "group-a",
|
||
"reply": "第一段喵",
|
||
},
|
||
{
|
||
"event_id": "event-1:part-2",
|
||
"group_fingerprint": "group-a",
|
||
"reply": "第二段喵",
|
||
},
|
||
],
|
||
)
|
||
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 = []
|
||
entered_lock = threading.Lock()
|
||
both_started = threading.Event()
|
||
release = threading.Event()
|
||
outcomes = []
|
||
errors = []
|
||
|
||
def complete(**kwargs):
|
||
with entered_lock:
|
||
entered.append(kwargs["event_id"])
|
||
if len(entered) == 2:
|
||
both_started.set()
|
||
if not release.wait(5):
|
||
raise MODULE.DeepSeekUnavailable("parallel test timed out")
|
||
return f"{kwargs['event_id']} 回复喵"
|
||
|
||
def generate():
|
||
try:
|
||
outcomes.append(
|
||
self.application.process_direct_chat_generation_once()
|
||
)
|
||
except BaseException as error:
|
||
errors.append(error)
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(event("parallel-1", "@JWeBot 第一条"))
|
||
self.application.accept(event("parallel-2", "@JWeBot 第二条"))
|
||
workers = [threading.Thread(target=generate) for _ in range(2)]
|
||
for worker in workers:
|
||
worker.start()
|
||
try:
|
||
self.assertTrue(both_started.wait(3))
|
||
finally:
|
||
release.set()
|
||
for worker in workers:
|
||
worker.join(5)
|
||
|
||
self.assertFalse(errors)
|
||
self.assertEqual(outcomes, [True, True])
|
||
self.assertEqual(set(entered), {"parallel-1", "parallel-2"})
|
||
self.assertTrue(all(not worker.is_alive() for worker in workers))
|
||
self.application.deliver.assert_not_called()
|
||
|
||
def test_out_of_order_generation_still_delivers_in_event_order(self):
|
||
first_started = threading.Event()
|
||
release_first = threading.Event()
|
||
second_finished = threading.Event()
|
||
delivered = []
|
||
errors = []
|
||
|
||
def complete(**kwargs):
|
||
if kwargs["event_id"] == "order-1":
|
||
first_started.set()
|
||
if not release_first.wait(5):
|
||
raise MODULE.DeepSeekUnavailable("ordering test timed out")
|
||
else:
|
||
second_finished.set()
|
||
return f"{kwargs['event_id']} 回复喵"
|
||
|
||
def generate():
|
||
try:
|
||
self.application.process_direct_chat_generation_once()
|
||
except BaseException as error:
|
||
errors.append(error)
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = lambda row: delivered.append(row["event_id"])
|
||
for event_id in ("order-1", "order-2"):
|
||
response = self.application.accept(
|
||
event(event_id, f"@JWeBot {event_id}")
|
||
)
|
||
self.application.confirm_fast_acknowledgement(
|
||
response["event_id"],
|
||
delivered=True,
|
||
)
|
||
|
||
first_worker = threading.Thread(target=generate)
|
||
second_worker = threading.Thread(target=generate)
|
||
first_worker.start()
|
||
try:
|
||
self.assertTrue(first_started.wait(3))
|
||
second_worker.start()
|
||
self.assertTrue(second_finished.wait(3))
|
||
second_worker.join(5)
|
||
self.assertFalse(second_worker.is_alive())
|
||
self.assertFalse(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
finally:
|
||
release_first.set()
|
||
first_worker.join(5)
|
||
if second_worker.ident is not None:
|
||
second_worker.join(5)
|
||
|
||
self.assertFalse(errors)
|
||
self.assertFalse(first_worker.is_alive())
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
self.assertEqual(delivered, ["order-1", "order-2"])
|
||
|
||
def test_failed_ack_during_generation_is_terminal_for_ordering(self):
|
||
first_started = threading.Event()
|
||
release_first = threading.Event()
|
||
delivered = []
|
||
errors = []
|
||
|
||
def complete(**kwargs):
|
||
if kwargs["event_id"] == "terminal-1":
|
||
first_started.set()
|
||
if not release_first.wait(5):
|
||
raise MODULE.DeepSeekUnavailable("ACK race test timed out")
|
||
return f"{kwargs['event_id']} 回复喵"
|
||
|
||
def generate_first():
|
||
try:
|
||
self.application.process_direct_chat_generation_once()
|
||
except BaseException as error:
|
||
errors.append(error)
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = lambda row: delivered.append(row["event_id"])
|
||
first = self.application.accept(
|
||
event("terminal-1", "@JWeBot 第一条")
|
||
)
|
||
second = self.application.accept(
|
||
event("terminal-2", "@JWeBot 第二条")
|
||
)
|
||
self.application.confirm_fast_acknowledgement(
|
||
second["event_id"],
|
||
delivered=True,
|
||
)
|
||
|
||
first_worker = threading.Thread(target=generate_first)
|
||
first_worker.start()
|
||
try:
|
||
self.assertTrue(first_started.wait(3))
|
||
self.application.confirm_fast_acknowledgement(
|
||
first["event_id"],
|
||
delivered=False,
|
||
error_code="RuntimeError",
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_generation_once()
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
finally:
|
||
release_first.set()
|
||
first_worker.join(5)
|
||
|
||
self.assertFalse(errors)
|
||
self.assertFalse(first_worker.is_alive())
|
||
self.assertEqual(delivered, ["terminal-2"])
|
||
self.assertFalse(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
self.assertEqual(
|
||
self.application.ledger.row("terminal-1")["status"],
|
||
"chat_ack_failed",
|
||
)
|
||
|
||
def test_direct_generation_and_delivery_retries_are_independent(self):
|
||
self.application.deepseek.complete = mock.Mock(
|
||
side_effect=[
|
||
MODULE.DeepSeekUnavailable("temporary provider failure"),
|
||
"最终回复喵",
|
||
]
|
||
)
|
||
self.application.deliver = mock.Mock(
|
||
side_effect=[
|
||
MODULE.EdgeError("temporary GUI failure"),
|
||
None,
|
||
]
|
||
)
|
||
response = self.application.accept(event())
|
||
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_generation_once()
|
||
)
|
||
with self.application.ledger.lock, self.application.ledger.connection:
|
||
self.application.ledger.connection.execute(
|
||
"""
|
||
UPDATE direct_chat_outbox
|
||
SET next_attempt_at = 0
|
||
WHERE event_id = ?
|
||
""",
|
||
("event-1",),
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_generation_once()
|
||
)
|
||
self.application.confirm_fast_acknowledgement(
|
||
response["event_id"],
|
||
delivered=True,
|
||
)
|
||
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
with self.application.ledger.lock, self.application.ledger.connection:
|
||
self.application.ledger.connection.execute(
|
||
"""
|
||
UPDATE direct_chat_outbox
|
||
SET next_attempt_at = 0
|
||
WHERE event_id = ?
|
||
""",
|
||
("event-1",),
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
|
||
with self.application.ledger.lock:
|
||
row = self.application.ledger.connection.execute(
|
||
"""
|
||
SELECT state, generation_attempts, delivery_attempts
|
||
FROM direct_chat_outbox
|
||
WHERE event_id = ?
|
||
""",
|
||
("event-1",),
|
||
).fetchone()
|
||
self.assertEqual(
|
||
(
|
||
row["state"],
|
||
row["generation_attempts"],
|
||
row["delivery_attempts"],
|
||
),
|
||
("delivered", 2, 2),
|
||
)
|
||
self.assertEqual(self.application.deepseek.complete.call_count, 2)
|
||
self.assertEqual(self.application.deliver.call_count, 2)
|
||
|
||
def test_direct_duplicate_generates_and_delivers_exactly_once(self):
|
||
self.application.deepseek.complete = mock.Mock(return_value="最终回复喵")
|
||
self.application.deliver = mock.Mock()
|
||
|
||
first = self.application.accept(event())
|
||
duplicate = self.application.accept(event())
|
||
|
||
self.assertTrue(duplicate["duplicate"])
|
||
self.assertEqual(
|
||
self.application.ledger.status()["direct_chat_pending"],
|
||
1,
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_generation_once()
|
||
)
|
||
self.assertFalse(
|
||
self.application.process_direct_chat_generation_once()
|
||
)
|
||
self.application.confirm_fast_acknowledgement(
|
||
first["event_id"],
|
||
delivered=True,
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
self.assertFalse(
|
||
self.application.process_direct_chat_delivery_once()
|
||
)
|
||
self.application.deepseek.complete.assert_called_once()
|
||
self.application.deliver.assert_called_once()
|
||
|
||
def test_semantic_duplicate_with_new_event_id_is_ignored(self):
|
||
first = self.application.accept(
|
||
{**event("event-1", "@JWeBot /金币"), "source": "rapidocr-local"}
|
||
)
|
||
duplicate = self.application.accept(
|
||
{**event("event-2", "@JWeBot /金币"), "source": "rapidocr-local"}
|
||
)
|
||
|
||
self.assertEqual(first["status"], "reply")
|
||
self.assertEqual(duplicate["status"], "ignored")
|
||
self.assertTrue(duplicate["duplicate"])
|
||
with self.application.ledger.lock:
|
||
count = self.application.ledger.connection.execute(
|
||
"SELECT COUNT(*) FROM events"
|
||
).fetchone()[0]
|
||
self.assertEqual(count, 1)
|
||
|
||
def test_resource_mutation_semantic_duplicate_uses_longer_window(self):
|
||
first = self.application.accept(
|
||
{
|
||
**event("event-1", "@JWeBot /原石祈愿"),
|
||
"source": "rapidocr-local",
|
||
}
|
||
)
|
||
old_timestamp = (
|
||
MODULE.datetime.now(MODULE.timezone.utc)
|
||
- MODULE.timedelta(seconds=30)
|
||
).isoformat(timespec="milliseconds")
|
||
with self.application.ledger.lock, self.application.ledger.connection:
|
||
self.application.ledger.connection.execute(
|
||
"""
|
||
UPDATE events
|
||
SET created_at = ?, updated_at = ?
|
||
WHERE event_id = ?
|
||
""",
|
||
(old_timestamp, old_timestamp, "event-1"),
|
||
)
|
||
duplicate = self.application.accept(
|
||
{
|
||
**event("event-2", "@JWeBot /原石祈愿"),
|
||
"source": "rapidocr-local",
|
||
}
|
||
)
|
||
|
||
self.assertEqual(first["status"], "reply")
|
||
self.assertEqual(duplicate["status"], "ignored")
|
||
self.assertTrue(duplicate["duplicate"])
|
||
with self.application.ledger.lock:
|
||
count = self.application.ledger.connection.execute(
|
||
"SELECT COUNT(*) FROM events"
|
||
).fetchone()[0]
|
||
self.assertEqual(count, 1)
|
||
|
||
def test_failed_fast_ack_never_delivers_the_generated_reply(self):
|
||
self.application.deepseek.complete = mock.Mock(return_value="最终回复喵")
|
||
self.application.deliver = mock.Mock()
|
||
|
||
self.application.accept(event())
|
||
self.assertTrue(self.application.process_direct_chat_once())
|
||
self.application.ledger.mark_direct_chat_acknowledged(
|
||
"event-1",
|
||
delivered=False,
|
||
error_code="RuntimeError",
|
||
)
|
||
|
||
self.assertFalse(self.application.process_direct_chat_once())
|
||
self.application.deliver.assert_not_called()
|
||
self.assertEqual(
|
||
self.application.ledger.row("event-1")["status"],
|
||
"chat_ack_failed",
|
||
)
|
||
|
||
def test_feature_list_is_local_and_does_not_call_a_model(self):
|
||
self.application.deepseek.complete = mock.Mock(side_effect=AssertionError)
|
||
|
||
response = self.application.accept(event(text="@JWeBot /功能列表"))
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertEqual(response["execution_path"], "local-command")
|
||
self.assertIn("由内核和插件描述自动生成", response["text"])
|
||
self.assertIn("/丢骰子", response["text"])
|
||
self.assertIn("骰子和多人骰子挑战", response["text"])
|
||
self.assertIn("/steam", response["text"])
|
||
self.assertIn("[DeepSeek]", response["text"])
|
||
self.assertIn("[OpenClaw]", response["text"])
|
||
self.assertEqual(
|
||
self.application.ledger.row("event-1")["status"],
|
||
"local_completed",
|
||
)
|
||
self.application.deepseek.complete.assert_not_called()
|
||
|
||
def test_response_contract_names_all_three_execution_paths(self):
|
||
local = self.application.accept(
|
||
event("local-path", "@JWeBot /翻硬币")
|
||
)
|
||
deepseek = self.application.accept(
|
||
event("deepseek-path", "@JWeBot 普通聊天")
|
||
)
|
||
openclaw = self.application.accept(
|
||
event("openclaw-path", "@JWeBot #云端任务")
|
||
)
|
||
|
||
self.assertEqual(local["execution_path"], "local-command")
|
||
self.assertEqual(deepseek["execution_path"], "deepseek")
|
||
self.assertEqual(openclaw["execution_path"], "openclaw")
|
||
|
||
def test_local_adapter_path_is_used_by_default(self):
|
||
original = self.application.extensions.handle
|
||
self.application.extensions.handle = mock.Mock(side_effect=original)
|
||
|
||
response = self.application.accept(event(text="@JWeBot /翻硬币"))
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.application.extensions.handle.assert_called_once()
|
||
|
||
def test_local_adapter_path_has_a_compatibility_rollback(self):
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
application = MODULE.EdgeApplication(
|
||
config(
|
||
Path(directory),
|
||
local_extensions_enabled=False,
|
||
)
|
||
)
|
||
try:
|
||
application.extensions.handle = mock.Mock(
|
||
side_effect=AssertionError
|
||
)
|
||
response = application.accept(
|
||
event(text="@JWeBot /翻硬币")
|
||
)
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertIn("硬币落地", response["text"])
|
||
application.extensions.handle.assert_not_called()
|
||
finally:
|
||
application.close()
|
||
|
||
def test_kernel_commands_bypass_local_extensions(self):
|
||
self.application.extensions.handle = mock.Mock(side_effect=AssertionError)
|
||
self.application.local.balance_provider = mock.Mock(
|
||
return_value=[
|
||
{
|
||
"currency": "CNY",
|
||
"total_balance": "12.50",
|
||
"granted_balance": "2.50",
|
||
"topped_up_balance": "10.00",
|
||
}
|
||
]
|
||
)
|
||
with mock.patch.object(
|
||
MODULE.LocalCommandEngine,
|
||
"service_state",
|
||
return_value="active",
|
||
):
|
||
for event_id, command in (
|
||
("direct-help", "/功能列表"),
|
||
("direct-balance", "/余额"),
|
||
):
|
||
with self.subTest(command=command):
|
||
response = self.application.accept(
|
||
event(event_id, f"@JWeBot {command}")
|
||
)
|
||
self.assertEqual(response["route"], "local")
|
||
self.application.extensions.handle.assert_not_called()
|
||
|
||
def test_steam_acknowledges_then_delivers_without_a_model(self):
|
||
self.application.deepseek.complete = mock.Mock(side_effect=AssertionError)
|
||
self.application.local.steam = mock.Mock()
|
||
self.application.local.steam.render.return_value = (
|
||
"Steam：在线 1/3，游戏中 1；Alice 正在玩 Dota 2"
|
||
)
|
||
self.application.deliver = mock.Mock()
|
||
|
||
response = self.application.accept(event(text="@JWeBot /steam"))
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertEqual(response["text"], MODULE.STEAM_ACKNOWLEDGEMENT)
|
||
self.assertTrue(response["fast_ack"])
|
||
self.assertTrue(self.application.process_direct_chat_once())
|
||
self.application.local.steam.render.assert_called_once_with("")
|
||
self.application.deepseek.complete.assert_not_called()
|
||
self.application.deliver.assert_not_called()
|
||
with self.application.ledger.lock:
|
||
state = self.application.ledger.connection.execute(
|
||
"""
|
||
SELECT kind, state FROM direct_chat_outbox WHERE event_id = ?
|
||
""",
|
||
("event-1",),
|
||
).fetchone()
|
||
self.assertEqual((state["kind"], state["state"]), ("steam", "result_waiting_ack"))
|
||
|
||
self.application.confirm_fast_acknowledgement(
|
||
"event-1",
|
||
delivered=True,
|
||
)
|
||
self.assertTrue(self.application.process_direct_chat_once())
|
||
|
||
self.application.deliver.assert_called_once()
|
||
self.assertEqual(
|
||
self.application.ledger.row("event-1")["status"],
|
||
"steam_delivered",
|
||
)
|
||
|
||
def test_local_status_and_balance_commands_do_not_call_a_model(self):
|
||
self.application.deepseek.complete = mock.Mock(side_effect=AssertionError)
|
||
with mock.patch.object(
|
||
MODULE.LocalCommandEngine,
|
||
"service_state",
|
||
return_value="active",
|
||
), mock.patch.object(
|
||
MODULE.DeepSeekClient,
|
||
"balance",
|
||
return_value=[
|
||
{
|
||
"currency": "CNY",
|
||
"total_balance": "12.50",
|
||
"granted_balance": "2.50",
|
||
"topped_up_balance": "10.00",
|
||
}
|
||
],
|
||
):
|
||
status = self.application.accept(
|
||
event("local-status", "@JWeBot /状态")
|
||
)
|
||
server_status = self.application.accept(
|
||
event("server-status", "@JWeBot /status")
|
||
)
|
||
version = self.application.accept(
|
||
event("jwebot-version", "@JWeBot /version")
|
||
)
|
||
debug = self.application.accept(
|
||
event("debug-telemetry", "@JWeBot /debug")
|
||
)
|
||
balance = self.application.accept(
|
||
event("local-balance", "@JWeBot /余额")
|
||
)
|
||
|
||
self.assertEqual(status["route"], "local")
|
||
self.assertIn("Edge active", status["text"])
|
||
self.assertEqual(server_status["route"], "local")
|
||
self.assertIn("Edge active", server_status["text"])
|
||
self.assertEqual(version["route"], "local")
|
||
self.assertIn("JWeBot 版本", version["text"])
|
||
self.assertIn("kernel", version["text"])
|
||
self.assertIn("最新版本：beta 2.0.2.1021", version["text"])
|
||
self.assertIn("更新时间：2026-10-21 18:11:35", version["text"])
|
||
self.assertIn("更新摘要：上下文追问小版本更新完成", version["text"])
|
||
self.assertEqual(debug["route"], "local")
|
||
self.assertIn("调试埋点", debug["text"])
|
||
self.assertEqual(balance["route"], "local")
|
||
self.assertIn("CNY 总 12.50", balance["text"])
|
||
self.application.deepseek.complete.assert_not_called()
|
||
|
||
def test_dice_is_deterministic_and_idempotent(self):
|
||
first = self.application.accept(event(text="@JWeBot /丢骰子 2d6"))
|
||
second = self.application.accept(event(text="@JWeBot /丢骰子 2d6"))
|
||
|
||
self.assertEqual(first["route"], "local")
|
||
self.assertEqual(first["text"], second["text"])
|
||
self.assertTrue(second["duplicate"])
|
||
self.assertIn("骰子 2d6", first["text"])
|
||
|
||
def test_local_slash_commands_accept_parameters_without_spaces(self):
|
||
dice = self.application.accept(event(text="@JWeBot /丢骰子5D20"))
|
||
unknown = self.application.accept(
|
||
event("unknown-local", "@JWeBot /不存在的命令")
|
||
)
|
||
|
||
self.assertEqual(dice["route"], "local")
|
||
self.assertIn("骰子 5d20", dice["text"])
|
||
self.assertEqual(unknown["route"], "local")
|
||
self.assertIn("本地命令不存在", unknown["text"])
|
||
self.assertEqual(
|
||
self.application.ledger.status()["outbox_pending"],
|
||
0,
|
||
)
|
||
|
||
def test_dice_challenge_is_local_and_records_winners_by_canonical_identity(self):
|
||
started = self.application.accept(
|
||
event(
|
||
"challenge-alice",
|
||
"@JWeBot /丢骰子挑战",
|
||
sender="alice",
|
||
)
|
||
)
|
||
joined = self.application.accept(
|
||
event(
|
||
"challenge-gb",
|
||
"@JWeBot /丢骰子挑战",
|
||
sender="Bob",
|
||
)
|
||
)
|
||
state = self.application.ledger.local_game("group-a", "dice_challenge")
|
||
settled = self.application.accept(
|
||
event(
|
||
"challenge-settle",
|
||
"@JWeBot /丢骰子挑战结算",
|
||
sender="Alice",
|
||
)
|
||
)
|
||
|
||
self.assertEqual(started["route"], "local")
|
||
self.assertIn("挑战掷出", started["text"])
|
||
self.assertIn("bob 挑战掷出", joined["text"])
|
||
self.assertEqual(state["dice_count"], 5)
|
||
self.assertEqual(state["faces"], 6)
|
||
self.assertEqual(len(state["entries"]), 2)
|
||
self.assertIn("结算", settled["text"])
|
||
self.assertIn("1.", settled["text"])
|
||
self.assertIn("冠军", settled["text"])
|
||
self.assertIsNone(
|
||
self.application.ledger.local_game("group-a", "dice_challenge")
|
||
)
|
||
achievements = self.application.ledger.dice_challenge_achievements("group-a")
|
||
self.assertTrue(
|
||
any(entry["achievement"] == "总点冠军" for entry in achievements)
|
||
)
|
||
|
||
def test_dice_challenge_accepts_no_space_5d20_specification(self):
|
||
response = self.application.accept(
|
||
event(
|
||
"challenge-5d20",
|
||
"@JWeBot /丢骰子挑战5D20",
|
||
sender="Alice",
|
||
group="group-b",
|
||
)
|
||
)
|
||
state = self.application.ledger.local_game("group-b", "dice_challenge")
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertEqual(state["dice_count"], 5)
|
||
self.assertEqual(state["faces"], 20)
|
||
|
||
def test_dice_challenge_stays_open_for_one_hour_and_reports_timeout(self):
|
||
self.application.accept(
|
||
event(
|
||
"challenge-timeout",
|
||
"@JWeBot /丢骰子挑战",
|
||
sender="Alice",
|
||
)
|
||
)
|
||
state = self.application.ledger.local_game("group-a", "dice_challenge")
|
||
self.assertGreaterEqual(
|
||
float(state["expires_at"]) - MODULE.time.time(),
|
||
MODULE.DICE_CHALLENGE_TIMEOUT_SECONDS - 2,
|
||
)
|
||
state["expires_at"] = MODULE.time.time() - 1
|
||
self.application.ledger.save_local_game("group-a", "dice_challenge", state)
|
||
|
||
response = self.application.accept(
|
||
event(
|
||
"challenge-timeout-settle",
|
||
"@JWeBot /丢骰子挑战结算",
|
||
sender="Alice",
|
||
)
|
||
)
|
||
|
||
self.assertIn("已超时", response["text"])
|
||
|
||
def test_dice_challenge_records_five_straight_immediately(self):
|
||
response = self.application.accept(
|
||
event(
|
||
"straight-2",
|
||
"@JWeBot /丢骰子挑战",
|
||
sender="Bob",
|
||
)
|
||
)
|
||
achievements = self.application.ledger.dice_challenge_achievements("group-a")
|
||
|
||
self.assertIn("五连号达成", response["text"])
|
||
self.assertTrue(
|
||
any(
|
||
entry["display_name"] == "bob"
|
||
and entry["achievement"] == "五连号"
|
||
and "2-3-4-5-6" in entry["details"]
|
||
for entry in achievements
|
||
)
|
||
)
|
||
|
||
def test_local_game_events_stay_out_of_cloud_memory_projection(self):
|
||
self.application.accept(event(text="@JWeBot /丢骰子"))
|
||
|
||
projection = MODULE.MemoryProjector(
|
||
self.application.ledger,
|
||
self.config,
|
||
).render()
|
||
|
||
self.assertNotIn("丢骰子", projection)
|
||
|
||
def test_hourly_clock_broadcast_is_local_and_daily_ranking_uses_deliveries(self):
|
||
ledger = self.application.ledger
|
||
for event_id, sender in (("clock-a", "Alice"), ("clock-b", "Bob")):
|
||
self.assertTrue(
|
||
ledger.record_person_observation(
|
||
event_id=event_id,
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text="群内聊天记录",
|
||
observed_at=MODULE.utc_now(),
|
||
source="test",
|
||
)
|
||
)
|
||
self.application.deepseek.complete = mock.Mock(side_effect=AssertionError)
|
||
self.application.deliver = mock.Mock()
|
||
hourly_now = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
3,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=hourly_now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
hourly_delivery = self.application.deliver.call_args.args[0]
|
||
self.assertIn("现在是北京时间 3:00", hourly_delivery["result_text"])
|
||
self.assertIn("扫码了", hourly_delivery["result_text"])
|
||
self.assertFalse(
|
||
self.application.process_clock_once(
|
||
now=hourly_now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
ranking = ledger.clock_scan_ranking("group-a", "2026-10-13")
|
||
self.assertEqual(len(ranking), 1)
|
||
self.assertEqual(ranking[0]["scan_count"], 1)
|
||
|
||
midnight = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
14,
|
||
0,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=midnight,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
daily_delivery = self.application.deliver.call_args.args[0]
|
||
self.assertIn("北京时间 0:00", daily_delivery["result_text"])
|
||
self.assertIn("扫码排行榜", daily_delivery["result_text"])
|
||
self.assertIn(ranking[0]["display_name"], daily_delivery["result_text"])
|
||
self.application.deepseek.complete.assert_not_called()
|
||
|
||
def test_low_frequency_ambient_chat_uses_recent_context(self):
|
||
ledger = self.application.ledger
|
||
observed_at = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
12,
|
||
42,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
).isoformat()
|
||
for index, (sender, text) in enumerate(
|
||
(
|
||
("Alice", "今天这个加班安排是不是有点离谱"),
|
||
("Bob", "我也觉得还要继续修 bug"),
|
||
("Alice", "先把问题处理完再说"),
|
||
("Bob", "感觉咖啡已经救不了我了"),
|
||
),
|
||
start=1,
|
||
):
|
||
self.assertTrue(
|
||
ledger.record_person_observation(
|
||
event_id=f"ambient-observation-{index}",
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text=text,
|
||
observed_at=observed_at,
|
||
source="test",
|
||
)
|
||
)
|
||
ledger.record_repeat_history(
|
||
event_id=f"ambient-context-{index}",
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text=text,
|
||
observed_at=observed_at,
|
||
)
|
||
self.application.deliver = mock.Mock()
|
||
now = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
13,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
|
||
with mock.patch.object(MODULE, "TRIVIA_QUIZ_PROBABILITY_PERCENT", 0), \
|
||
mock.patch.object(MODULE, "AMBIENT_CHAT_PROBABILITY_PERCENT", 100):
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
|
||
delivery = self.application.deliver.call_args.args[0]
|
||
self.assertIn("加班味", delivery["result_text"])
|
||
row = ledger.connection.execute(
|
||
"""
|
||
SELECT kind, state FROM clock_broadcasts
|
||
WHERE group_fingerprint = ? AND slot_key = ?
|
||
""",
|
||
("group-a", "hourly:2026-10-13T13"),
|
||
).fetchone()
|
||
self.assertEqual(row["kind"], "ambient_chat")
|
||
self.assertEqual(row["state"], "delivered")
|
||
self.assertFalse(
|
||
self.application.process_clock_once(
|
||
now=now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
|
||
def test_trivia_quiz_broadcast_and_answer_rewards_mora(self):
|
||
bank = MODULE.trivia_question_bank()
|
||
self.assertGreater(len(bank), 200)
|
||
display_text = "\n".join(
|
||
"\n".join(
|
||
[
|
||
str(question["domain"]),
|
||
str(question["question"]),
|
||
*[str(option) for option in question["options"]],
|
||
str(question["explanation"]),
|
||
]
|
||
)
|
||
for question in bank
|
||
)
|
||
self.assertIn("新牛仔（NewJeans）", display_text)
|
||
self.assertIn("碧碧（BiBi）", display_text)
|
||
self.assertNotIn("地平线一直想回去见到的儿子", display_text)
|
||
self.assertNotIn("蒙德风神的真实魔神名", display_text)
|
||
topics = {str(question["topic"]) for question in bank}
|
||
self.assertIn("缪斯学园偶像", topics)
|
||
self.assertIn("我的世界", topics)
|
||
self.assertEqual(
|
||
MODULE.canonical_trivia_topic("LoveLive!"),
|
||
"缪斯学园偶像",
|
||
)
|
||
self.assertEqual(
|
||
MODULE.canonical_trivia_topic("Minecraft"),
|
||
"我的世界",
|
||
)
|
||
ledger = self.application.ledger
|
||
observed_at = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
12,
|
||
42,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
).isoformat()
|
||
for sender in ("Alice", "Bob"):
|
||
ledger.record_person_observation(
|
||
event_id=f"trivia-member-{sender}",
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text="冷门题准备一下",
|
||
observed_at=observed_at,
|
||
source="test",
|
||
)
|
||
self.application.deliver = mock.Mock()
|
||
now = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
13,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
|
||
with mock.patch.object(MODULE, "TRIVIA_QUIZ_PROBABILITY_PERCENT", 100):
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
|
||
delivery = self.application.deliver.call_args.args[0]
|
||
self.assertIn("冷门知识问答", delivery["result_text"])
|
||
self.assertIn("限时5分钟", delivery["result_text"])
|
||
self.assertIn("/回答 A", delivery["result_text"])
|
||
state = ledger.local_game("group-a", MODULE.TRIVIA_GAME_KEY)
|
||
self.assertIsNotNone(state)
|
||
assert state is not None
|
||
state["expires_at"] = time.time() + 300
|
||
ledger.save_local_game("group-a", MODULE.TRIVIA_GAME_KEY, state)
|
||
answer = str(state["answer"])
|
||
reward = int(state["reward"])
|
||
result = self.application.accept(
|
||
event(
|
||
"trivia-answer",
|
||
f"@JWeBot /回答 {answer}",
|
||
sender="Alice",
|
||
)
|
||
)
|
||
|
||
self.assertEqual(result["route"], "local")
|
||
self.assertIn("答对了", result["text"])
|
||
self.assertIn(f"奖励摩拉 +{reward}", result["text"])
|
||
self.assertIsNone(ledger.local_game("group-a", MODULE.TRIVIA_GAME_KEY))
|
||
person = ledger.resolve_person("group-a", "Alice")
|
||
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)
|
||
|
||
unknown = self.application.accept(
|
||
event("trivia-force-unknown", "@JWeBot /知识问答 星际")
|
||
)
|
||
first = self.application.accept(
|
||
event("trivia-force", "@JWeBot /知识问答 LoveLive!")
|
||
)
|
||
second = self.application.accept(
|
||
event("trivia-force-existing", "@JWeBot /知识问答 Minecraft")
|
||
)
|
||
|
||
self.assertIn("支持主题：", unknown["text"])
|
||
self.assertIn("原神", unknown["text"])
|
||
self.assertIn("少女乐团派对", unknown["text"])
|
||
self.assertIn("缪斯学园偶像", unknown["text"])
|
||
self.assertIn("我的世界", unknown["text"])
|
||
self.assertEqual(first["route"], "local")
|
||
self.assertEqual(first["execution_path"], "local-command")
|
||
self.assertIn("冷门知识问答", first["text"])
|
||
self.assertIn("缪斯学园偶像", first["text"])
|
||
self.assertIn("回答格式：/回答 A", first["text"])
|
||
self.assertIn("已经有一题在进行中", second["text"])
|
||
self.assertIn("缪斯学园偶像", second["text"])
|
||
self.assertEqual(
|
||
self.application.ledger.row("trivia-force")["status"],
|
||
"local_completed",
|
||
)
|
||
state = self.application.ledger.local_game(
|
||
"group-a",
|
||
MODULE.TRIVIA_GAME_KEY,
|
||
)
|
||
self.assertIsNotNone(state)
|
||
assert state is not None
|
||
self.assertEqual(state["topic"], "缪斯学园偶像")
|
||
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(
|
||
2026,
|
||
10,
|
||
13,
|
||
12,
|
||
45,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
).isoformat()
|
||
for index, (sender, text) in enumerate(
|
||
(
|
||
("Alice", "今天加班真的有点顶不住"),
|
||
("Bob", "先把这个线上问题看完"),
|
||
("Alice", "咖啡续上"),
|
||
("Bob", "修完再开一把游戏"),
|
||
),
|
||
start=1,
|
||
):
|
||
ledger.record_person_observation(
|
||
event_id=f"ambient-cooldown-observation-{index}",
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text=text,
|
||
observed_at=observed_at,
|
||
source="test",
|
||
)
|
||
ledger.record_repeat_history(
|
||
event_id=f"ambient-cooldown-context-{index}",
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text=text,
|
||
observed_at=observed_at,
|
||
)
|
||
self.application.deliver = mock.Mock()
|
||
first = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
13,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
second = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
14,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
|
||
with mock.patch.object(MODULE, "TRIVIA_QUIZ_PROBABILITY_PERCENT", 0), \
|
||
mock.patch.object(MODULE, "AMBIENT_CHAT_PROBABILITY_PERCENT", 100):
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=first,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=second,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
|
||
second_delivery = self.application.deliver.call_args.args[0]
|
||
self.assertIn("扫码了", second_delivery["result_text"])
|
||
row = ledger.connection.execute(
|
||
"""
|
||
SELECT kind, state FROM clock_broadcasts
|
||
WHERE group_fingerprint = ? AND slot_key = ?
|
||
""",
|
||
("group-a", "hourly:2026-10-13T14"),
|
||
).fetchone()
|
||
self.assertEqual(row["kind"], "hourly")
|
||
self.assertEqual(row["state"], "delivered")
|
||
|
||
def test_failed_hourly_broadcast_does_not_increment_scan_ranking(self):
|
||
ledger = self.application.ledger
|
||
for event_id, sender in (("clock-c", "Alice"), ("clock-d", "Bob")):
|
||
ledger.record_person_observation(
|
||
event_id=event_id,
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text="群内聊天记录",
|
||
observed_at=MODULE.utc_now(),
|
||
source="test",
|
||
)
|
||
self.application.deliver = mock.Mock(side_effect=MODULE.EdgeError("chat changed"))
|
||
now = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
4,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
self.assertEqual(ledger.clock_scan_ranking("group-a", "2026-10-13"), [])
|
||
self.assertEqual(ledger.status()["clock_broadcasts_pending"], 0)
|
||
|
||
def test_existing_pending_hourly_broadcast_is_delivered(self):
|
||
ledger = self.application.ledger
|
||
for event_id, sender in (("clock-pending-a", "Alice"), ("clock-pending-b", "Bob")):
|
||
ledger.record_person_observation(
|
||
event_id=event_id,
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text="群内聊天记录",
|
||
observed_at=MODULE.utc_now(),
|
||
source="test",
|
||
)
|
||
person = ledger.resolve_person("group-a", "Bob")
|
||
self.assertTrue(
|
||
ledger.reserve_clock_broadcast(
|
||
group_fingerprint="group-a",
|
||
slot_key="hourly:2026-10-13T05",
|
||
local_date="2026-10-13",
|
||
kind="hourly",
|
||
message="现在是北京时间 5:00，bob 扫码了。",
|
||
scan_person_key=str(person["person_key"]),
|
||
scan_label="bob",
|
||
)
|
||
)
|
||
self.application.deliver = mock.Mock()
|
||
now = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
5,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
|
||
self.assertTrue(
|
||
self.application.process_clock_once(
|
||
now=now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
)
|
||
|
||
delivery = self.application.deliver.call_args.args[0]
|
||
self.assertIn("bob 扫码了", delivery["result_text"])
|
||
self.assertEqual(
|
||
ledger.clock_scan_ranking("group-a", "2026-10-13")[0]["display_name"],
|
||
"bob",
|
||
)
|
||
|
||
def test_scan_boost_pity_forces_target_after_two_misses(self):
|
||
ledger = self.application.ledger
|
||
for event_id, sender in (("pity-member-a", "Alice"), ("pity-member-b", "Bob")):
|
||
ledger.record_person_observation(
|
||
event_id=event_id,
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text="群内聊天记录",
|
||
observed_at=MODULE.utc_now(),
|
||
source="test",
|
||
)
|
||
alice = ledger.resolve_person("group-a", "Alice")
|
||
bob = ledger.resolve_person("group-a", "Bob")
|
||
carol = ledger.resolve_person("group-a", "Carol")
|
||
account = MODULE.normalize_player_account({})
|
||
account["scan_boosts"] = [
|
||
{
|
||
"target_person_key": str(bob["person_key"]),
|
||
"target_display_name": "bob",
|
||
"weight": 5,
|
||
"created_at": MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
3,
|
||
30,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
).timestamp(),
|
||
"expires_at": MODULE.datetime(
|
||
2099,
|
||
10,
|
||
14,
|
||
3,
|
||
30,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
).timestamp(),
|
||
}
|
||
]
|
||
ledger._save_player_account(
|
||
"group-a",
|
||
str(alice["person_key"]),
|
||
account,
|
||
MODULE.utc_now(),
|
||
)
|
||
for hour, person, label in (
|
||
(4, alice, "alice"),
|
||
(5, carol, "carol"),
|
||
):
|
||
slot_key = f"hourly:2026-10-13T{hour:02d}"
|
||
self.assertTrue(
|
||
ledger.reserve_clock_broadcast(
|
||
group_fingerprint="group-a",
|
||
slot_key=slot_key,
|
||
local_date="2026-10-13",
|
||
kind="hourly",
|
||
message="test",
|
||
scan_person_key=str(person["person_key"]),
|
||
scan_label=label,
|
||
)
|
||
)
|
||
self.assertTrue(ledger.complete_clock_broadcast("group-a", slot_key))
|
||
now = MODULE.datetime(
|
||
2026,
|
||
10,
|
||
13,
|
||
6,
|
||
0,
|
||
tzinfo=MODULE.BEIJING_TIMEZONE,
|
||
)
|
||
|
||
status = ledger.scan_boost_pity_status("group-a", local_now=now)
|
||
self.assertTrue(status[0]["guaranteed"])
|
||
self.assertEqual(status[0]["display_name"], "bob")
|
||
broadcast = self.application.clock.due_broadcast(
|
||
now=now,
|
||
group_fingerprint="group-a",
|
||
)
|
||
|
||
self.assertIsNotNone(broadcast)
|
||
self.assertEqual(broadcast.scan_label, "bob")
|
||
|
||
def test_scan_ranking_command_uses_current_beijing_date(self):
|
||
ledger = self.application.ledger
|
||
local_date = MODULE.datetime.now(MODULE.BEIJING_TIMEZONE).date().isoformat()
|
||
person = ledger.resolve_person("group-a", "b1")
|
||
self.assertTrue(
|
||
ledger.reserve_clock_broadcast(
|
||
group_fingerprint="group-a",
|
||
slot_key=f"test-hourly:{local_date}",
|
||
local_date=local_date,
|
||
kind="hourly",
|
||
message="test",
|
||
scan_person_key=str(person["person_key"]),
|
||
scan_label="bob",
|
||
)
|
||
)
|
||
self.assertTrue(
|
||
ledger.complete_clock_broadcast(
|
||
"group-a",
|
||
f"test-hourly:{local_date}",
|
||
)
|
||
)
|
||
|
||
response = self.application.accept(
|
||
event("scan-ranking", "@JWeBot /扫码排行", sender="Alice")
|
||
)
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertIn("北京时间今日扫码排行", response["text"])
|
||
self.assertIn("bob 1 次", response["text"])
|
||
self.assertIn("alice 0 次", response["text"])
|
||
self.assertIn("carol 0 次", response["text"])
|
||
self.assertIn("下一小时扫码权重：", response["text"])
|
||
self.assertIn("alice 1", response["text"])
|
||
self.assertIn("bob 1", response["text"])
|
||
self.assertIn("carol 1", response["text"])
|
||
|
||
def test_repeat_ranking_tracks_origin_and_later_repeats_by_display_id(self):
|
||
ledger = self.application.ledger
|
||
observed_at = MODULE.datetime.now(MODULE.BEIJING_TIMEZONE).isoformat()
|
||
for event_id, sender, text in (
|
||
("repeat-origin", "Alice", "我先说一句测试"),
|
||
("repeat-second", "Bob", "我先说一句测试！"),
|
||
("repeat-third", "b1", "我 先说一句测试"),
|
||
):
|
||
self.assertTrue(
|
||
ledger.record_repeat_history(
|
||
event_id=event_id,
|
||
group_fingerprint="group-a",
|
||
sender=sender,
|
||
text=text,
|
||
observed_at=observed_at,
|
||
)
|
||
)
|
||
|
||
response = self.application.accept(
|
||
event("repeat-ranking", "@JWeBot /复读榜", sender="Alice")
|
||
)
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertIn("首发 alice", response["text"])
|
||
self.assertIn("后续复读 2 次", response["text"])
|
||
self.assertIn("bob×2", response["text"])
|
||
|
||
searched = self.application.accept(
|
||
event("history-search", "@JWeBot /查记录 先说", sender="Alice")
|
||
)
|
||
self.assertEqual(searched["route"], "local")
|
||
self.assertIn("本地记录命中“先说”", searched["text"])
|
||
self.assertIn("alice：我先说一句测试", searched["text"])
|
||
|
||
def test_explicit_person_memory_is_local_and_injected_into_reply_context(self):
|
||
remembered = self.application.accept(
|
||
event("remember", "@JWeBot /记住 我更喜欢家庭服务器和技术细节")
|
||
)
|
||
captured = {}
|
||
|
||
def complete(**kwargs):
|
||
captured.update(kwargs)
|
||
return "收到喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(event("chat-after-remember", "@JWeBot 今天聊什么"))
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertEqual(remembered["route"], "local")
|
||
self.assertIn("家庭服务器", captured["person_memory"][0])
|
||
profile = self.application.ledger.person_profile(
|
||
"group-a",
|
||
MODULE.person_scope_key("group-a", "Alice"),
|
||
)
|
||
self.assertEqual(profile["facts"][0]["source"], "explicit")
|
||
|
||
def test_person_memory_can_be_viewed_and_forgotten(self):
|
||
self.application.accept(
|
||
event("remember", "@JWeBot /记住 我喜欢简短回复")
|
||
)
|
||
viewed = self.application.accept(
|
||
event("view", "@JWeBot /人物记忆 Alice")
|
||
)
|
||
forgotten = self.application.accept(
|
||
event("forget", "@JWeBot /忘记 全部")
|
||
)
|
||
|
||
self.assertIn("简短回复", viewed["text"])
|
||
self.assertIn("已清空", forgotten["text"])
|
||
self.assertIsNone(
|
||
self.application.ledger.person_profile(
|
||
"group-a",
|
||
MODULE.person_scope_key("group-a", "Alice"),
|
||
)
|
||
)
|
||
|
||
def test_automatic_profile_requires_evidence_and_filters_unsuitable_fact(self):
|
||
engine = MODULE.PersonMemoryEngine(self.application.ledger)
|
||
for index in range(3):
|
||
incoming = event(
|
||
f"observation-{index}",
|
||
"@JWeBot 我最近在研究家庭服务器和 NAS",
|
||
)
|
||
row, _ = self.application.ledger.record_event(
|
||
incoming,
|
||
question=MODULE.event_question(incoming),
|
||
route="chat",
|
||
route_reason="direct_chat",
|
||
)
|
||
engine.ingest(row, incoming)
|
||
person_key = MODULE.person_scope_key("group-a", "Alice")
|
||
self.application.ledger.queue_profile_job(
|
||
"group-a",
|
||
person_key,
|
||
delay_seconds=0,
|
||
)
|
||
client = mock.Mock()
|
||
client.profile_facts.return_value = [
|
||
{
|
||
"text": "持续研究家庭服务器和 NAS",
|
||
"kind": "topic",
|
||
"evidence_count": 3,
|
||
},
|
||
{
|
||
"text": "工资很高",
|
||
"kind": "topic",
|
||
"evidence_count": 3,
|
||
},
|
||
{
|
||
"text": "群里今晚八点一起打游戏",
|
||
"kind": "ongoing",
|
||
"evidence_count": 3,
|
||
},
|
||
{
|
||
"text": "打游戏比较菜",
|
||
"kind": "topic",
|
||
"evidence_count": 3,
|
||
},
|
||
]
|
||
|
||
self.assertTrue(engine.process_once(client))
|
||
|
||
profile = self.application.ledger.person_profile("group-a", person_key)
|
||
facts = [fact["text"] for fact in profile["facts"]]
|
||
self.assertIn("Alice is a configured test profile.", facts)
|
||
self.assertIn("持续研究家庭服务器和 NAS", facts)
|
||
self.assertNotIn("工资很高", facts)
|
||
self.assertNotIn("群里今晚八点一起打游戏", facts)
|
||
self.assertIn("打游戏比较菜", facts)
|
||
|
||
def test_automatic_profile_collapses_audio_memory_paraphrases(self):
|
||
facts = MODULE.PersonMemoryEngine.merge_facts(
|
||
existing=[],
|
||
generated=[
|
||
{
|
||
"text": "常和群友讨论家庭影院、功放和音响配置，常劝人上 5.1 声道",
|
||
"kind": "topic",
|
||
"evidence_count": 3,
|
||
},
|
||
{
|
||
"text": "经常在群里劝人上 5.1 声道家庭影院",
|
||
"kind": "topic",
|
||
"evidence_count": 3,
|
||
},
|
||
{
|
||
"text": "对 5.1 声道家庭影院感兴趣，也在考虑升级到 5.1.2 或 7.1",
|
||
"kind": "topic",
|
||
"evidence_count": 3,
|
||
},
|
||
],
|
||
)
|
||
|
||
self.assertEqual(len(facts), 2)
|
||
self.assertTrue(
|
||
any("音响配置" in fact["text"] for fact in facts)
|
||
)
|
||
self.assertTrue(
|
||
any("5.1.2" in fact["text"] for fact in facts)
|
||
)
|
||
|
||
def test_configured_aliases_share_a_group_scoped_profile(self):
|
||
remembered = self.application.accept(
|
||
event(
|
||
"alias-remember",
|
||
"@JWeBot /记住 我喜欢研究家用服务器",
|
||
sender="bob-ocr",
|
||
)
|
||
)
|
||
viewed = self.application.accept(
|
||
event(
|
||
"alias-view",
|
||
"@JWeBot /人物记忆 b1",
|
||
sender="other",
|
||
)
|
||
)
|
||
prefix = self.application.ledger.resolve_person("group-a", "bob-scan")
|
||
ocr_variants = {
|
||
"alice-ocr": "Alice",
|
||
"bob-ocr": "Bob",
|
||
"carol-ocr": "Carol",
|
||
}
|
||
|
||
self.assertIn("同一身份", remembered["text"])
|
||
self.assertIn("Bob 的群内记忆", viewed["text"])
|
||
self.assertIn("喜欢研究家用服务器", viewed["text"])
|
||
self.assertEqual(
|
||
prefix["person_key"],
|
||
MODULE.person_scope_key("group-a", "Bob"),
|
||
)
|
||
for sender, canonical_name in ocr_variants.items():
|
||
self.assertEqual(
|
||
self.application.ledger.resolve_person("group-a", sender)["person_key"],
|
||
MODULE.person_scope_key("group-a", canonical_name),
|
||
)
|
||
carol = self.application.ledger.resolve_person("group-a", "c1")
|
||
self.application.ledger.ensure_person_identity("group-a", carol)
|
||
carol_profile = self.application.ledger.person_profile(
|
||
"group-a",
|
||
str(carol["person_key"]),
|
||
)
|
||
self.assertIn(
|
||
"Carol is a configured test profile.",
|
||
[fact["text"] for fact in carol_profile["facts"]],
|
||
)
|
||
|
||
def test_alias_seed_is_cached_per_group(self):
|
||
self.application.ledger.resolve_person("group-a", "b1")
|
||
changes = self.application.ledger.connection.total_changes
|
||
|
||
self.application.ledger.resolve_person("group-a", "b1")
|
||
|
||
self.assertEqual(
|
||
self.application.ledger.connection.total_changes,
|
||
changes,
|
||
)
|
||
|
||
def test_configured_identity_is_idempotent(self):
|
||
person = self.application.ledger.resolve_person("group-a", "c1")
|
||
self.application.ledger.ensure_person_identity("group-a", person)
|
||
changes = self.application.ledger.connection.total_changes
|
||
|
||
self.application.ledger.ensure_person_identity("group-a", person)
|
||
|
||
self.assertEqual(
|
||
self.application.ledger.connection.total_changes,
|
||
changes,
|
||
)
|
||
|
||
def test_mentioned_configured_alias_is_resolved_locally(self):
|
||
self.application.deepseek.complete = mock.Mock(side_effect=AssertionError)
|
||
response = self.application.accept(
|
||
event(
|
||
"ask-bob",
|
||
"@JWeBot b1是谁啊",
|
||
sender="Alice",
|
||
)
|
||
)
|
||
|
||
self.assertEqual(response["route"], "local")
|
||
self.assertIn("在这个群里，b1 是Bob", response["text"])
|
||
self.assertIn("Bob is a configured test profile", response["text"])
|
||
self.assertEqual(
|
||
self.application.ledger.row("ask-bob")["status"],
|
||
"alias_resolved",
|
||
)
|
||
self.application.deepseek.complete.assert_not_called()
|
||
|
||
def test_aliases_are_group_scoped_and_identity_is_injected(self):
|
||
captured = {}
|
||
|
||
def complete(**kwargs):
|
||
captured.update(kwargs)
|
||
return "知道了喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(
|
||
event(
|
||
"alice-identity",
|
||
"@JWeBot 请完整说说你是谁",
|
||
sender="a1",
|
||
)
|
||
)
|
||
self.application.process_direct_chat_once()
|
||
group_a = self.application.ledger.resolve_person("group-a", "alice-ocr")
|
||
group_b = self.application.ledger.resolve_person("group-b", "alice-ocr")
|
||
profile = self.application.ledger.person_profile(
|
||
"group-a",
|
||
str(group_a["person_key"]),
|
||
)
|
||
|
||
self.assertNotEqual(group_a["person_key"], group_b["person_key"])
|
||
self.assertEqual(profile["display_name"], "Alice")
|
||
self.assertTrue(
|
||
any(
|
||
line == "[identity] Alice is a configured test profile."
|
||
for line in captured["person_memory"]
|
||
)
|
||
)
|
||
|
||
def test_sender_alias_is_injected_for_a_short_self_identity_question(self):
|
||
captured = {}
|
||
|
||
def complete(**kwargs):
|
||
captured.update(kwargs)
|
||
return "你是 Alice 喵"
|
||
|
||
self.application.deepseek.complete = complete
|
||
self.application.deliver = mock.Mock()
|
||
response = self.application.accept(
|
||
event(
|
||
"who-am-i",
|
||
"@JWeBot 我是谁",
|
||
sender="alice",
|
||
)
|
||
)
|
||
self.application.process_direct_chat_once()
|
||
|
||
self.assertEqual(response["route"], "chat")
|
||
self.assertEqual(response["text"], MODULE.FAST_ACKNOWLEDGEMENT)
|
||
self.assertEqual(captured["sender"], "alice")
|
||
self.assertTrue(
|
||
any(
|
||
line == "[identity] Alice is a configured test profile."
|
||
for line in captured["person_memory"]
|
||
)
|
||
)
|
||
|
||
def test_alias_version_change_skips_rebuild_when_mapping_is_unchanged(self):
|
||
ledger = self.application.ledger
|
||
person = ledger.resolve_person("group-a", "b1")
|
||
ledger.ensure_person_identity("group-a", person)
|
||
with ledger.lock:
|
||
before = ledger.connection.execute(
|
||
"""
|
||
SELECT facts_json FROM person_profiles
|
||
WHERE group_fingerprint = ? AND person_key = ?
|
||
""",
|
||
("group-a", person["person_key"]),
|
||
).fetchone()["facts_json"]
|
||
ledger.set_metadata("person_alias_registry_version", "legacy-version")
|
||
|
||
migrated = ledger.migrate_person_aliases()
|
||
|
||
with ledger.lock:
|
||
after = ledger.connection.execute(
|
||
"""
|
||
SELECT facts_json FROM person_profiles
|
||
WHERE group_fingerprint = ? AND person_key = ?
|
||
""",
|
||
("group-a", person["person_key"]),
|
||
).fetchone()["facts_json"]
|
||
self.assertEqual(
|
||
migrated,
|
||
{
|
||
"observations_rekeyed": 0,
|
||
"profiles_rebuilt": 0,
|
||
"jobs_queued": 0,
|
||
},
|
||
)
|
||
self.assertEqual(before, after)
|
||
self.assertEqual(
|
||
ledger.metadata("person_alias_registry_version"),
|
||
MODULE.PERSON_ALIAS_MIGRATION_VERSION,
|
||
)
|
||
|
||
def test_alias_migration_rekeys_legacy_observations_and_profiles(self):
|
||
ledger = self.application.ledger
|
||
legacy_key = MODULE.person_scope_key("group-a", "b1")
|
||
expires_at = MODULE.time.time() + 60
|
||
fact = {
|
||
"text": "喜欢研究 NAS 存储",
|
||
"kind": "preference",
|
||
"source": "explicit",
|
||
"evidence_count": 1,
|
||
"created_at": MODULE.utc_now(),
|
||
"updated_at": MODULE.utc_now(),
|
||
"expires_at": expires_at,
|
||
}
|
||
with ledger.lock, ledger.connection:
|
||
ledger.connection.execute(
|
||
"""
|
||
INSERT INTO person_observations (
|
||
event_id, group_fingerprint, person_key, display_name, text,
|
||
source, observed_at, created_at
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
"legacy-bob",
|
||
"group-a",
|
||
legacy_key,
|
||
"b1",
|
||
"我喜欢研究 NAS 存储",
|
||
"history",
|
||
MODULE.utc_now(),
|
||
MODULE.utc_now(),
|
||
),
|
||
)
|
||
ledger.connection.execute(
|
||
"""
|
||
INSERT INTO person_profiles (
|
||
group_fingerprint, person_key, display_name, facts_json, updated_at
|
||
) VALUES (?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
"group-a",
|
||
legacy_key,
|
||
"b1",
|
||
json.dumps([fact], ensure_ascii=False),
|
||
MODULE.utc_now(),
|
||
),
|
||
)
|
||
|
||
migrated = ledger.migrate_person_aliases()
|
||
canonical_key = MODULE.person_scope_key("group-a", "Bob")
|
||
profile = ledger.person_profile("group-a", canonical_key)
|
||
|
||
self.assertEqual(migrated["observations_rekeyed"], 1)
|
||
self.assertEqual(profile["display_name"], "Bob")
|
||
self.assertIn("喜欢研究 NAS 存储", [fact["text"] for fact in profile["facts"]])
|
||
self.assertIsNone(ledger.person_profile("group-a", legacy_key))
|
||
self.assertEqual(
|
||
ledger.migrate_person_aliases(),
|
||
{"observations_rekeyed": 0, "profiles_rebuilt": 0, "jobs_queued": 0},
|
||
)
|
||
|
||
def test_alias_migration_preserves_automatic_facts_by_default(self):
|
||
ledger = self.application.ledger
|
||
expires_at = MODULE.time.time() + 60
|
||
automatic_bob = {
|
||
"text": "经常研究家庭服务器",
|
||
"kind": "topic",
|
||
"source": "automatic",
|
||
"evidence_count": 3,
|
||
"created_at": MODULE.utc_now(),
|
||
"updated_at": MODULE.utc_now(),
|
||
"expires_at": expires_at,
|
||
}
|
||
automatic_carol = {
|
||
"text": "最近研究 5.1 音响配置",
|
||
"kind": "topic",
|
||
"source": "automatic",
|
||
"evidence_count": 3,
|
||
"created_at": MODULE.utc_now(),
|
||
"updated_at": MODULE.utc_now(),
|
||
"expires_at": expires_at,
|
||
}
|
||
for display_name, fact in (
|
||
("b1", automatic_bob),
|
||
("c1", automatic_carol),
|
||
):
|
||
with ledger.lock, ledger.connection:
|
||
ledger.connection.execute(
|
||
"""
|
||
INSERT INTO person_profiles (
|
||
group_fingerprint, person_key, display_name, facts_json, updated_at
|
||
) VALUES (?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
"group-a",
|
||
MODULE.person_scope_key("group-a", display_name),
|
||
display_name,
|
||
json.dumps([fact], ensure_ascii=False),
|
||
MODULE.utc_now(),
|
||
),
|
||
)
|
||
|
||
ledger.migrate_person_aliases()
|
||
bob = ledger.person_profile(
|
||
"group-a",
|
||
MODULE.person_scope_key("group-a", "Bob"),
|
||
)
|
||
carol = ledger.person_profile(
|
||
"group-a",
|
||
MODULE.person_scope_key("group-a", "Carol"),
|
||
)
|
||
|
||
self.assertEqual(bob["display_name"], "Bob")
|
||
self.assertEqual(carol["display_name"], "Carol")
|
||
self.assertIn(
|
||
"经常研究家庭服务器",
|
||
[fact["text"] for fact in bob["facts"]],
|
||
)
|
||
self.assertIn(
|
||
"最近研究 5.1 音响配置",
|
||
[fact["text"] for fact in carol["facts"]],
|
||
)
|
||
|
||
def test_cloud_projection_contains_profile_summary_not_observations(self):
|
||
self.application.accept(
|
||
event("remember", "@JWeBot /记住 我更喜欢技术细节")
|
||
)
|
||
projection = MODULE.MemoryProjector(
|
||
self.application.ledger,
|
||
self.config,
|
||
).render()
|
||
|
||
self.assertIn("## Person-Scoped Profiles", projection)
|
||
self.assertIn("display:Alice", projection)
|
||
self.assertIn("我更喜欢技术细节", projection)
|
||
|
||
def test_guess_number_can_start_and_finish(self):
|
||
started = self.application.accept(
|
||
event("guess-start", "@JWeBot /猜数字 开始 10")
|
||
)
|
||
state = self.application.ledger.local_game("group-a", "guess_number")
|
||
self.assertIn("范围 1 到 10", started["text"])
|
||
self.assertIsNotNone(state)
|
||
|
||
finished = self.application.accept(
|
||
event(
|
||
"guess-win",
|
||
f"@JWeBot /猜 {state['target']}",
|
||
)
|
||
)
|
||
|
||
self.assertIn("猜中了", finished["text"])
|
||
self.assertIsNone(
|
||
self.application.ledger.local_game("group-a", "guess_number")
|
||
)
|
||
|
||
def test_rps_resolves_after_all_joined_players_choose(self):
|
||
self.application.accept(
|
||
event("rps-start", "@JWeBot /石头剪刀布 开始", sender="A")
|
||
)
|
||
self.application.accept(
|
||
event("rps-join", "@JWeBot /石头剪刀布 加入", sender="B")
|
||
)
|
||
waiting = self.application.accept(
|
||
event("rps-a", "@JWeBot /石头剪刀布 出 石头", sender="A")
|
||
)
|
||
resolved = self.application.accept(
|
||
event("rps-b", "@JWeBot /石头剪刀布 出 剪刀", sender="B")
|
||
)
|
||
|
||
self.assertIn("还差 1 人", waiting["text"])
|
||
self.assertIn("石头 赢了", resolved["text"])
|
||
state = self.application.ledger.local_game("group-a", "rps")
|
||
self.assertEqual(state["choices"], {})
|
||
|
||
def test_cloud_request_has_idempotent_acknowledgement_and_outbox(self):
|
||
first = self.application.accept(event(text="@JWeBot #status"))
|
||
second = self.application.accept(event(text="@JWeBot #status"))
|
||
|
||
self.assertEqual(first["text"], MODULE.ACKNOWLEDGEMENT)
|
||
self.assertTrue(first["fast_ack"])
|
||
self.assertTrue(second["duplicate"])
|
||
self.assertEqual(self.application.ledger.status()["outbox_pending"], 1)
|
||
|
||
def test_outbox_executes_once_and_marks_safe_delivery(self):
|
||
self.application.cloud.complete = mock.Mock(return_value="云端完成喵")
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(event(text="@JWeBot #status"))
|
||
|
||
processed = self.application.process_outbox_once()
|
||
|
||
self.assertTrue(processed)
|
||
self.application.cloud.complete.assert_called_once()
|
||
self.application.deliver.assert_not_called()
|
||
|
||
self.application.confirm_fast_acknowledgement(
|
||
"event-1",
|
||
delivered=True,
|
||
)
|
||
self.assertTrue(self.application.process_outbox_once())
|
||
self.application.deliver.assert_called_once()
|
||
self.assertEqual(self.application.ledger.status()["outbox_pending"], 0)
|
||
self.assertEqual(
|
||
self.application.ledger.row("event-1")["status"],
|
||
"cloud_delivered",
|
||
)
|
||
|
||
def test_latest_failed_cloud_task_can_be_explicitly_requeued(self):
|
||
self.application.accept(event(text="@JWeBot #status"))
|
||
with self.application.ledger.lock, self.application.ledger.connection:
|
||
self.application.ledger.connection.execute(
|
||
"""
|
||
UPDATE outbox
|
||
SET state = 'failed', attempts = 3, error_code = 'EdgeError'
|
||
WHERE event_id = ?
|
||
""",
|
||
("event-1",),
|
||
)
|
||
self.application.ledger.set_event(
|
||
"event-1",
|
||
status="cloud_failed",
|
||
error_code="EdgeError",
|
||
)
|
||
|
||
event_id = self.application.ledger.retry_latest_failed_cloud()
|
||
claimed = self.application.ledger.claim_outbox()
|
||
|
||
self.assertEqual(event_id, "event-1")
|
||
self.assertIsNotNone(claimed)
|
||
self.assertEqual(claimed["claimed_state"], "cloud_running_acknowledged")
|
||
|
||
def test_memory_projection_is_stable_and_redacts_secrets(self):
|
||
self.application.deepseek.complete = lambda **_: "已记录 api_key=secret-value 喵"
|
||
self.application.deliver = mock.Mock()
|
||
self.application.accept(event(text="@JWeBot api_key=secret-value"))
|
||
self.application.process_direct_chat_once()
|
||
projector = MODULE.MemoryProjector(self.application.ledger, self.config)
|
||
|
||
first = projector.render()
|
||
second = projector.render()
|
||
|
||
self.assertEqual(first, second)
|
||
self.assertIn("[REDACTED SECRET]", first)
|
||
self.assertNotIn("secret-value", first)
|
||
|
||
|
||
class DirectClientTests(unittest.TestCase):
|
||
def test_openai_compatible_request_is_bounded(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
client = MODULE.DeepSeekClient(
|
||
config(
|
||
Path(temporary.name),
|
||
api_key="test-key",
|
||
model="deepseek-pro",
|
||
)
|
||
)
|
||
response = mock.MagicMock()
|
||
response.read1.return_value = (
|
||
'data: {"choices":[{"delta":{"content":"正"}}]}\n\n'
|
||
'data: {"choices":[{"delta":{"content":"常"}}]}\n\n'
|
||
"data: [DONE]\n\n"
|
||
).encode("utf-8")
|
||
response.__enter__.return_value = response
|
||
response.__exit__.return_value = False
|
||
with mock.patch.object(MODULE.urllib.request, "urlopen", return_value=response) as urlopen:
|
||
reply = client.complete(
|
||
event_id="event-1",
|
||
group_fingerprint="group-a",
|
||
sender="Alice",
|
||
question="你好",
|
||
context=[],
|
||
group_references=("群内昵称映射：b1 -> Bob。",),
|
||
)
|
||
|
||
request = urlopen.call_args.args[0]
|
||
payload = json.loads(request.data)
|
||
self.assertEqual(reply, "正常喵")
|
||
self.assertEqual(payload["model"], "deepseek-flash")
|
||
self.assertTrue(payload["stream"])
|
||
self.assertEqual(payload["max_tokens"], 160)
|
||
self.assertEqual(payload["thinking"], {"type": "disabled"})
|
||
self.assertNotIn("temperature", payload)
|
||
self.assertEqual(request.get_header("Authorization"), "Bearer test-key")
|
||
self.assertIn(MODULE.BOT_IDENTITY_NOTE, payload["messages"][0]["content"])
|
||
self.assertIn(
|
||
MODULE.TAFFY_CHAT_STYLE_NOTE,
|
||
payload["messages"][0]["content"],
|
||
)
|
||
self.assertTrue(
|
||
any(
|
||
message["content"].startswith("CURRENT_GROUP_SENDER_LABEL:")
|
||
for message in payload["messages"]
|
||
)
|
||
)
|
||
self.assertIn(
|
||
"isolated test assistant",
|
||
payload["messages"][0]["content"],
|
||
)
|
||
self.assertTrue(
|
||
any(
|
||
message["content"].startswith("CONFIGURED_GROUP_ALIAS_REFERENCES:")
|
||
for message in payload["messages"]
|
||
)
|
||
)
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_stream_reader_stops_at_done_without_waiting_for_eof(self):
|
||
response = mock.MagicMock()
|
||
response.read1.side_effect = [
|
||
b'data: {"choices":[{"delta":{"content":"ok"}}]}\n'
|
||
b"data: [DO",
|
||
b"NE]\n\n",
|
||
AssertionError("reader continued after the stream completed"),
|
||
]
|
||
|
||
raw = MODULE.read_direct_api_response(response)
|
||
|
||
self.assertEqual(
|
||
raw,
|
||
b'data: {"choices":[{"delta":{"content":"ok"}}]}\n'
|
||
b"data: [DONE]\n\n",
|
||
)
|
||
self.assertEqual(response.read1.call_count, 2)
|
||
|
||
def test_stream_reader_does_not_treat_content_as_done_marker(self):
|
||
response = mock.MagicMock()
|
||
response.read1.side_effect = [
|
||
b'data: {"choices":[{"delta":{"content":"data: [DONE]"}}]}\n\n',
|
||
b"data: [DONE]\n\n",
|
||
AssertionError("reader continued after the stream completed"),
|
||
]
|
||
|
||
raw = MODULE.read_direct_api_response(response)
|
||
|
||
self.assertIn(b'"content":"data: [DONE]"', raw)
|
||
self.assertTrue(raw.endswith(b"data: [DONE]\n\n"))
|
||
self.assertEqual(response.read1.call_count, 2)
|
||
|
||
def test_stream_reader_enforces_total_wall_time(self):
|
||
response = mock.MagicMock()
|
||
response.read1.return_value = b": keep-alive\n\n"
|
||
with mock.patch.object(
|
||
MODULE.time,
|
||
"monotonic",
|
||
side_effect=[0.0, 0.0, 31.0],
|
||
):
|
||
with self.assertRaisesRegex(
|
||
MODULE.DeepSeekUnavailable,
|
||
"exceeded total timeout",
|
||
):
|
||
MODULE.read_direct_api_response(response)
|
||
|
||
response.fp.raw._sock.settimeout.assert_called_once_with(18.0)
|
||
response.read1.assert_called_once()
|
||
|
||
def test_stream_parser_accepts_a_non_streaming_compatibility_payload(self):
|
||
self.assertEqual(
|
||
MODULE.deepseek_response_content(
|
||
b'{"choices":[{"message":{"content":"fallback"}}]}'
|
||
),
|
||
"fallback",
|
||
)
|
||
|
||
def test_stream_parser_rejects_reasoning_without_a_visible_reply(self):
|
||
with self.assertRaises(MODULE.DeepSeekUnavailable):
|
||
MODULE.deepseek_response_content(
|
||
b'data: {"choices":[{"delta":{"reasoning_content":"hidden"}}]}\n\n'
|
||
b"data: [DONE]\n\n"
|
||
)
|
||
|
||
|
||
class SteamClientTests(unittest.TestCase):
|
||
OWNER = "10000000000000001"
|
||
BOB = "10000000000000002"
|
||
CAROL = "10000000000000003"
|
||
|
||
@staticmethod
|
||
def response(payload):
|
||
result = mock.MagicMock()
|
||
result.read.return_value = json.dumps(payload).encode("utf-8")
|
||
result.__enter__.return_value = result
|
||
result.__exit__.return_value = False
|
||
return result
|
||
|
||
def client(
|
||
self,
|
||
directory: Path,
|
||
*,
|
||
steam_roster: tuple[dict[str, object], ...] = (),
|
||
):
|
||
return MODULE.SteamClient(
|
||
config(
|
||
directory,
|
||
steam_api_key="A" * 32,
|
||
steam_id64=self.OWNER,
|
||
steam_roster=steam_roster,
|
||
)
|
||
)
|
||
|
||
def test_presence_discovers_public_friend_list_and_caches_requests(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
client = self.client(Path(temporary.name))
|
||
|
||
def request(path, parameters):
|
||
if path.endswith("GetFriendList/v1/"):
|
||
return {
|
||
"friendslist": {
|
||
"friends": [
|
||
{"steamid": self.BOB},
|
||
{"steamid": self.CAROL},
|
||
]
|
||
}
|
||
}
|
||
self.assertTrue(path.endswith("GetPlayerSummaries/v2/"))
|
||
return {
|
||
"response": {
|
||
"players": [
|
||
{
|
||
"steamid": self.OWNER,
|
||
"personaname": "Alice",
|
||
"personastate": 1,
|
||
"gameextrainfo": "Dota 2",
|
||
},
|
||
{
|
||
"steamid": self.BOB,
|
||
"personaname": "Bob",
|
||
"personastate": 2,
|
||
},
|
||
{
|
||
"steamid": self.CAROL,
|
||
"personaname": "Carol",
|
||
"personastate": 0,
|
||
},
|
||
]
|
||
}
|
||
}
|
||
|
||
with mock.patch.object(
|
||
client,
|
||
"request_json",
|
||
side_effect=request,
|
||
) as request_json:
|
||
first = client.render("")
|
||
second = client.render("")
|
||
selected = client.render("Bob")
|
||
|
||
self.assertEqual(request_json.call_count, 2)
|
||
self.assertIn("Steam：在线 2/3", first)
|
||
self.assertIn("Alice 正在玩 Dota 2", first)
|
||
self.assertIn("Bob 忙碌", first)
|
||
self.assertEqual(first, second)
|
||
self.assertEqual(selected, "Steam：Bob 忙碌")
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_private_roster_limits_accounts_and_uses_group_aliases(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
roster = (
|
||
{
|
||
"steam_id": self.OWNER,
|
||
"label": "alice",
|
||
"selectors": ("Alice", "alice-steam"),
|
||
},
|
||
{
|
||
"steam_id": self.BOB,
|
||
"label": "bob",
|
||
"selectors": ("Bob", "bob-steam"),
|
||
},
|
||
)
|
||
client = self.client(
|
||
Path(temporary.name),
|
||
steam_roster=roster,
|
||
)
|
||
summaries = {
|
||
self.OWNER: {
|
||
"steamid": self.OWNER,
|
||
"personaname": "Untrusted Persona One",
|
||
"personastate": 1,
|
||
},
|
||
self.BOB: {
|
||
"steamid": self.BOB,
|
||
"personaname": "Untrusted Persona Two",
|
||
"personastate": 2,
|
||
},
|
||
}
|
||
with mock.patch.object(
|
||
client,
|
||
"fetch_summaries",
|
||
return_value=summaries,
|
||
) as fetch_summaries, mock.patch.object(
|
||
client,
|
||
"request_json",
|
||
) as request_json:
|
||
listing = client.render("")
|
||
selected = client.render("bob-steam")
|
||
|
||
fetch_summaries.assert_called_once_with([self.OWNER, self.BOB])
|
||
request_json.assert_not_called()
|
||
self.assertEqual(client.friend_ids(), [self.OWNER, self.BOB])
|
||
self.assertIn("Steam：在线 2/2", listing)
|
||
self.assertIn("alice 在线", listing)
|
||
self.assertIn("bob 忙碌", listing)
|
||
self.assertNotIn("Untrusted Persona", listing)
|
||
self.assertEqual(selected, "Steam：bob 忙碌")
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_private_roster_playtime_queries_only_scoped_accounts(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
roster = (
|
||
{
|
||
"steam_id": self.OWNER,
|
||
"label": "alice",
|
||
"selectors": ("Alice",),
|
||
},
|
||
{
|
||
"steam_id": self.BOB,
|
||
"label": "bob",
|
||
"selectors": ("Bob",),
|
||
},
|
||
)
|
||
client = self.client(
|
||
Path(temporary.name),
|
||
steam_roster=roster,
|
||
)
|
||
records = [
|
||
{
|
||
"steam_id": self.OWNER,
|
||
"label": "alice",
|
||
"selectors": ("Alice",),
|
||
"scoped": True,
|
||
"state": 1,
|
||
"game": "",
|
||
},
|
||
{
|
||
"steam_id": self.BOB,
|
||
"label": "bob",
|
||
"selectors": ("Bob",),
|
||
"scoped": True,
|
||
"state": 1,
|
||
"game": "",
|
||
},
|
||
]
|
||
|
||
def games(steam_id, *, app_id=None):
|
||
values = {
|
||
self.OWNER: 600,
|
||
self.BOB: 120,
|
||
}
|
||
return [
|
||
{
|
||
"appid": 570,
|
||
"name": "Dota 2",
|
||
"playtime_forever": values[steam_id],
|
||
}
|
||
]
|
||
|
||
with mock.patch.object(
|
||
client,
|
||
"presence",
|
||
return_value=records,
|
||
), mock.patch.object(
|
||
client,
|
||
"fetch_owned_games",
|
||
side_effect=games,
|
||
) as fetch:
|
||
result = client.render("时长Dota2")
|
||
|
||
self.assertIn("已查询 2 个账号", result)
|
||
self.assertEqual(
|
||
{call.args[0] for call in fetch.call_args_list},
|
||
{self.OWNER, self.BOB},
|
||
)
|
||
self.assertNotIn(self.CAROL, {
|
||
call.args[0] for call in fetch.call_args_list
|
||
})
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_request_uses_mihomo_and_never_exposes_api_key(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
api_key = "A" * 32
|
||
client = MODULE.SteamClient(
|
||
config(
|
||
Path(temporary.name),
|
||
steam_api_key=api_key,
|
||
steam_id64=self.OWNER,
|
||
)
|
||
)
|
||
response = self.response({"response": {}})
|
||
opener = mock.Mock()
|
||
opener.open.return_value = response
|
||
with mock.patch.object(
|
||
MODULE.urllib.request,
|
||
"build_opener",
|
||
return_value=opener,
|
||
) as build_opener:
|
||
payload = client.request_json("/test/v1/", {"value": "safe"})
|
||
|
||
proxy_handler = build_opener.call_args.args[0]
|
||
request = opener.open.call_args.args[0]
|
||
query = MODULE.urllib.parse.parse_qs(
|
||
MODULE.urllib.parse.urlsplit(request.full_url).query
|
||
)
|
||
|
||
self.assertEqual(
|
||
proxy_handler.proxies,
|
||
{"https": MODULE.DEFAULT_STEAM_MIHOMO_PROXY_URL},
|
||
)
|
||
self.assertEqual(query["key"], [api_key])
|
||
self.assertEqual(query["value"], ["safe"])
|
||
self.assertEqual(payload, {"response": {}})
|
||
self.assertNotIn(api_key, str(payload))
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_compact_playtime_command_fuzzy_matches_and_queries_every_account(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
client = self.client(Path(temporary.name))
|
||
records = [
|
||
{"steam_id": self.OWNER, "label": "Alice", "state": 1, "game": ""},
|
||
{"steam_id": self.BOB, "label": "Bob", "state": 1, "game": ""},
|
||
{"steam_id": self.CAROL, "label": "Carol", "state": 0, "game": ""},
|
||
]
|
||
|
||
def games(steam_id, *, app_id=None):
|
||
if steam_id == self.OWNER:
|
||
return [
|
||
{
|
||
"appid": 570,
|
||
"name": "Dota 2",
|
||
"playtime_forever": 600,
|
||
}
|
||
]
|
||
if steam_id == self.BOB:
|
||
return [{"appid": 570, "playtime_forever": 120}]
|
||
return []
|
||
|
||
with mock.patch.object(
|
||
client,
|
||
"presence",
|
||
return_value=records,
|
||
), mock.patch.object(
|
||
client,
|
||
"fetch_owned_games",
|
||
side_effect=games,
|
||
) as fetch:
|
||
result = client.render("时长Dta2")
|
||
|
||
self.assertIn("Steam [■■■] Dota 2", result)
|
||
self.assertIn("已查询 3 个账号", result)
|
||
self.assertIn("Alice 10h", result)
|
||
self.assertIn("Bob 2h", result)
|
||
self.assertEqual(
|
||
{call.args[0] for call in fetch.call_args_list},
|
||
{self.OWNER, self.BOB, self.CAROL},
|
||
)
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_compact_arguments_do_not_require_ocr_spaces(self):
|
||
self.assertEqual(
|
||
MODULE.SteamClient.parse_arguments("时长Counter Strike"),
|
||
("playtime", "Counter Strike", False),
|
||
)
|
||
self.assertEqual(
|
||
MODULE.SteamClient.parse_arguments("游戏"),
|
||
("game", "", False),
|
||
)
|
||
|
||
def test_playtime_person_selector_supports_explicit_ocr_safe_forms(self):
|
||
values = {
|
||
"Dota 2、Bob": ("Dota 2", "Bob"),
|
||
"Dota 2，Bob": ("Dota 2", "Bob"),
|
||
"Dota 2@Bob": ("Dota 2", "Bob"),
|
||
"Dota 2人=Bob": ("Dota 2", "Bob"),
|
||
"Counter Strike": ("Counter Strike", ""),
|
||
}
|
||
for query, expected in values.items():
|
||
with self.subTest(query=query):
|
||
self.assertEqual(
|
||
MODULE.SteamClient.parse_playtime_query(query),
|
||
expected,
|
||
)
|
||
|
||
def test_person_playtime_fuzzy_matches_and_queries_only_that_account(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
client = self.client(Path(temporary.name))
|
||
records = [
|
||
{"steam_id": self.OWNER, "label": "Alice", "state": 1, "game": ""},
|
||
{"steam_id": self.BOB, "label": "Bobby", "state": 1, "game": ""},
|
||
{"steam_id": self.CAROL, "label": "Carol", "state": 0, "game": ""},
|
||
]
|
||
|
||
def games(steam_id, *, app_id=None):
|
||
self.assertEqual(steam_id, self.BOB)
|
||
return [
|
||
{
|
||
"appid": 570,
|
||
"name": "Dota 2",
|
||
"playtime_forever": 135,
|
||
}
|
||
]
|
||
|
||
with mock.patch.object(
|
||
client,
|
||
"presence",
|
||
return_value=records,
|
||
), mock.patch.object(
|
||
client,
|
||
"fetch_owned_games",
|
||
side_effect=games,
|
||
) as fetch:
|
||
result = client.render("时长Dta2、Boby")
|
||
|
||
self.assertIn("Steam [■■■] Dota 2", result)
|
||
self.assertIn("已查询 1 个账号", result)
|
||
self.assertIn("Bobby 2.2h", result)
|
||
self.assertNotIn("Alice", result)
|
||
self.assertNotIn("Carol", result)
|
||
self.assertEqual(
|
||
[call.args[0] for call in fetch.call_args_list],
|
||
[self.BOB],
|
||
)
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_person_playtime_rejects_ambiguous_or_unknown_selector(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
client = self.client(Path(temporary.name))
|
||
records = [
|
||
{"steam_id": self.OWNER, "label": "Sam One", "state": 1, "game": ""},
|
||
{"steam_id": self.BOB, "label": "Sam Two", "state": 1, "game": ""},
|
||
]
|
||
with mock.patch.object(
|
||
client,
|
||
"presence",
|
||
return_value=records,
|
||
), mock.patch.object(client, "fetch_owned_games") as fetch:
|
||
ambiguous = client.render("时长Dota2、Sam")
|
||
unknown = client.render("时长Dota2、Nobody")
|
||
|
||
self.assertIn("匹配不唯一", ambiguous)
|
||
self.assertIn("没有匹配到玩家", unknown)
|
||
fetch.assert_not_called()
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_person_playtime_names_target_without_public_hours(self):
|
||
records = [
|
||
{"steam_id": self.BOB, "label": "Bobby", "state": 0, "game": ""},
|
||
]
|
||
for games, expected in (
|
||
(None, "Bobby 资料不可见"),
|
||
([], "Bobby 暂无公开游玩时长"),
|
||
):
|
||
with self.subTest(expected=expected):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
client = self.client(Path(temporary.name))
|
||
with mock.patch.object(
|
||
client,
|
||
"presence",
|
||
return_value=records,
|
||
), mock.patch.object(
|
||
client,
|
||
"fetch_owned_games",
|
||
return_value=games,
|
||
) as fetch:
|
||
result = client.render("时长Dota2、Bobby")
|
||
|
||
self.assertIn(expected, result)
|
||
self.assertEqual(
|
||
[call.args[0] for call in fetch.call_args_list],
|
||
[self.BOB],
|
||
)
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_invalid_or_unreachable_proxy_never_falls_back_to_direct_steam(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
client = MODULE.SteamClient(
|
||
config(
|
||
Path(temporary.name),
|
||
steam_api_key="A" * 32,
|
||
steam_id64=self.OWNER,
|
||
)
|
||
)
|
||
opener = mock.Mock()
|
||
opener.open.side_effect = MODULE.urllib.error.URLError("mihomo down")
|
||
with mock.patch.object(
|
||
MODULE.urllib.request,
|
||
"build_opener",
|
||
return_value=opener,
|
||
), mock.patch.object(MODULE.urllib.request, "urlopen") as direct:
|
||
with self.assertRaisesRegex(
|
||
MODULE.SteamUnavailable,
|
||
"Mihomo",
|
||
):
|
||
client.render_group_presence("")
|
||
|
||
direct.assert_not_called()
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
|
||
class CloudRunnerTests(unittest.TestCase):
|
||
def test_cloud_runner_ignores_system_ssh_config(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
try:
|
||
runner = MODULE.CloudRunner(config(Path(temporary.name)))
|
||
key = Path(temporary.name) / "openclaw_tunnel"
|
||
key.write_text("test", encoding="utf-8")
|
||
result = mock.Mock(
|
||
returncode=0,
|
||
stdout=json.dumps(
|
||
{
|
||
"status": "ok",
|
||
"result": {
|
||
"terminalReply": {
|
||
"text": '{"route":"reply","text":"云端正常"}'
|
||
}
|
||
},
|
||
}
|
||
),
|
||
stderr="",
|
||
)
|
||
with mock.patch.object(MODULE, "SSH_KEY", key), mock.patch.object(
|
||
MODULE.subprocess,
|
||
"run",
|
||
return_value=result,
|
||
) as run:
|
||
reply = runner.complete(
|
||
{
|
||
"event_id": "event-1",
|
||
"group_fingerprint": "group-a",
|
||
"sender": "Alice",
|
||
"question": "/status",
|
||
}
|
||
)
|
||
|
||
command = run.call_args.args[0]
|
||
self.assertEqual(command[:3], ["/usr/bin/ssh", "-F", "/dev/null"])
|
||
self.assertNotIn("--model", command)
|
||
self.assertNotIn("--thinking", command)
|
||
self.assertEqual(reply, "云端正常喵")
|
||
finally:
|
||
temporary.cleanup()
|
||
|
||
def test_cloud_protocol_rejects_an_unsupported_route(self):
|
||
with self.assertRaises(MODULE.EdgeError):
|
||
MODULE.cloud_reply_text('{"route":"main","task":"do something"}')
|
||
|
||
def test_cloud_protocol_unwraps_a_nested_reply_contract(self):
|
||
nested = json.dumps(
|
||
{
|
||
"route": "reply",
|
||
"text": json.dumps(
|
||
{"route": "reply", "text": "嵌套内容"},
|
||
ensure_ascii=False,
|
||
),
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
self.assertEqual(MODULE.cloud_reply_text(nested), "嵌套内容喵")
|
||
|
||
def test_cloud_protocol_applies_chat_reply_style(self):
|
||
self.assertEqual(
|
||
MODULE.cloud_reply_text(
|
||
json.dumps(
|
||
{
|
||
"route": "reply",
|
||
"text": (
|
||
"李神是啥我都不清楚，怎么比，"
|
||
"总不能凭空给自己打满分吧"
|
||
),
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
),
|
||
"李神是啥喵",
|
||
)
|
||
|
||
def test_cloud_protocol_ignores_a_suffix_after_the_json_object(self):
|
||
self.assertEqual(
|
||
MODULE.cloud_reply_text(
|
||
'{"route":"reply","text":"尾随后缀"}喵'
|
||
),
|
||
"尾随后缀喵",
|
||
)
|
||
|
||
|
||
class UnixSocketTests(unittest.TestCase):
|
||
def test_existing_direct_outbox_schema_gains_phase_columns(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
ledger = None
|
||
try:
|
||
database_path = Path(temporary.name) / "edge.sqlite3"
|
||
connection = sqlite3.connect(database_path)
|
||
connection.execute(
|
||
"""
|
||
CREATE TABLE direct_chat_outbox (
|
||
event_id TEXT PRIMARY KEY,
|
||
state TEXT NOT NULL,
|
||
attempts INTEGER NOT NULL DEFAULT 0,
|
||
next_attempt_at REAL NOT NULL DEFAULT 0,
|
||
result_text TEXT NOT NULL DEFAULT '',
|
||
error_code TEXT NOT NULL DEFAULT '',
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
"""
|
||
)
|
||
connection.close()
|
||
|
||
ledger = MODULE.Ledger(database_path)
|
||
columns = {
|
||
row["name"]
|
||
for row in ledger.connection.execute(
|
||
"PRAGMA table_info(direct_chat_outbox)"
|
||
).fetchall()
|
||
}
|
||
|
||
self.assertTrue(
|
||
{"kind", "generation_attempts", "delivery_attempts"}
|
||
<= columns
|
||
)
|
||
finally:
|
||
if ledger is not None:
|
||
ledger.close()
|
||
temporary.cleanup()
|
||
|
||
def test_socket_health_round_trip_does_not_create_an_event(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
application = None
|
||
server = None
|
||
try:
|
||
configured = config(Path(temporary.name))
|
||
application = MODULE.EdgeApplication(configured)
|
||
server = MODULE.EdgeServer(configured.socket_path, application)
|
||
os.chmod(configured.socket_path, 0o600)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
|
||
response = MODULE.request_edge({"op": "health"}, socket_path=configured.socket_path)
|
||
|
||
self.assertEqual(response["status"], "ok")
|
||
self.assertEqual(response["events"], 0)
|
||
finally:
|
||
if server is not None:
|
||
server.shutdown()
|
||
server.server_close()
|
||
if application is not None:
|
||
application.close()
|
||
temporary.cleanup()
|
||
|
||
def test_socket_fast_ack_releases_an_already_generated_final_reply(self):
|
||
temporary = tempfile.TemporaryDirectory()
|
||
application = None
|
||
server = None
|
||
try:
|
||
configured = config(Path(temporary.name))
|
||
application = MODULE.EdgeApplication(configured)
|
||
application.deepseek.complete = mock.Mock(return_value="最终回复喵")
|
||
application.deliver = mock.Mock()
|
||
server = MODULE.EdgeServer(configured.socket_path, application)
|
||
os.chmod(configured.socket_path, 0o600)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
|
||
accepted = MODULE.request_edge(event(), socket_path=configured.socket_path)
|
||
self.assertTrue(accepted["fast_ack"])
|
||
self.assertTrue(application.process_direct_chat_once())
|
||
application.deliver.assert_not_called()
|
||
|
||
confirmed = MODULE.request_edge(
|
||
{"op": "fast_ack_sent", "event_id": "event-1"},
|
||
socket_path=configured.socket_path,
|
||
)
|
||
self.assertEqual(confirmed["status"], "ok")
|
||
self.assertTrue(application.process_direct_chat_once())
|
||
application.deliver.assert_called_once()
|
||
finally:
|
||
if server is not None:
|
||
server.shutdown()
|
||
server.server_close()
|
||
if application is not None:
|
||
application.close()
|
||
temporary.cleanup()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|