""" AutoPOE2 - Closed-Loop Logistics & Resilience Unit Tests (Doc 69) ================================================================ Kiểm thử tự động: 1. Auto-Restock Waystone từ Stash khi túi đồ rỗng (INV-WAYSTONE-RESTOCK-MIN). 2. Lọc bỏ Waystone chứa Brick Mod (INV-MAP-NO-BRICK). 3. Tự động kích hoạt chu trình cất đồ hậu map (Post-Map Stash Routine). 4. Phục hồi và click hồi sinh khi phát hiện màn hình tử trận (INV-DEATH-RECOVERY-TIMEOUT). """ import os import unittest from unittest.mock import MagicMock, patch from src.assistant_tool.pickit.waystone_crafter import ( CraftAction, WaystoneAutoCrafter, WaystoneItem, WaystoneRarity, ) from tools.autonomous_mapping_coordinator import AutonomousMappingCoordinator class TestClosedLoopLogistics(unittest.TestCase): def setUp(self): self.coordinator = AutonomousMappingCoordinator( screen_w=2560, screen_h=1440, capture_dir="tests/test_fixtures_real_captures", is_simulated=True, ) self.coordinator.ensure_focus = MagicMock(return_value=True) self.coordinator.press_key = MagicMock() self.coordinator.click = MagicMock() self.coordinator.capture_step = MagicMock(return_value="mock_capture.png") @patch("os.path.exists", return_value=True) def test_stage1_auto_restock_when_inventory_empty(self, mock_exists): """Khi túi đồ rỗng Waystone, Stage 1 phải tự động gọi execute_stage0_stash_routine để rút Waystone.""" self.coordinator.is_inventory_open = MagicMock(return_value=True) self.coordinator.scanner.scan_full_inventory = MagicMock( return_value={"occupied_count": 5, "is_full": False} ) # Lần 1: quét thấy 0 Waystone. Sau khi rút từ Stash, lần 2 quét thấy 1 Waystone mock_ws = {"col": 0, "row": 0, "center_x": 1730, "center_y": 820, "avg_rgb": 150.0} self.coordinator.scanner.scan_all_waystones = MagicMock( side_effect=[[], [mock_ws]] ) # Mock execute_stage0_stash_routine self.coordinator.execute_stage0_stash_routine = MagicMock( return_value={"status": "SUCCESS", "success": True, "waystone_withdrawn": True} ) res = self.coordinator.execute_stage1_precheck(target_slot_idx=0) # Kiểm tra Stage 1 thành công self.assertEqual(res["status"], "SUCCESS") # Kiểm tra execute_stage0_stash_routine đã được gọi để nạp lại Waystone self.coordinator.execute_stage0_stash_routine.assert_called_once() self.assertEqual(len(res["waystones_found"]), 1) @patch("os.path.exists", return_value=True) def test_stage1_skips_brick_waystone(self, mock_exists): """Stage 1 phải tự động bỏ qua viên Waystone dính Brick Mod và chọn viên an toàn tiếp theo.""" self.coordinator.skip_dangerous = True self.coordinator.is_inventory_open = MagicMock(return_value=True) self.coordinator.scanner.scan_full_inventory = MagicMock( return_value={"occupied_count": 5, "is_full": False} ) # 2 viên Waystone trong túi đồ ws_brick = {"col": 0, "row": 0, "center_x": 1730, "center_y": 820, "avg_rgb": 150.0} ws_safe = {"col": 1, "row": 0, "center_x": 1782, "center_y": 820, "avg_rgb": 160.0} self.coordinator.scanner.scan_all_waystones = MagicMock(return_value=[ws_brick, ws_safe]) # Mock crafter: viên đầu tiên bị SKIP_DANGEROUS (Brick), viên thứ 2 là SKIP_CRAFT (Safe) def mock_eval(waystone, alchemy_count, is_in_safe_area=True, scour_count=0): if waystone.inventory_col == 0: return CraftAction.SKIP_DANGEROUS return CraftAction.SKIP_CRAFT self.coordinator.crafter.evaluate_craft_action = MagicMock(side_effect=mock_eval) res = self.coordinator.execute_stage1_precheck(target_slot_idx=0) self.assertEqual(res["status"], "SUCCESS") # Phải chọn viên an toàn ở col 1 thay vì viên brick ở col 0 chosen_col = res["selected_slot"]["col"] if isinstance(res["selected_slot"], dict) else res["selected_slot"][0] self.assertEqual(chosen_col, 1) def test_post_map_stash_routine_executed_in_pipeline(self): """Khi cày xong map, pipeline phải tự động gọi chu trình cất đồ hậu map (Post-Map Stash Routine).""" self.coordinator.dismiss_modals = MagicMock() self.coordinator.execute_stage0_stash_routine = MagicMock( return_value={"status": "SUCCESS", "items_deposited": 3} ) self.coordinator.execute_stage1_precheck = MagicMock( return_value={"status": "SUCCESS", "selected_slot": [0, 0, 100, 100, 150.0]} ) self.coordinator.execute_stage2_map_device = MagicMock(return_value={"status": "SUCCESS"}) self.coordinator.execute_stage3_atlas_activation = MagicMock(return_value={"status": "SUCCESS"}) self.coordinator.execute_stage4_map_entrance = MagicMock( return_value={"status": "SUCCESS", "entered_zone": "Deserted"} ) self.coordinator.execute_stage5_map_clear = MagicMock( return_value={"status": "SUCCESS", "runner_instance": None, "runner_result": None} ) self.coordinator.execute_stage6_portal_return = MagicMock( return_value={"status": "SUCCESS", "returned_to_hideout": True} ) report = self.coordinator.execute_pipeline( target_slot_idx=0, enable_stashing=True, enable_post_stashing=True ) self.assertEqual(report["status"], "SUCCESS") # Phải thực thi cất đồ cả ở Stage 0 đầu tiên và Stage 0 Post-stash sau Stage 6 self.assertIn("STAGE0_POST_STASH", report["stages_completed"]) self.assertEqual(self.coordinator.execute_stage0_stash_routine.call_count, 2) def test_death_resurrect_detected_and_handled(self): """Khi phát hiện màn hình tử trận, Stage 6 phải tự động click Resurrect in Hideout.""" self.coordinator._find_resurrect_button = MagicMock(return_value=(1280, 800)) self.coordinator.log_watcher = None res = self.coordinator.execute_stage6_portal_return(runner=None, map_zone="Deserted") self.assertEqual(res["status"], "SUCCESS") self.assertTrue(res["returned_to_hideout"]) # Đã gọi click vào tọa độ hồi sinh (1280, 800) self.coordinator.click.assert_called_with(1280, 800, delay_after=unittest.mock.ANY) def test_client_crash_recovery_in_coordinator(self): """INV-CLIENT-CRASH-RECOVER: Khi game client bị tắt/crash, coordinator phải gọi recover_client_if_crashed.""" from src.assistant_tool.client_launcher import Poe2ClientLauncher self.coordinator.is_simulated = False with patch.object(Poe2ClientLauncher, "is_client_running", return_value=False), \ patch.object(Poe2ClientLauncher, "recover_client_if_crashed", return_value=True) as mock_recover, \ patch("time.sleep"): recovered = self.coordinator.recover_from_client_crash_if_needed() self.assertTrue(recovered) mock_recover.assert_called_once() if __name__ == "__main__": unittest.main()