"""AutoPOE2 - Comprehensive Pickit & Economy Tests (Doc 29 / Doc 43 / Doc 68). =========================================================================== Kiểm thử toàn diện hệ thống Pickit, Waystone Auto-Crafter và Currency Exchange: 1. Dynamic Pickit Filter: - Ngưỡng định giá real-time >= 1.0 Chaos Orb - Bảo toàn danh mục Invariant BASELINE_ALWAYS_PICK (Gold, Waystone, Mirror, Divine, Chaos, Alchemy...) - Loại trừ triệt để rác túi đồ BASELINE_REJECT_SUBSTRINGS (< 1.0 Chaos, scrolls, shards) 2. Waystone T15 Stage 3 Map Device Readiness: - Bất biến INV-CRAFT-WAYSTONE-01: Cấm dùng Orb of Alchemy lên Rare/Unique - Tự động nâng cấp Normal Waystone T15 lên Rare tối ưu - Phát hiện và xử lý Brick Mod nguy hiểm (INV-MAP-NO-BRICK: Reflect, No Regen) - Fail-Closed on Unknown và kích hoạt Ange Exchange khi hết Alchemy 3. Sàn Giao Dịch Faustus Currency Exchange: - Giao thức phím '/' (VK_OEM_2 0xBF) mở sàn, Spacebar (VK_SPACE 0x20) đóng panels - Khóa an toàn khu vực INV-EXCHANGE-SAFE-AREA (Hideout/Town only) - Khóa quỹ vàng sinh tồn INV-EXCHANGE-GOLD-TAX (bảo lưu >= 5,000 vàng) 4. DSL Evaluator & Ground Item Rules: - Phân tích cú pháp IPD và đánh giá vật phẩm rơi trên mặt đất - Bất biến INV-PRE-01: Từ chối UniqueName khi vật phẩm chưa giám định (pre-ID) """ from __future__ import annotations import json import os import tempfile import unittest from unittest.mock import MagicMock, patch import pytest from src.assistant_tool.currency_exchange_orchestrator import ( CurrencyExchangeOrchestrator, ExchangeOrderStatus, ExchangePortfolioConfig, ) from src.assistant_tool.pickit.dynamic_pickit_updater import ( BASELINE_ALWAYS_PICK, BASELINE_REJECT_SUBSTRINGS, get_valuable_market_items, update_pickit_filter_from_market, ) from src.assistant_tool.pickit.evaluator import PickitEvaluator from src.assistant_tool.pickit.market_price_fetcher import MarketPriceFetcher from src.assistant_tool.pickit.parser import parse_ipd from src.assistant_tool.pickit.waystone_crafter import ( CraftAction, WaystoneAutoCrafter, WaystoneItem, WaystoneRarity, ) class TestDynamicPickitFilter(unittest.TestCase): """Kiểm thử Dynamic Pickit Filter và quy tắc định giá thị trường.""" def test_dynamic_pickit_threshold_and_baseline_preservation(self) -> None: """Đảm bảo chỉ nhặt item >= 1.0 Chaos và bảo toàn BASELINE_ALWAYS_PICK.""" mock_market_data = { "schema": "autopoE2.market_prices.v1", "league": "Standard", "updated_at": "2026-09-18 00:00:00", "chaos_orb_raw_price": 50.0, "items": { "Mirror of Kalandra": {"name": "Mirror of Kalandra", "chaos_value": 2500.0}, "Divine Orb": {"name": "Divine Orb", "chaos_value": 2.2}, "Chaos Orb": {"name": "Chaos Orb", "chaos_value": 1.0}, "Exalted Orb": {"name": "Exalted Orb", "chaos_value": 1.5}, "Transmutation Shard": {"name": "Transmutation Shard", "chaos_value": 0.01}, "Scroll of Wisdom": {"name": "Scroll of Wisdom", "chaos_value": 0.005}, "Valuable Unique Amulet": {"name": "Valuable Unique Amulet", "chaos_value": 15.0}, }, } with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as tmp_filter: out_filter_path = tmp_filter.name try: with patch("src.assistant_tool.pickit.dynamic_pickit_updater.get_valuable_market_items") as mock_get: valuable = { "Mirror of Kalandra": 2500.0, "Divine Orb": 2.2, "Chaos Orb": 1.0, "Exalted Orb": 1.5, "Valuable Unique Amulet": 15.0, } mock_get.return_value = (valuable, mock_market_data) result = update_pickit_filter_from_market( min_chaos=1.0, output_paths=[out_filter_path], ) self.assertEqual(result["status"], "success") with open(out_filter_path, "r", encoding="utf-8") as f: data = json.load(f) exact_types = data["exact_types"] # 1. Các item >= 1.0 Chaos phải có trong bộ lọc self.assertIn("Mirror of Kalandra", exact_types) self.assertIn("Divine Orb", exact_types) self.assertIn("Valuable Unique Amulet", exact_types) # 2. Invariant BASELINE_ALWAYS_PICK phải luôn hiện diện for item in BASELINE_ALWAYS_PICK: self.assertIn(item, exact_types, f"Invariant violated: {item} must always be picked!") # 3. Item rác < 1.0 Chaos tuyệt đối không có self.assertNotIn("Scroll of Wisdom", exact_types) self.assertNotIn("Transmutation Shard", exact_types) finally: if os.path.exists(out_filter_path): os.remove(out_filter_path) def test_reject_substrings_integrity(self) -> None: """Kiểm tra danh mục loại trừ rác túi đồ không chứa currency thiết yếu.""" for base in BASELINE_ALWAYS_PICK: base_lower = base.lower() for reject in BASELINE_REJECT_SUBSTRINGS: self.assertNotEqual( base_lower, reject.lower(), f"Conflict detected: Baseline item '{base}' is in reject substrings!", ) class TestWaystoneAutoCrafterT15(unittest.TestCase): """Kiểm thử Waystone Auto-Crafter cho Waystone Tier 15 chuẩn bị vào Stage 3 Map Device.""" def setUp(self) -> None: self.crafter = WaystoneAutoCrafter() def test_waystone_t15_normal_upgraded_to_rare(self) -> None: """Waystone T15 Normal (trắng) phải được tự động nâng cấp thành Rare tối ưu.""" white_t15 = WaystoneItem( name="Deserted Waystone", tier=15, rarity=WaystoneRarity.NORMAL, inventory_col=0, inventory_row=0, ) # Kiểm tra trước khi craft: chưa tối ưu cho Map Device is_opt_before, _ = self.crafter.is_optimal_for_map_device(white_t15) self.assertFalse(is_opt_before) # Nâng cấp bằng prepare_optimal_waystone upgraded, craft_res = self.crafter.prepare_optimal_waystone( waystone=white_t15, alchemy_count=5, ) self.assertTrue(craft_res.success) self.assertEqual(upgraded.rarity, WaystoneRarity.RARE) self.assertGreaterEqual(upgraded.affix_count, 4) # Kiểm tra sau khi craft: đạt chuẩn nạp Map Device is_opt_after, msg = self.crafter.is_optimal_for_map_device(upgraded) self.assertTrue(is_opt_after) self.assertIn("tối ưu", msg) def test_inv_craft_waystone_01_rare_t15_retained(self) -> None: """Bất biến INV-CRAFT-WAYSTONE-01: Waystone T15 Rare không được dùng Alchemy.""" rare_t15 = WaystoneItem( name="Creek Waystone", tier=15, rarity=WaystoneRarity.RARE, inventory_col=1, inventory_row=0, affix_count=4, mods_text="35% increased Monster Pack Size\n70% increased Item Quantity", ) action = self.crafter.evaluate_craft_action(rare_t15, alchemy_count=10) self.assertEqual(action, CraftAction.SKIP_CRAFT) kept, craft_res = self.crafter.prepare_optimal_waystone(rare_t15, alchemy_count=10) self.assertTrue(craft_res.success) self.assertEqual(craft_res.action, CraftAction.SKIP_CRAFT) self.assertEqual(kept.rarity, WaystoneRarity.RARE) def test_waystone_t15_brick_mod_defense(self) -> None: """Bất biến INV-MAP-NO-BRICK: Phát hiện mod phản dame (Reflect) và đề xuất reroll hoặc skip.""" brick_t15 = WaystoneItem( name="Dangerous T15 Waystone", tier=15, rarity=WaystoneRarity.RARE, inventory_col=2, inventory_row=0, mods_text="Monsters reflect 18% of Physical Damage\n40% increased Monster Damage", ) # 1. Phát hiện brick mod is_brick, red_mods = self.crafter.is_brick_waystone(brick_t15.mods_text) self.assertTrue(is_brick) self.assertGreater(len(red_mods), 0) # 2. Không tối ưu nạp Map Device is_opt, reason = self.crafter.is_optimal_for_map_device(brick_t15) self.assertFalse(is_opt) self.assertIn("Brick Mod", reason) # 3. Reroll nếu có Scour + Alch act_reroll = self.crafter.evaluate_craft_action(brick_t15, alchemy_count=5, scour_count=3) self.assertEqual(act_reroll, CraftAction.REROLL_SCOUR_ALCH) # 4. Skip nếu không có Scour act_skip = self.crafter.evaluate_craft_action(brick_t15, alchemy_count=5, scour_count=0) self.assertEqual(act_skip, CraftAction.SKIP_DANGEROUS) def test_waystone_fail_closed_and_ange_trigger(self) -> None: """Fail-Closed on Unknown và chuyển sang Ange Exchange khi hết Alchemy.""" unk_t15 = WaystoneItem( name="Corrupt Waystone", tier=15, rarity=WaystoneRarity.UNKNOWN, inventory_col=0, inventory_row=0, ) self.assertEqual( self.crafter.evaluate_craft_action(unk_t15, alchemy_count=5), CraftAction.ABORT_SAFETY, ) white_t15 = WaystoneItem( name="Normal Waystone", tier=15, rarity=WaystoneRarity.NORMAL, inventory_col=0, inventory_row=0, ) self.assertEqual( self.crafter.evaluate_craft_action(white_t15, alchemy_count=0), CraftAction.TRIGGER_ANGE, ) class TestCurrencyExchangeProtocol(unittest.TestCase): """Kiểm thử giao thức phím '/' và an toàn sàn ngoại hối Faustus Currency Exchange.""" def test_currency_exchange_key_slash_and_spacebar(self) -> None: """Kiểm tra phím '/' (0xBF) mở sàn và Spacebar (0x20) đóng giao diện.""" keys_pressed: list[int] = [] def mock_press_key(key_code: int) -> None: keys_pressed.append(key_code) orchestrator = CurrencyExchangeOrchestrator( press_key_func=mock_press_key, ) wallet = {"Orb of Alteration": 1500} prices = {"Orb of Alteration": 0.05, "Chaos Orb": 1.0} report = orchestrator.execute_session( wallet=wallet, market_prices_in_chaos=prices, current_gold=50000, area_name="Shoreline Hideout", is_simulated=False, now_ms=1000000, ) self.assertEqual(report.status, "SUCCESS") # Phím '/' là 0xBF (VK_OEM_2) self.assertIn(0xBF, keys_pressed) # Phím Spacebar là 0x20 (VK_SPACE) self.assertIn(0x20, keys_pressed) def test_currency_exchange_safety_interlocks(self) -> None: """Kiểm tra khóa an toàn khu vực và bảo lưu quỹ vàng sinh tồn.""" orchestrator = CurrencyExchangeOrchestrator() # 1. Từ chối mở sàn trong Map chiến đấu (INV-EXCHANGE-SAFE-AREA) safe, reason = orchestrator.evaluate_preconditions("The Dread Cellar", current_gold=50000) self.assertFalse(safe) self.assertIn("Vi phạm khu vực an toàn", reason) # 2. Từ chối khi quỹ vàng <= 5,000 (INV-EXCHANGE-GOLD-TAX) safe_gold, reason_gold = orchestrator.evaluate_preconditions("Shoreline Hideout", current_gold=4000) self.assertFalse(safe_gold) self.assertIn("ngưỡng an toàn", reason_gold) class TestPickitDslAndGroundItemRules(unittest.TestCase): """Kiểm thử DSL Parser và Evaluator theo Doc 29.""" def test_ground_item_pre_id_rejects_uniquename(self) -> None: """Bất biến INV-PRE-01: Từ chối lọc UniqueName đối với item chưa giám định trên đất.""" ipd_text = '[UniqueName] == "Headhunter" # [StashItem] == "true"' rules = parse_ipd(ipd_text) evaluator = PickitEvaluator(rules) ground_item = { "identified": False, "type": "Heavy Belt", "unique_name": "Headhunter", } res = evaluator.evaluate(ground_item) self.assertFalse(res.pickup_candidate) def test_waystone_tier_rule_evaluation(self) -> None: """Đánh giá rule nhặt Waystone Tier cao.""" ipd_text = '[Category] == "Waystone" && [WaystoneTier] >= "15" # [RunMap] == "true"' rules = parse_ipd(ipd_text) evaluator = PickitEvaluator(rules) t15_item = { "identified": True, "category": "Waystone", "waystone_tier": 15, } res = evaluator.evaluate(t15_item) self.assertTrue(res.pickup_candidate) self.assertIn("RunMap", res.actions) if __name__ == "__main__": unittest.main()