""" Sprint 3 Unit & Concurrency Stress Test Suite for FreeExile: - Nền Kinh Tế Bản Vị Huyết Thạch & Cổ Cốt (Primal Bloodstones & Bone Relics) - Ma Trận 15 Tiers Thuộc Tính & Phí Vàng Chế Tác Lũy Thừa - Giao thức Distributed Two-Phase Commit (2PC) chống dupe trên Hắc Thị Hoang Vực - Kiểm thử tải 50.000 giao dịch song song tranh mua cùng 1 món đồ (0% dupe) """ import concurrent.futures import json import os import sys import unittest PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) SERVER_DIR = os.path.join(PROJECT_ROOT, "server") if SERVER_DIR not in sys.path: sys.path.insert(0, SERVER_DIR) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) from server.world.item_crafting_types import ( ItemSlot, ItemRarity, AffixType, PrimalCurrencyType, CraftingItem, AffixMod, ) from server.world.item_crafting_engine import ( ItemCraftingEngine, calculate_crafting_gold_fee, ) from server.world.primal_stones_crafting import ( CraftingEngine as PrimalBarterEngine, GearItem, ItemRarity as PrimalRarity, ) from server.trade.two_phase_commit import ( InstantBuyoutEngine, StashItem, ) from server.trade.consignment_vault import ( UniversalConsignmentVault, VaultItemFilter, ) class TestSprint3EconomyAndBazaar2PC(unittest.TestCase): def setUp(self): self.craft_engine = ItemCraftingEngine() self.barter_engine = PrimalBarterEngine() self.bazaar_engine = InstantBuyoutEngine(market_tax_rate=0.05, db_path=":memory:") self.seller_id = "seller_master_01" self.buyer_id = "buyer_exile_02" self.bazaar_engine.register_account(self.seller_id) self.bazaar_engine.register_account(self.buyer_id) def test_huyet_hon_thach_normal_to_magic(self): """Huyết Hồn Thạch awakens Normal item into Magic with 1-2 affixes.""" item = CraftingItem( item_uuid="primal_axe_01", base_name="Hung Thần Rìu Thô", item_level=75, slot=ItemSlot.WEAPON_1H, rarity=ItemRarity.NORMAL, ) ok, msg, res_item = self.craft_engine.apply_currency(item, PrimalCurrencyType.HUYET_HON_THACH) self.assertTrue(ok) self.assertEqual(res_item.rarity, ItemRarity.MAGIC) self.assertTrue(1 <= res_item.total_affixes_count <= 2) gear = GearItem("g_axe_1", "Rìu Gỉ", 60, PrimalRarity.PHAM_PHAM) b_ok, _ = self.barter_engine.apply_huyet_hon_thach(gear) self.assertTrue(b_ok) self.assertEqual(gear.rarity, PrimalRarity.LINH_PHAM) def test_co_cot_an_regal_magic_to_rare(self): """Cổ Cốt Ấn evolves Magic item into Rare with an additional affix.""" item = CraftingItem( item_uuid="bone_spear_02", base_name="U Minh Cốt Thương", item_level=80, slot=ItemSlot.WEAPON_2H, rarity=ItemRarity.MAGIC, ) item.prefixes.append(AffixMod("p_1", "Sát Thương % (T2)", AffixType.PREFIX, 2, 130)) ok, msg, res_item = self.craft_engine.apply_currency(item, PrimalCurrencyType.CO_COT_AN) self.assertTrue(ok) self.assertEqual(res_item.rarity, ItemRarity.RARE) self.assertEqual(res_item.total_affixes_count, 2) gear = GearItem("g_spear_2", "Ma Thương", 70, PrimalRarity.LINH_PHAM) b_ok, _ = self.barter_engine.apply_co_cot_an(gear) self.assertTrue(b_ok) self.assertEqual(gear.rarity, PrimalRarity.CUC_PHAM) def test_u_minh_cot_dinh_exalted_slam(self): """U Minh Cốt Đinh slams an additional affix onto Rare item (max 6).""" item = CraftingItem( item_uuid="wraith_helm_03", base_name="U Minh Cốt Khôi", item_level=85, slot=ItemSlot.HELMET, rarity=ItemRarity.RARE, ) item.prefixes.append(AffixMod("p_life", "Sinh Lực (T1)", AffixType.PREFIX, 1, 140)) item.suffixes.append(AffixMod("s_res", "Kháng Lôi (T1)", AffixType.SUFFIX, 1, 45)) ok, msg, res_item = self.craft_engine.apply_currency(item, PrimalCurrencyType.U_MINH_COT_DINH) self.assertTrue(ok) self.assertEqual(res_item.total_affixes_count, 3) def test_uan_ma_thach_scour_cleanses_affixes(self): """Uẩn Ma Thạch scours Rare item, cleansing affixes while preserving fractured mods.""" item = CraftingItem( item_uuid="cursed_body_04", base_name="Huyết Bì Giáp", item_level=86, slot=ItemSlot.BODY_ARMOUR, rarity=ItemRarity.RARE, ) item.prefixes.append(AffixMod("p_frac", "Bất Diệt Huyết Mạch (T1)", AffixType.PREFIX, 1, 150, is_fractured=True)) item.suffixes.append(AffixMod("s_temp", "Né Tránh Nhẹ", AffixType.SUFFIX, 10, 15)) ok, msg, res_item = self.craft_engine.apply_currency(item, PrimalCurrencyType.UAN_MA_THACH) self.assertTrue(ok) self.assertEqual(res_item.rarity, ItemRarity.NORMAL) self.assertEqual(len(res_item.prefixes), 1) self.assertTrue(res_item.prefixes[0].is_fractured) self.assertEqual(len(res_item.suffixes), 0) def test_hac_so_thach_reroll_tiers(self): """Hắc Sơ Thạch rerolls tiers for existing affixes while preserving family.""" item = CraftingItem( item_uuid="ring_tier_05", base_name="Cốt Giới", item_level=88, slot=ItemSlot.RING, rarity=ItemRarity.MAGIC, ) item.prefixes.append(AffixMod("pref_life_t14", "Khí Tức Sơ Khai (T14)", AffixType.PREFIX, 14, 12, tier_min=10, tier_max=15)) ok, msg, res_item = self.craft_engine.apply_currency(item, PrimalCurrencyType.HAC_SO_THACH) self.assertTrue(ok) self.assertEqual(msg, "REROLL_TIERS_SUCCESS") self.assertEqual(len(res_item.prefixes), 1) self.assertTrue(res_item.prefixes[0].mod_id.startswith("pref_life_t")) self.assertTrue(1 <= res_item.prefixes[0].tier <= 15) def test_exponential_gold_fee_and_depletion(self): """Crafting fee scales exponentially with iLvl; fails when gold is insufficient.""" fee_low = calculate_crafting_gold_fee(PrimalCurrencyType.HUYET_HON_THACH, 20) fee_high = calculate_crafting_gold_fee(PrimalCurrencyType.HUYET_HON_THACH, 85) self.assertGreater(fee_high, fee_low * 3) item = CraftingItem("costly_base", "Thần Kiếm", 85, ItemSlot.WEAPON_1H, ItemRarity.NORMAL) res_fail = self.craft_engine.apply_primal_currency(item, PrimalCurrencyType.HUYET_HON_THACH, player_gold=10) self.assertFalse(res_fail.success) self.assertEqual(res_fail.status_code, "INSUFFICIENT_GOLD_FEE") res_ok = self.craft_engine.apply_primal_currency(item, PrimalCurrencyType.HUYET_HON_THACH, player_gold=50000) self.assertTrue(res_ok.success) self.assertGreater(res_ok.gold_fee_consumed, 0) def test_2pc_prepare_and_commit_lifecycle(self): """2PC Phase 1 prepares & locks item; Phase 2 commits transfer and updates ledger.""" item_uuid = "sword_2pc_01" self.bazaar_engine.upsert_item(StashItem(item_uuid, self.seller_id, "Tà Ảnh Kiếm", "HUYET_HON_THACH", 100)) self.bazaar_engine.set_balance(self.buyer_id, "HUYET_HON_THACH", 200) prep_ok, msg, tx_id = self.bazaar_engine.prepare_buyout( self.buyer_id, self.seller_id, item_uuid, "HUYET_HON_THACH", 100 ) self.assertTrue(prep_ok) self.assertIsNotNone(tx_id) locked_item = self.bazaar_engine.get_item(item_uuid) self.assertTrue(locked_item.is_locked) self.assertEqual(locked_item.locked_by_tx, tx_id) # Interleaved buy attempt rejected while locked fail_ok, fail_msg, _ = self.bazaar_engine.execute_instant_buyout( self.buyer_id, self.seller_id, item_uuid, "HUYET_HON_THACH", 100 ) self.assertFalse(fail_ok) # Commit Phase 2 com_ok, _ = self.bazaar_engine.commit_buyout(tx_id) self.assertTrue(com_ok) transferred = self.bazaar_engine.get_item(item_uuid) self.assertEqual(transferred.owner_account_id, self.buyer_id) self.assertFalse(transferred.is_locked) def test_2pc_prepare_and_abort_lifecycle(self): """2PC Phase 2 abort unlocks item and reverts transaction state cleanly.""" item_uuid = "bow_abort_02" self.bazaar_engine.upsert_item(StashItem(item_uuid, self.seller_id, "Ma Cung", "HUYET_HON_THACH", 50)) self.bazaar_engine.set_balance(self.buyer_id, "HUYET_HON_THACH", 100) prep_ok, _, tx_id = self.bazaar_engine.prepare_buyout( self.buyer_id, self.seller_id, item_uuid, "HUYET_HON_THACH", 50 ) self.assertTrue(prep_ok) abort_ok, abort_msg = self.bazaar_engine.abort_buyout(tx_id, "Buyer aborted payment") self.assertFalse(abort_ok) self.assertIn("TRANSACTION_ABORTED", abort_msg) item = self.bazaar_engine.get_item(item_uuid) self.assertEqual(item.owner_account_id, self.seller_id) self.assertFalse(item.is_locked) def test_anti_inflation_5pct_tax_burning(self): """Verifies 5% market tax is deducted from seller credit and burned permanently.""" item_uuid = "ring_tax_03" self.bazaar_engine.upsert_item(StashItem(item_uuid, self.seller_id, "Cốt Nhẫn", "HUYET_HON_THACH", 1000)) self.bazaar_engine.set_balance(self.buyer_id, "HUYET_HON_THACH", 1000) self.bazaar_engine.set_balance(self.seller_id, "HUYET_HON_THACH", 0) ok, msg, tx_id = self.bazaar_engine.execute_instant_buyout( self.buyer_id, self.seller_id, item_uuid, "HUYET_HON_THACH", 1000 ) self.assertTrue(ok) self.assertEqual(self.bazaar_engine.get_balance(self.buyer_id, "HUYET_HON_THACH"), 0) self.assertEqual(self.bazaar_engine.get_balance(self.seller_id, "HUYET_HON_THACH"), 950) self.assertEqual(self.bazaar_engine.total_tax_burned, 50) def test_instant_buyout_cross_account_via_consignment_vault(self): """Instant Buyout via UniversalConsignmentVault delists item and transfers ownership.""" vault = UniversalConsignmentVault(market_tax_rate=0.05, db_path=":memory:") vault.buyout_engine.register_account("vault_seller") vault.buyout_engine.register_account("vault_buyer") vault.buyout_engine.set_balance("vault_buyer", "HUYET_HON_THACH", 500) gear = GearItem("v_item_1", "Cuồng Ma Nhận", 85, PrimalRarity.CUC_PHAM) vault.list_item_for_sale("vault_seller", gear, "HUYET_HON_THACH", 200) ok, msg, tx = vault.buy_instant("vault_buyer", "v_item_1") self.assertTrue(ok) self.assertIsNotNone(tx) listings = vault.search_listings(VaultItemFilter()) self.assertEqual(len(listings), 0) def test_scourge_soulbound_item_rejected_from_trade(self): """Scourge soulbound items are rejected from 2PC trade preparation.""" item_uuid = "soulbound_item_04" soul_item = StashItem(item_uuid, self.seller_id, "Hồn Phách Cốt", "HUYET_HON_THACH", 100, is_scourge_soulbound=True) self.bazaar_engine.upsert_item(soul_item) self.bazaar_engine.set_balance(self.buyer_id, "HUYET_HON_THACH", 500) ok, msg, tx = self.bazaar_engine.execute_instant_buyout( self.buyer_id, self.seller_id, item_uuid, "HUYET_HON_THACH", 100 ) self.assertFalse(ok) self.assertIn("soulbound", msg.lower()) def test_50k_concurrent_buyout_stress_zero_duplication(self): """Simulates 50,000 competing buyout requests on 1 single item: exactly 1 wins, 0% dupe.""" target_uuid = "pinnacle_apex_sword_999" apex_sword = StashItem(target_uuid, self.seller_id, "Vô Cực Cổ Ma Trảm", "HUYET_HON_THACH", 500) self.bazaar_engine.upsert_item(apex_sword) total_tx = 50000 bals = json.dumps({"HUYET_HON_THACH": 1000}) bulk_accounts = [(f"swarm_buyer_{i}", bals) for i in range(total_tx)] with self.bazaar_engine.conn: self.bazaar_engine.conn.executemany("INSERT INTO accounts VALUES (?, ?)", bulk_accounts) def attempt_buy(idx: int): return self.bazaar_engine.execute_instant_buyout( buyer_account_id=f"swarm_buyer_{idx}", seller_account_id=self.seller_id, item_uuid=target_uuid, offered_currency="HUYET_HON_THACH", offered_amount=500, ) with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool: results = list(pool.map(attempt_buy, range(total_tx))) successes = [r for r in results if r[0]] failures = [r for r in results if not r[0]] self.assertEqual(len(successes), 1, "Exactly 1 buyer must succeed in 50k contention") self.assertEqual(len(failures), 49999, "All 49,999 competing buyers must fail") # Every failure must indicate item already sold or transaction aborted sample_fail_reasons = {f[1] for f in failures[:50]} for reason in sample_fail_reasons: self.assertTrue( "ITEM_ALREADY_SOLD" in reason or "TRANSACTION_ABORTED" in reason, f"Unexpected failure reason: {reason}" ) # Database audit: verify 0% dupe cur = self.bazaar_engine.conn.cursor() cur.execute("SELECT COUNT(*) FROM items WHERE item_uuid = ?", (target_uuid,)) self.assertEqual(cur.fetchone()[0], 1, "Item must exist exactly once in the database (0% dupe)") item = self.bazaar_engine.get_item(target_uuid) self.assertNotEqual(item.owner_account_id, self.seller_id) self.assertTrue(item.owner_account_id.startswith("swarm_buyer_")) def test_2pc_double_spend_prevention(self): """Verifies Phase 1 escrow prevents double-spending currency on multiple items.""" engine = InstantBuyoutEngine(market_tax_rate=0.05, db_path=":memory:") engine.register_account("seller_ds") engine.register_account("buyer_ds") engine.set_balance("buyer_ds", "HUYET_HON_THACH", 100) engine.upsert_item(StashItem("item_ds_1", "seller_ds", "Rìu 1", "HUYET_HON_THACH", 100)) engine.upsert_item(StashItem("item_ds_2", "seller_ds", "Rìu 2", "HUYET_HON_THACH", 100)) ok1, _, tx1 = engine.prepare_buyout("buyer_ds", "seller_ds", "item_ds_1", "HUYET_HON_THACH", 100) self.assertTrue(ok1) self.assertEqual(engine.get_balance("buyer_ds", "HUYET_HON_THACH"), 0, "Funds must be escrowed in Phase 1") ok2, msg2, _ = engine.prepare_buyout("buyer_ds", "seller_ds", "item_ds_2", "HUYET_HON_THACH", 100) self.assertFalse(ok2) self.assertIn("Insufficient funds", msg2) com1, _ = engine.commit_buyout(tx1) self.assertTrue(com1) self.assertEqual(engine.get_balance("buyer_ds", "HUYET_HON_THACH"), 0) self.assertEqual(engine.get_item("item_ds_1").owner_account_id, "buyer_ds") self.assertEqual(engine.get_item("item_ds_2").owner_account_id, "seller_ds") def test_2pc_abort_refunds_escrowed_funds(self): """Verifies Phase 2 abort refunds escrowed currency back to buyer balance.""" engine = InstantBuyoutEngine(market_tax_rate=0.05, db_path=":memory:") engine.register_account("seller_ab") engine.register_account("buyer_ab") engine.set_balance("buyer_ab", "HUYET_HON_THACH", 100) engine.upsert_item(StashItem("item_abort_2", "seller_ab", "Đao", "HUYET_HON_THACH", 100)) ok, _, tx_id = engine.prepare_buyout("buyer_ab", "seller_ab", "item_abort_2", "HUYET_HON_THACH", 100) self.assertTrue(ok) self.assertEqual(engine.get_balance("buyer_ab", "HUYET_HON_THACH"), 0) engine.abort_buyout(tx_id, "Test cancellation") self.assertEqual(engine.get_balance("buyer_ab", "HUYET_HON_THACH"), 100, "Funds must be restored on abort") self.assertFalse(engine.get_item("item_abort_2").is_locked) def test_2pc_self_trade_rejection(self): """Verifies accounts cannot buy their own listings.""" self.bazaar_engine.upsert_item(StashItem("self_item", self.seller_id, "Tự Mua Kiếm", "HUYET_HON_THACH", 50)) self.bazaar_engine.set_balance(self.seller_id, "HUYET_HON_THACH", 100) ok, msg, _ = self.bazaar_engine.execute_instant_buyout( self.seller_id, self.seller_id, "self_item", "HUYET_HON_THACH", 50 ) self.assertFalse(ok) self.assertIn("own listing", msg.lower()) def test_file_backed_sqlite_concurrency(self): """Verifies multi-threaded concurrency on a real file-backed SQLite database with busy timeout.""" import tempfile with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: tmp_path = tmp.name try: file_engine = InstantBuyoutEngine(market_tax_rate=0.05, db_path=tmp_path) file_engine.register_account("f_seller") file_engine.register_account("f_buyer") file_engine.set_balance("f_buyer", "HUYET_HON_THACH", 1000) file_engine.upsert_item(StashItem("f_item", "f_seller", "Thần Kiếm", "HUYET_HON_THACH", 200)) ok, msg, tx = file_engine.execute_instant_buyout( "f_buyer", "f_seller", "f_item", "HUYET_HON_THACH", 200 ) self.assertTrue(ok) self.assertEqual(file_engine.get_item("f_item").owner_account_id, "f_buyer") file_engine.conn.close() finally: try: os.remove(tmp_path) for ext in ["-wal", "-shm"]: if os.path.exists(tmp_path + ext): os.remove(tmp_path + ext) except OSError: pass if __name__ == "__main__": unittest.main()