""" Unit tests for Guild Shared Stash Service with Two-Phase Commit (2PC) protection. Guarantees zero item duplication, strict role permissions, and immutable audit logs. """ import unittest from server.trade.guild_stash_service import ( GuildStashService, GuildRole, TabAccessType, GuildMember, GuildItem, ) class TestGuildStash2PC(unittest.TestCase): def setUp(self): self.service = GuildStashService() self.guild_id = "guild_celestial_order_001" self.service.create_guild(self.guild_id, guild_name="Celestial Order", master_id="player_master_1") # Add members self.service.add_member(self.guild_id, GuildMember(player_id="player_master_1", player_name="Võ Đế", role=GuildRole.GUILD_MASTER)) self.service.add_member(self.guild_id, GuildMember(player_id="player_officer_1", player_name="Kiếm Tôn", role=GuildRole.OFFICER)) self.service.add_member(self.guild_id, GuildMember(player_id="player_member_1", player_name="Hiệp Khách A", role=GuildRole.MEMBER)) self.service.add_member(self.guild_id, GuildMember(player_id="player_member_2", player_name="Hiệp Khách B", role=GuildRole.MEMBER)) # Add tabs self.tab_public = "tab_general_public" self.tab_officer = "tab_officer_vault" self.tab_deposit_only = "tab_member_donations" self.service.create_tab(self.guild_id, tab_id=self.tab_public, tab_name="Đan Dược & Phôi Đồ", access_type=TabAccessType.PUBLIC) self.service.create_tab(self.guild_id, tab_id=self.tab_officer, tab_name="Bảo Khố Trưởng Lão", access_type=TabAccessType.OFFICER_ONLY) self.service.create_tab(self.guild_id, tab_id=self.tab_deposit_only, tab_name="Đóng Góp Bang Hội", access_type=TabAccessType.DEPOSIT_ONLY) def test_deposit_and_withdraw_public_tab_success(self): item = GuildItem(item_uuid="item_azure_sword_01", item_name="Thanh Phong Kiếm Phôi", item_tier=3) # Member deposits item into public tab success, msg = self.service.deposit_item(self.guild_id, self.tab_public, "player_member_1", item) self.assertTrue(success, msg) # Check item exists in tab tab_items = self.service.get_tab_items(self.guild_id, self.tab_public) self.assertEqual(len(tab_items), 1) self.assertEqual(tab_items[0].item_uuid, "item_azure_sword_01") # Member 2 withdraws item success, msg, withdrawn_item = self.service.withdraw_item(self.guild_id, self.tab_public, "player_member_2", "item_azure_sword_01") self.assertTrue(success, msg) self.assertIsNotNone(withdrawn_item) self.assertEqual(withdrawn_item.item_uuid, "item_azure_sword_01") # Check tab is now empty self.assertEqual(len(self.service.get_tab_items(self.guild_id, self.tab_public)), 0) def test_member_cannot_withdraw_from_officer_tab(self): item = GuildItem(item_uuid="item_celestial_pill_01", item_name="Cửu Chuyển Hoàn Hồn Đan", item_tier=5) self.service.deposit_item(self.guild_id, self.tab_officer, "player_master_1", item) # Regular member tries to withdraw from officer tab -> MUST FAIL success, msg, withdrawn = self.service.withdraw_item(self.guild_id, self.tab_officer, "player_member_1", "item_celestial_pill_01") self.assertFalse(success) self.assertIn("PERMISSION_DENIED", msg) self.assertIsNone(withdrawn) # Officer withdraws -> MUST SUCCEED success, msg, withdrawn = self.service.withdraw_item(self.guild_id, self.tab_officer, "player_officer_1", "item_celestial_pill_01") self.assertTrue(success, msg) self.assertIsNotNone(withdrawn) def test_member_can_deposit_but_not_withdraw_from_deposit_only_tab(self): item = GuildItem(item_uuid="item_spirit_stone_pack", item_name="Hỗn Nguyên Thạch x50", item_tier=4) # Member deposits into deposit-only donation tab success, msg = self.service.deposit_item(self.guild_id, self.tab_deposit_only, "player_member_1", item) self.assertTrue(success, msg) # Member attempts to withdraw their own deposit -> MUST FAIL success, msg, withdrawn = self.service.withdraw_item(self.guild_id, self.tab_deposit_only, "player_member_1", "item_spirit_stone_pack") self.assertFalse(success) self.assertIn("DEPOSIT_ONLY_TAB", msg) self.assertIsNone(withdrawn) # Master withdraws -> MUST SUCCEED success, msg, withdrawn = self.service.withdraw_item(self.guild_id, self.tab_deposit_only, "player_master_1", "item_spirit_stone_pack") self.assertTrue(success) self.assertIsNotNone(withdrawn) def test_concurrent_withdrawal_prevents_item_duplication(self): item = GuildItem(item_uuid="item_rare_dragon_saber", item_name="Thanh Long Đao Phôi", item_tier=5) self.service.deposit_item(self.guild_id, self.tab_public, "player_master_1", item) # Simulate concurrent lock acquired by Thread 1 tx_id_1 = "tx_lock_concurrent_001" acquired = self.service.acquire_lock("item_rare_dragon_saber", tx_id_1) self.assertTrue(acquired) # Thread 2 tries to withdraw while locked -> MUST BE REJECTED success, msg, withdrawn = self.service.withdraw_item(self.guild_id, self.tab_public, "player_member_1", "item_rare_dragon_saber") self.assertFalse(success) self.assertIn("ITEM_LOCKED", msg) self.assertIsNone(withdrawn) # Release lock and complete Thread 1 withdrawal self.service.release_lock("item_rare_dragon_saber", tx_id_1) success, msg, withdrawn = self.service.withdraw_item(self.guild_id, self.tab_public, "player_member_2", "item_rare_dragon_saber") self.assertTrue(success) self.assertIsNotNone(withdrawn) # Once withdrawn, subsequent attempts return not found success, msg, withdrawn = self.service.withdraw_item(self.guild_id, self.tab_public, "player_member_1", "item_rare_dragon_saber") self.assertFalse(success) self.assertIn("ITEM_NOT_FOUND", msg) def test_immutable_audit_log_tracking(self): item = GuildItem(item_uuid="item_audit_check_01", item_name="Thái Hư Kính Mảnh", item_tier=5) self.service.deposit_item(self.guild_id, self.tab_public, "player_member_1", item) self.service.withdraw_item(self.guild_id, self.tab_public, "player_officer_1", "item_audit_check_01") logs = self.service.get_audit_logs(self.guild_id) self.assertGreaterEqual(len(logs), 2) # Check deposit log deposit_log = next(l for l in logs if l.action == "DEPOSIT") self.assertEqual(deposit_log.player_id, "player_member_1") self.assertEqual(deposit_log.item_name, "Thái Hư Kính Mảnh") # Check withdraw log withdraw_log = next(l for l in logs if l.action == "WITHDRAW") self.assertEqual(withdraw_log.player_id, "player_officer_1") self.assertEqual(withdraw_log.item_name, "Thái Hư Kính Mảnh") if __name__ == "__main__": unittest.main()