"""Behavioral coverage for multiplayer lottery and indexed game records.""" import json import tempfile import threading import unittest from pathlib import Path from unittest import mock from test_jwebot_edge import MODULE, config class Release063GameTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.engine = MODULE.EdgeApplication(config(Path(self.temp.name))) 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.engine.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_lottery_number_forms_sort_and_validate(self): parse = self.engine.local.players.parse_lottery_entry self.assertEqual(parse("347"), ((3, 4, 7), 10)) self.assertEqual(parse("743"), ((3, 4, 7), 10)) self.assertEqual(parse("报名 3 4 7 100"), ((3, 4, 7), 100)) self.assertEqual(parse("报名347100"), ((3, 4, 7), 100)) self.assertEqual(parse("743金额500"), ((3, 4, 7), 500)) with self.assertRaisesRegex(ValueError, "互不重复"): parse("337") with self.assertRaisesRegex(ValueError, "1～500"): parse("347501") def test_multiplayer_lottery_status_draw_and_payouts(self): self.assertIn("第 1 把已开始", self.command("/彩票开始")) self.assertIn("alice #1", self.command("/彩票347100")) self.assertIn( "bob #1", self.command("/彩票 报名 9 3 7 200", sender="Bob"), ) self.assertIn( "carol #1", self.command("/彩票 1 2 3 100", sender="Carol"), ) self.assertIn( "Dave #1", self.command("/彩票045100", sender="Dave"), ) status = self.command("/彩票当前状态", sender="Bob") self.assertIn("已报名 4 人", status) self.assertIn("alice #1：3 4 7，下注 100", status) self.assertIn("bob #1：3 7 9，下注 200", status) self.assertIn("只有本轮主持人 alice", self.command( "/彩票开奖", sender="Bob")) with mock.patch.object( self.engine.local.players.rng, "sample", return_value=[7, 3, 9], ): result = self.command("/彩票开奖", event_id="draw-round-1") self.assertIn("开奖号：3 7 9", result) self.assertIn("alice #1：命中 2 个", result) self.assertIn("返还 120", result) self.assertIn("bob #1：命中 3 个", result) self.assertIn("返还 600", result) self.assertIn("carol #1：命中 1 个", result) self.assertIn("返还 50", result) self.assertIn("Dave #1：命中 0 个", result) self.assertEqual(self.state()["balance"], 1020) self.assertEqual(self.state("Bob")["balance"], 1400) self.assertEqual(self.state("Carol")["balance"], 950) self.assertEqual(self.state("Dave")["balance"], 900) self.assertIn("命中2个", self.command("/游戏记录")) self.assertIn("命中3个", self.command( "/个人记录", sender="Bob")) self.assertIn("没有进行中", self.command("/彩票当前状态")) def test_registration_update_keeps_index_and_cancel_refunds(self): self.command("/彩票 开始") first = self.command("/彩票 347 100") updated = self.command("/彩票 743 200", sender="a1") self.assertIn("alice #1", first) self.assertIn("alice #1", updated) self.assertIn("报名已更新", updated) self.assertEqual(self.state()["balance"], 800) status = self.command("/彩票 状态") self.assertIn("已报名 1 人", status) self.assertIn("3 4 7，下注 200", status) self.assertIn("只有本轮主持人 alice", self.command( "/彩票 取消", sender="Bob")) self.assertIn("已全额退回", self.command("/彩票取消")) self.assertEqual(self.state()["balance"], 1000) self.assertIn("净变化 +0", self.command("/游戏记录")) self.command("/彩票开始") second = self.command("/彩票347") self.assertIn("alice #2", second) def test_concurrent_registration_replay_debits_once(self): self.command("/彩票开始") replies = [] errors = [] def register(): try: replies.append( self.command("/彩票347500", event_id="same-entry") ) except BaseException as error: errors.append(error) threads = [threading.Thread(target=register) for _ in range(8)] for thread in threads: thread.start() for thread in threads: thread.join() self.assertFalse(errors) self.assertEqual(len(set(replies)), 1) self.assertEqual(self.state()["balance"], 500) self.assertIn("已报名 1 人", self.command("/彩票状态")) def test_blackjack_uses_1_to_500_and_indexed_history(self): def tied_deck(cards): first = [9, 6, 8, 7] cards[:] = first + [card for card in cards if card not in first] with mock.patch.object( self.engine.local.players.rng, "shuffle", side_effect=tied_deck, ): started = self.command("/21点开始500") self.assertTrue(started.startswith("alice #1：")) self.assertEqual(self.state()["balance"], 500) settled = self.command("/21点停牌") self.assertTrue(settled.startswith("alice #1：")) self.assertIn("平局", settled) self.assertEqual(self.state()["balance"], 1000) history = self.command("/游戏记录") self.assertIn("alice #1：21点 平局", history) self.assertIn("下注 500，返还 500", history) self.assertIn("1～500", self.command("/21点开始501")) def test_multiplayer_blackjack_table_guides_turns_and_settles(self): self.assertIn("黑杰克Blackjack 21点牌桌", self.command("/21点 多人 启动")) self.assertIn("目前玩家：alice", self.command("/21点 状态")) self.assertIn("bob 已加入", self.command("/21点 加入", sender="Bob")) self.assertIn("请alice 放自己的筹码", self.command("/21点 多人 开始")) alice_bet = self.command("/21点 放200", event_id="alice-table-bet") replay = self.command( "/21点 放200", sender="a1", event_id="alice-table-bet", ) self.assertEqual(replay, alice_bet) self.assertIn("请bob 放自己的筹码", alice_bet) self.assertEqual(self.state()["balance"], 800) def table_deck(cards): first = [12, 8, 9, 24, 19, 32, 25] cards[:] = first + [card for card in cards if card not in first] with mock.patch.object( self.engine.local.players.rng, "shuffle", side_effect=table_deck, ): dealt = self.command("/21点 放100", sender="Bob") self.assertIn("全部筹码已确认", dealt) self.assertIn("alice #1：K Q（20点）；下注 200", dealt) self.assertIn("bob #1：9 7（16点）；下注 100", dealt) self.assertIn("请alice 操作", dealt) self.assertEqual(self.state("Bob")["balance"], 900) alice_stand = self.command("/21点 停牌") self.assertIn("alice 停牌", alice_stand) self.assertIn("请bob 操作", alice_stand) settled = self.command("/21点 要牌", sender="Bob") self.assertIn("这一桌每个人的变化", settled) self.assertIn("alice #1：你赢了", settled) self.assertIn("下注 200，返还 400，净变化 +200；摩拉 1200", settled) self.assertIn("bob #1：你输了", settled) self.assertIn("下注 100，返还 0，净变化 -100；摩拉 900", settled) self.assertEqual(self.state()["balance"], 1200) self.assertEqual(self.state("Bob")["balance"], 900) self.assertIn("alice #1：21点 你赢了", self.command("/游戏记录")) self.assertIn( "bob #1：21点 你输了", self.command("/游戏记录", sender="Bob"), ) self.assertIn("+200：21点 获胜", self.command("/金币")) self.assertIn( "-100：21点 失败", self.command("/金币", sender="Bob"), ) def test_prize_draw_has_indexed_history(self): with mock.patch.object( self.engine.local.players.rng, "randrange", return_value=99, ): first = self.command("/抽奖10") second = self.command("/抽奖10") self.assertTrue(first.startswith("alice #1：")) self.assertTrue(second.startswith("alice #2：")) history = self.command("/gamehistory") self.assertIn("alice #2：抽奖 未中奖", history) self.assertIn("alice #1：抽奖 未中奖", history) def test_transfer_requires_initialized_target_and_replays_once(self): self.assertIn("摩拉 1000", self.command("/摩拉")) missing = self.command("/转账 Bob 100", event_id="missing-target") self.assertIn("bob 还没有摩拉账户", missing) self.assertEqual(self.state()["balance"], 1000) self.assertIn("摩拉 1000", self.command("/金币", sender="Bob")) first = self.command("/转账 Bob 250", event_id="transfer-1") replay = self.command( "/转账 Bob 250", sender="a1", event_id="transfer-1", ) self.assertEqual(replay, first) self.assertIn("alice：已向 bob 转账 250 摩拉", first) self.assertEqual(self.state()["balance"], 750) self.assertEqual(self.state("Bob")["balance"], 1250) alice_wallet = self.command("/金币", event_id="wallet-after-transfer-a") bob_wallet = self.command( "/金币", sender="Bob", event_id="wallet-after-transfer-b", ) self.assertIn("alice：当前摩拉 750；无贷款；", alice_wallet) self.assertIn("最近10次账户变动：", alice_wallet) self.assertIn("-250：转账给 bob", alice_wallet) self.assertIn("+1000：首次初始化", alice_wallet) self.assertIn("bob：当前摩拉 1250；无贷款；", bob_wallet) self.assertIn("+250：转账 来自 alice", bob_wallet) def test_transfer_rejects_invalid_self_and_insufficient_cases(self): self.assertIn("摩拉 1000", self.command("/金币")) self.assertIn("转账金额必须为 1～100000", self.command("/转账 Bob 0")) self.assertIn("不能给自己转账", self.command("/转账 a1 100")) self.assertIn("摩拉 1000", self.command("/金币", sender="Bob")) self.assertIn("摩拉不足", self.command("/transfer 1001 Bob")) self.assertEqual(self.state()["balance"], 1000) self.assertEqual(self.state("Bob")["balance"], 1000) def test_loan_repay_and_replay_once(self): self.assertIn("摩拉 1000", self.command("/金币")) first = self.command("/贷款 300", event_id="loan-1") replay = self.command("/贷款 300", event_id="loan-1") self.assertEqual(replay, first) self.assertIn("贷款 300 摩拉已到账", first) self.assertEqual(self.state()["balance"], 1300) self.assertEqual(self.state()["loan"]["principal"], 300) self.assertIn("未还贷款 300", self.command("/金币")) self.assertIn("已有未还贷款 300", self.command("/贷款 100")) partial = self.command("/还款 120") self.assertIn("已还款 120 摩拉", partial) self.assertEqual(self.state()["balance"], 1180) self.assertEqual(self.state()["loan"]["principal"], 180) done = self.command("/还款 全部") self.assertIn("已还清贷款 180 摩拉", done) self.assertEqual(self.state()["balance"], 1000) self.assertIsNone(self.state()["loan"]) wallet = self.command("/金币", event_id="wallet-after-repay") self.assertIn("+300：贷款到账", wallet) self.assertIn("-120：还款", wallet) self.assertIn("-180：还款", wallet) def test_loan_rejects_invalid_amounts_and_empty_repay(self): self.assertIn("贷款金额必须为 1～500", self.command("/贷款 0")) self.assertIn("贷款金额必须为 1～500", self.command("/loan 501")) self.assertIn("当前没有未还贷款", self.command("/还款 全部")) def test_coin_ranking_lists_accounts_by_balance(self): self.assertIn("摩拉 1000", self.command("/金币")) self.assertIn("摩拉 1000", self.command("/金币", sender="Bob")) self.assertIn("摩拉 1000", self.command("/金币", sender="Carol")) self.command("/转账 Bob 250") self.command("/贷款 100", sender="Carol") ranking = self.command("/大富翁榜") self.assertIn("大富翁榜：", ranking) self.assertIn("1. bob 1250 摩拉", ranking) self.assertIn("2. carol 1100 摩拉（未还贷款 100）", ranking) self.assertIn("3. alice 750 摩拉", ranking) self.assertNotIn("Dave", ranking) self.assertIn("摩拉排行人数必须为 1～20", self.command("/摩拉排行 0")) def test_taffy_coin_stats_sums_all_game_history(self): with mock.patch.object( self.engine.local.players.rng, "randrange", return_value=99, ): self.command("/抽奖10") with mock.patch.object( self.engine.local.players.rng, "randrange", return_value=0, ): self.command("/抽奖10", sender="Bob") stats = self.command("/Taffy摩拉") self.assertIn("玩家总下注 20，总返还 50", stats) self.assertIn("Taffy累计倒贴 30 摩拉", stats) self.assertIn("最大上贡者排行（含负数）：", stats) self.assertIn("alice：上贡 10", stats) self.assertIn("bob：从Taffy赢走 40", stats) self.assertIn("Taffy摩拉人数必须为 1～20", self.command("/Taffy摩拉 0")) def test_open_round_and_history_survive_restart(self): root = Path(self.temp.name) self.command("/彩票开始") self.command("/彩票347100") self.engine.ledger.connection.close() self.engine = MODULE.EdgeApplication(config(root)) self.assertIn("alice #1：3 4 7", self.command("/彩票状态")) with mock.patch.object( self.engine.local.players.rng, "sample", return_value=[3, 4, 7], ): self.command("/彩票开奖") self.engine.ledger.connection.close() self.engine = MODULE.EdgeApplication(config(root)) history = self.command("/游戏记录") self.assertIn("alice #1：彩票 命中3个", history) self.assertEqual(self.state()["balance"], 1200) def test_registration_rolls_back_when_response_building_fails(self): ledger = self.engine.ledger ledger.start_group_lottery( "group-a", "Alice", "start", "lottery:start", lambda _: "started", ) def fail(_): raise RuntimeError("render failed") with self.assertRaisesRegex(RuntimeError, "render failed"): ledger.register_group_lottery( "group-a", "Alice", "register", "lottery:347100", (3, 4, 7), 100, fail, ) account = ledger.connection.execute( "SELECT COUNT(*) FROM player_accounts" ).fetchone()[0] entries = ledger.connection.execute( "SELECT COUNT(*) FROM group_lottery_entries" ).fetchone()[0] actions = ledger.connection.execute( "SELECT COUNT(*) FROM player_actions WHERE event_id = 'register'" ).fetchone()[0] self.assertEqual((account, entries, actions), (0, 0, 0)) if __name__ == "__main__": unittest.main()