194 lines
7.0 KiB
Python
194 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
||
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
from tests.test_jwebot_edge import MODULE, config, event
|
||
|
||
|
||
class LocalCommandParityTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.enabled_temp = tempfile.TemporaryDirectory()
|
||
self.compat_temp = tempfile.TemporaryDirectory()
|
||
self.enabled = MODULE.EdgeApplication(
|
||
config(
|
||
Path(self.enabled_temp.name),
|
||
local_extensions_enabled=True,
|
||
)
|
||
)
|
||
self.compat = MODULE.EdgeApplication(
|
||
config(
|
||
Path(self.compat_temp.name),
|
||
local_extensions_enabled=False,
|
||
)
|
||
)
|
||
|
||
def tearDown(self):
|
||
self.enabled.close()
|
||
self.compat.close()
|
||
self.enabled_temp.cleanup()
|
||
self.compat_temp.cleanup()
|
||
|
||
def assert_command_equal(
|
||
self,
|
||
event_id: str,
|
||
text: str,
|
||
*,
|
||
sender: str = "Alice",
|
||
):
|
||
incoming = event(
|
||
event_id,
|
||
text,
|
||
sender=sender,
|
||
)
|
||
enabled = self.enabled.accept(dict(incoming))
|
||
compat = self.compat.accept(dict(incoming))
|
||
self.assertEqual(enabled["route"], compat["route"])
|
||
self.assertEqual(enabled["text"], compat["text"])
|
||
return enabled, compat
|
||
|
||
def test_stateless_command_families_match(self):
|
||
commands = (
|
||
("help", "@JWeBot /功能列表"),
|
||
("dice", "@JWeBot /丢骰子5d20"),
|
||
("coin", "@JWeBot /翻硬币"),
|
||
("random", "@JWeBot /随机数 -10 100"),
|
||
("lot", "@JWeBot /抽签"),
|
||
("fortune", "@JWeBot /今日运势"),
|
||
("memory-help", "@JWeBot /记忆帮助"),
|
||
("scan-empty", "@JWeBot /扫码排行"),
|
||
("repeat-empty", "@JWeBot /复读榜"),
|
||
("history-empty", "@JWeBot /查记录 测试"),
|
||
)
|
||
for event_id, text in commands:
|
||
with self.subTest(command=text):
|
||
self.assert_command_equal(event_id, text)
|
||
|
||
def test_memory_subcommands_match(self):
|
||
commands = (
|
||
("remember", "@JWeBot /记住 我喜欢家庭服务器"),
|
||
("my-memory", "@JWeBot /我的记忆"),
|
||
("person-memory", "@JWeBot /人物记忆 Alice"),
|
||
("forget-keyword", "@JWeBot /忘记 家庭服务器"),
|
||
("remember-again", "@JWeBot /记住 我喜欢技术细节"),
|
||
("forget-all", "@JWeBot /忘记 全部"),
|
||
)
|
||
for event_id, text in commands:
|
||
with self.subTest(command=text):
|
||
self.assert_command_equal(event_id, text)
|
||
|
||
def test_dice_challenge_subcommands_match(self):
|
||
commands = (
|
||
("challenge-start", "@JWeBot /丢骰子挑战 开始 5d20", "Alice"),
|
||
("challenge-alice", "@JWeBot /丢骰子挑战", "Alice"),
|
||
("challenge-gb", "@JWeBot /丢骰子挑战", "Bob"),
|
||
("challenge-state", "@JWeBot /丢骰子挑战 状态", "Alice"),
|
||
("challenge-settle", "@JWeBot /丢骰子挑战 结算", "Alice"),
|
||
("challenge-record", "@JWeBot /丢骰子挑战 战绩", "Alice"),
|
||
)
|
||
for event_id, text, sender in commands:
|
||
with self.subTest(command=text, sender=sender):
|
||
self.assert_command_equal(event_id, text, sender=sender)
|
||
|
||
def test_guess_number_subcommands_match(self):
|
||
self.assert_command_equal(
|
||
"guess-start",
|
||
"@JWeBot /猜数字 开始 50",
|
||
)
|
||
enabled_state = self.enabled.ledger.local_game(
|
||
"group-a",
|
||
"guess_number",
|
||
)
|
||
compat_state = self.compat.ledger.local_game(
|
||
"group-a",
|
||
"guess_number",
|
||
)
|
||
self.assertEqual(enabled_state["target"], compat_state["target"])
|
||
target = enabled_state["target"]
|
||
self.assert_command_equal(
|
||
"guess-status",
|
||
"@JWeBot /猜数字 状态",
|
||
)
|
||
self.assert_command_equal(
|
||
"guess-win",
|
||
f"@JWeBot /猜 {target}",
|
||
)
|
||
|
||
def test_rps_subcommands_match(self):
|
||
commands = (
|
||
("rps-start", "@JWeBot /石头剪刀布 开始", "Alice"),
|
||
("rps-join", "@JWeBot /石头剪刀布 加入", "Bob"),
|
||
("rps-first", "@JWeBot /石头剪刀布 出 石头", "Alice"),
|
||
("rps-status", "@JWeBot /石头剪刀布 状态", "Alice"),
|
||
("rps-second", "@JWeBot /石头剪刀布 出 剪刀", "Bob"),
|
||
("rps-end", "@JWeBot /石头剪刀布 结束", "Alice"),
|
||
)
|
||
for event_id, text, sender in commands:
|
||
with self.subTest(command=text, sender=sender):
|
||
self.assert_command_equal(event_id, text, sender=sender)
|
||
|
||
def test_status_balance_and_steam_match(self):
|
||
balance = [
|
||
{
|
||
"currency": "CNY",
|
||
"total_balance": "10.00",
|
||
"granted_balance": "2.00",
|
||
"topped_up_balance": "8.00",
|
||
}
|
||
]
|
||
for application in (self.enabled, self.compat):
|
||
application.local.steam = mock.Mock()
|
||
application.local.steam.render.return_value = (
|
||
"Steam：yr 正在玩 Dota 2"
|
||
)
|
||
with mock.patch.object(
|
||
MODULE.LocalCommandEngine,
|
||
"service_state",
|
||
return_value="active",
|
||
), mock.patch.object(
|
||
MODULE.DeepSeekClient,
|
||
"balance",
|
||
return_value=balance,
|
||
):
|
||
self.assert_command_equal("status", "@JWeBot /状态")
|
||
self.assert_command_equal("server-status", "@JWeBot /status")
|
||
self.assert_command_equal("jwebot-version", "@JWeBot /version")
|
||
self.assert_command_equal("debug-telemetry", "@JWeBot /debug")
|
||
self.assert_command_equal("balance", "@JWeBot /余额")
|
||
self.assert_command_equal("steam", "@JWeBot /steam")
|
||
enabled_row = self.enabled.ledger.row("steam")
|
||
compat_row = self.compat.ledger.row("steam")
|
||
self.assertEqual(
|
||
self.enabled.generate_steam_reply(enabled_row),
|
||
self.compat.generate_steam_reply(compat_row),
|
||
)
|
||
|
||
def test_extension_game_state_survives_application_restart(self):
|
||
response = self.enabled.accept(
|
||
event("restart-game", "@JWeBot /猜数字 开始 30")
|
||
)
|
||
self.assertIn("猜数字开局", response["text"])
|
||
expected = self.enabled.ledger.local_game(
|
||
"group-a",
|
||
"guess_number",
|
||
)
|
||
self.enabled.close()
|
||
self.enabled = MODULE.EdgeApplication(
|
||
config(
|
||
Path(self.enabled_temp.name),
|
||
local_extensions_enabled=True,
|
||
)
|
||
)
|
||
restored = self.enabled.ledger.local_game(
|
||
"group-a",
|
||
"guess_number",
|
||
)
|
||
self.assertEqual(restored["target"], expected["target"])
|
||
self.assertEqual(restored["upper"], 30)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|