209 lines
11 KiB
Python
209 lines
11 KiB
Python
"""Behavioral coverage for local context, identity and single-player funds."""
|
|
|
|
import json
|
|
import random
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from test_jwebot_edge import MODULE, config
|
|
from jwebot_official_extensions.local_commands import local_command_parts
|
|
from jwebot_official_extensions.player_games import hand_value
|
|
|
|
|
|
class Release061Tests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
self.engine = MODULE.EdgeApplication(config(self.root))
|
|
self.local = self.engine.local
|
|
self.counter = 0
|
|
|
|
def tearDown(self):
|
|
self.engine.ledger.connection.close()
|
|
self.temp.cleanup()
|
|
|
|
def command(self, question, *, sender="Alice", group="group-a", event_id=None):
|
|
self.counter += 1
|
|
row = {"question": question, "sender": sender, "group_fingerprint": group,
|
|
"event_id": event_id or f"test-{self.counter}"}
|
|
return self.local.handle(row, {})
|
|
|
|
def state(self, sender="Alice", group="group-a"):
|
|
key = self.engine.ledger.resolve_person(group, sender)["person_key"]
|
|
row = self.engine.ledger.connection.execute(
|
|
"SELECT state_json FROM player_accounts WHERE group_fingerprint=? AND person_key=?",
|
|
(group, key)).fetchone()
|
|
return json.loads(row[0])
|
|
|
|
def test_aliases_are_corresponding_and_case_insensitive(self):
|
|
pairs = [("上下文", "Context"), ("摩拉", "wallet"), ("抽奖", "draw"),
|
|
("买彩票", "lottery"), ("21点", "blackjack"),
|
|
("人物列表", "people"), ("人物", "person"), ("游戏帮助", "games")]
|
|
for chinese, english in pairs:
|
|
self.assertEqual(local_command_parts("/" + chinese), local_command_parts("/" + english))
|
|
self.assertEqual(local_command_parts("/Context200"), ("context_summary", "200"))
|
|
|
|
def test_context_scope_count_order_and_current_request_exclusion(self):
|
|
ledger = self.engine.ledger
|
|
for i in range(305):
|
|
ledger.record_repeat_history(event_id=f"history-{i:04d}", group_fingerprint="group-a",
|
|
sender="Alice", text=f"项目计划讨论编号 {i}",
|
|
observed_at=f"2026-10-15T00:{i // 60:02d}:{i % 60:02d}+00:00")
|
|
ledger.record_repeat_history(event_id="private-other", group_fingerprint="group-b",
|
|
sender="Bob", text="隔离群秘密话题", observed_at="2026-10-15T01:00:00+00:00")
|
|
for count in (100, 200, 300):
|
|
rows = ledger.recent_context_messages("group-a", count, exclude_event_id="history-0304")
|
|
self.assertEqual(len(rows), count)
|
|
self.assertEqual(rows[-1]["event_id"], "history-0303")
|
|
self.assertLess(rows[0]["observed_at"], rows[-1]["observed_at"])
|
|
with mock.patch.object(ledger, "bootstrap_repeat_history"):
|
|
result = self.command(f"/context {count}")
|
|
self.assertIn(f"实际{count}条", result)
|
|
self.assertNotIn("隔离群", result)
|
|
self.assertIn("用法", self.command("/context 101"))
|
|
|
|
def test_person_lookup_uses_alias_but_does_not_expose_steam_ids(self):
|
|
self.assertIn("alice", self.command("/people"))
|
|
self.assertIn("Alice is a configured test profile", self.command("/person a1"))
|
|
self.assertIn("群内索引", self.command("/个人属性"))
|
|
self.assertIn("没有", self.command("/person unconfigured"))
|
|
self.assertIn("只有", self.command("/people 99"))
|
|
|
|
def test_context_keeps_short_messages_omitted_from_repeat_statistics(self):
|
|
ledger = self.engine.ledger
|
|
self.assertFalse(ledger.record_repeat_history(
|
|
event_id="short", group_fingerprint="group-a", sender="Alice",
|
|
text="好", observed_at="2026-10-15T00:00:00+00:00"))
|
|
records = ledger.recent_context_messages("group-a", 100)
|
|
self.assertEqual([row["text"] for row in records], ["好"])
|
|
self.assertEqual(ledger.search_repeat_history("group-a", "好"), [])
|
|
|
|
def test_initial_funds_aliases_groups_and_replay(self):
|
|
self.assertIn("摩拉 1000", self.command("/wallet", event_id="wallet-1"))
|
|
with mock.patch.object(self.local.players.rng, "randrange", return_value=99):
|
|
first = self.command("/draw 10", event_id="draw-1")
|
|
replay = self.command("/draw 10", sender="a1", event_id="draw-1")
|
|
self.assertEqual(first, replay)
|
|
self.assertEqual(self.state("a1")["balance"], 990)
|
|
self.assertIn("摩拉 1000", self.command("/wallet", sender="Bob"))
|
|
self.assertIn("摩拉 1000", self.command("/wallet", group="group-b"))
|
|
self.engine.ledger.connection.close()
|
|
self.engine = MODULE.EdgeApplication(config(self.root))
|
|
self.local = self.engine.local
|
|
self.assertIn("摩拉 990", self.command("/wallet", event_id="restart"))
|
|
|
|
def test_concurrent_replay_debits_once_and_rolls_back_failure(self):
|
|
replies = []
|
|
with mock.patch.object(self.local.players.rng, "randrange", return_value=99):
|
|
threads = [threading.Thread(target=lambda: replies.append(
|
|
self.command("/draw 100", event_id="concurrent"))) for _ in range(8)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
self.assertEqual(len(replies), 8)
|
|
self.assertEqual(len(set(replies)), 1)
|
|
self.assertEqual(self.state()["balance"], 900)
|
|
def broken(state):
|
|
state["balance"] = -1
|
|
return "bad"
|
|
with self.assertRaises(MODULE.LocalCommandError):
|
|
self.engine.ledger.apply_player_action("group-a", "Alice", "broken", "test", broken)
|
|
self.assertEqual(self.state()["balance"], 900)
|
|
with self.assertRaises(MODULE.LocalCommandError):
|
|
self.command("/draw 100", sender="Bob", event_id="concurrent")
|
|
|
|
def test_invalid_bets_insufficient_funds_and_lottery(self):
|
|
for question in ("/draw -1", "/draw 0", "/draw 101", "/lottery xyz",
|
|
"/blackjack start -1", "/blackjack hit 10"):
|
|
self.assertRegex(self.command(question), "用法|下注")
|
|
with mock.patch.object(self.local.players.rng, "randrange", return_value=99):
|
|
for _ in range(10):
|
|
self.command("/draw 100")
|
|
self.assertIn("摩拉不足", self.command("/draw 2"))
|
|
self.assertEqual(self.state()["balance"], 0)
|
|
|
|
def test_blackjack_is_private_persistent_and_settles_exactly_once(self):
|
|
# Player 10+7, dealer 9+8: standing produces a tie.
|
|
def deck(cards):
|
|
ordered = [9, 6, 8, 7] + [card for card in cards if card not in {9, 6, 8, 7}]
|
|
cards[:] = ordered
|
|
with mock.patch.object(self.local.players.rng, "shuffle", side_effect=deck):
|
|
start = self.command("/21点 开始 50")
|
|
self.assertIn("暗牌", start)
|
|
self.assertEqual(self.state()["balance"], 950)
|
|
self.assertIn("没有进行中", self.command("/blackjack status", sender="Bob"))
|
|
self.assertIn("已有一局", self.command("/draw"))
|
|
self.engine.ledger.connection.close()
|
|
self.engine = MODULE.EdgeApplication(config(self.root))
|
|
self.local = self.engine.local
|
|
settled = self.command("/blackjack stand", sender="a1", event_id="stand-1")
|
|
self.assertIn("平局", settled)
|
|
self.assertEqual(self.state()["balance"], 1000)
|
|
self.assertEqual(self.command("/blackjack stand", event_id="stand-1"), settled)
|
|
self.assertIsNone(self.state()["game"])
|
|
self.assertEqual(self.state()["ties"], 1)
|
|
|
|
def test_aces_natural_bust_timeout_and_end(self):
|
|
self.assertEqual(hand_value([0, 13, 9]), 12)
|
|
self.assertEqual(hand_value([0, 9]), 21)
|
|
def natural(cards):
|
|
first = [0, 9, 8, 7]
|
|
cards[:] = first + [card for card in cards if card not in first]
|
|
with mock.patch.object(self.local.players.rng, "shuffle", side_effect=natural):
|
|
reply = self.command("/blackjack start 50")
|
|
self.assertIn("自然21点获胜", reply)
|
|
self.assertEqual(self.state()["balance"], 1075)
|
|
self.local.players.rng = random.Random(4)
|
|
self.command("/blackjack start 50")
|
|
if self.state()["game"]:
|
|
with mock.patch("jwebot_official_extensions.player_games.time.time", return_value=10**10):
|
|
self.assertIn("超时", self.command("/wallet"))
|
|
self.assertIsNone(self.state()["game"])
|
|
self.local.players.rng = random.Random(4)
|
|
self.command("/blackjack start 50")
|
|
if self.state()["game"]:
|
|
self.assertIn("不退下注", self.command("/blackjack end"))
|
|
self.assertIsNone(self.state()["game"])
|
|
|
|
def test_adapter_and_direct_paths_agree_for_new_plugins(self):
|
|
for index, question in enumerate(("/games", "/wallet", "/people", "/person a1", "/context 200")):
|
|
kind, _ = local_command_parts(question)
|
|
row = {"question": question, "sender": "Alice", "group_fingerprint": "group-a",
|
|
"event_id": f"adapter-{index}"}
|
|
with mock.patch.object(self.engine.ledger, "bootstrap_repeat_history"):
|
|
direct = self.local.handle(row, {})
|
|
adapter = self.engine.extensions.handle(row, {})
|
|
self.assertEqual(adapter, direct)
|
|
|
|
def test_blackjack_bust_cannot_be_acted_on_by_another_player(self):
|
|
def bust_deck(cards):
|
|
first = [9, 22, 8, 7]
|
|
cards[:] = first + [card for card in cards if card not in first and card != 35] + [35]
|
|
with mock.patch.object(self.local.players.rng, "shuffle", side_effect=bust_deck):
|
|
self.command("/blackjack start 100")
|
|
self.assertIn("没有进行中", self.command("/blackjack hit", sender="Bob"))
|
|
self.assertEqual(len(self.state()["game"]["player"]), 2)
|
|
result = self.command("/blackjack hit", event_id="bust")
|
|
self.assertIn("你输了", result)
|
|
self.assertEqual(self.state()["balance"], 900)
|
|
self.assertIsNone(self.state()["game"])
|
|
self.assertEqual(self.command("/blackjack hit", event_id="bust"), result)
|
|
self.assertEqual(self.state()["losses"], 1)
|
|
|
|
def test_simultaneous_natural_blackjack_returns_stake(self):
|
|
def tied_deck(cards):
|
|
first = [0, 9, 13, 22]
|
|
cards[:] = first + [card for card in cards if card not in first]
|
|
with mock.patch.object(self.local.players.rng, "shuffle", side_effect=tied_deck):
|
|
self.assertIn("平局", self.command("/blackjack start 50"))
|
|
self.assertEqual(self.state()["balance"], 1000)
|
|
self.assertIsNone(self.state()["game"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|