"""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()