""" Unit Tests for AutonomousMappingCoordinator (Doc 48 / Doc 56 / Doc 43) ====================================================================== Kiểm thử các bất biến kiến trúc: - INV-FARM-CLOSED-LOOP: Kiểm tra chuỗi 6 giai đoạn điều phối. - INV-TOWN-PORTAL-KEY: Bảo đảm phím '.' (VK 0xBE) được sử dụng. - INV-ATLAS-01: An toàn tại Hideout. - INV-ATLAS-02: Fail-closed khi không có Waystone hoặc lỗi tọa độ. - INV-CI-01: Cấu hình Vitals CI (4284 ES, 897 Mana, 241 Ward). """ import math import os import pytest from pathlib import Path from unittest.mock import MagicMock, patch from tools.autonomous_mapping_coordinator import AutonomousMappingCoordinator def test_coordinator_initialization(): coordinator = AutonomousMappingCoordinator( screen_w=2560, screen_h=1440, capture_dir="captures/test_pipeline", map_duration_sec=10, ) assert coordinator.screen_w == 2560 assert coordinator.screen_h == 1440 assert coordinator.map_duration_sec == 10 assert coordinator.scanner is not None assert coordinator.selector is not None def test_coordinator_scanner_integration(): coordinator = AutonomousMappingCoordinator(screen_w=2560, screen_h=1440) # Kiểm tra tọa độ Slot 0 Hàng 1 cx, cy = coordinator.scanner.get_slot_center(0, 0) assert cx == 1731 assert cy == 822 def test_coordinator_stage_definitions(): expected_stages = [ "STAGE1_PRECHECK", "STAGE2_MAP_DEVICE", "STAGE3_ATLAS_ACTIVATION", "STAGE4_MAP_ENTRANCE", "STAGE5_MAP_CLEAR", "STAGE6_PORTAL_RETURN", ] assert len(expected_stages) == 6 def test_stage1_precheck_success(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_inv.png") coordinator.is_inventory_open = MagicMock(return_value=True) with patch("os.path.exists", return_value=True): coordinator.scanner.scan_full_inventory = MagicMock(return_value={ "occupied_count": 20, "is_full": False, }) coordinator.scanner.scan_all_waystones = MagicMock(return_value=[ (0, 0, 1731, 822, 65.5), (1, 0, 1801, 822, 58.2), ]) res = coordinator.execute_stage1_precheck() assert res["status"] == "SUCCESS" assert res["occupied_slots"] == 20 assert not res["is_full"] assert not res["hooded_one_triggered"] assert len(res["waystones_found"]) == 2 assert res["selected_slot"] == (0, 0, 1731, 822, 65.5) def test_stage1_precheck_full_inventory_triggers_hooded_one(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_inv.png") coordinator.is_inventory_open = MagicMock(return_value=True) with patch("os.path.exists", return_value=True): coordinator.scanner.scan_full_inventory = MagicMock(return_value={ "occupied_count": 52, "is_full": True, }) coordinator.scanner.check_and_trigger_hooded_one_identify = MagicMock(return_value={ "triggered": True, "status": "IDENTIFY_REQUESTED", }) coordinator.scanner.scan_all_waystones = MagicMock(return_value=[ (0, 0, 1731, 822, 65.5), ]) res = coordinator.execute_stage1_precheck() assert res["status"] == "SUCCESS" assert res["is_full"] is True assert res["hooded_one_triggered"] is True coordinator.scanner.check_and_trigger_hooded_one_identify.assert_called_once() def test_stage1_precheck_no_waystones_fails_closed(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_inv.png") coordinator.is_inventory_open = MagicMock(return_value=True) with patch("os.path.exists", return_value=True): coordinator.scanner.scan_full_inventory = MagicMock(return_value={ "occupied_count": 10, "is_full": False, }) coordinator.scanner.scan_all_waystones = MagicMock(return_value=[]) res = coordinator.execute_stage1_precheck() assert res["status"] == "FAILED" assert "Fail-closed" in res["error"] assert res["selected_slot"] is None def test_stage1_precheck_inventory_not_open_fails_closed(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_inv.png") coordinator.is_inventory_open = MagicMock(return_value=False) with patch("os.path.exists", return_value=True): res = coordinator.execute_stage1_precheck() assert res["status"] == "FAILED" assert res["error"] == "INVENTORY_NOT_OPEN" def test_stage2_map_device(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.click = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_device.png") coordinator._core_owns_map_device_loop = MagicMock(return_value=False) coordinator.find_map_device_screen_pos = MagicMock(return_value=(1150, 600)) coordinator.send_wasd_direction = MagicMock() coordinator.is_atlas_ui_open = MagicMock(return_value=True) with patch("ctypes.windll.user32.mouse_event") as mock_mouse_event, patch("time.sleep"): res = coordinator.execute_stage2_map_device() assert res["status"] == "SUCCESS" assert res["zoom_applied"] is True # INV-KEY-NO-U-FOR-MAP-DEVICE & INV-KEY-NO-F-FOR-INTERACT: Cấm gọi VK_U (0x55) và VK_F (0x46) called_keys = [call[0][0] for call in coordinator.press_key.call_args_list] assert 0x55 not in called_keys, "CẤM TUYỆT ĐỐI gửi phím VK_U (0x55) khi mở Map Device!" assert 0x46 not in called_keys, "CẤM TUYỆT ĐỐI gửi phím VK_F (0x46) để tương tác Map Device (F là Skill 8)!" assert 0x1B not in called_keys, "CẤM TUYỆT ĐỐI gửi phím ESC (0x1B) mù khi mở Map Device!" assert coordinator.click.call_count >= 3 coordinator.click.assert_any_call(1150, 600, delay_after=2.5) coordinator.click.assert_any_call(1647, 94, delay_after=2.0) # Mouse event called for zoom-in and zoom-out assert mock_mouse_event.call_count >= 8 def test_stage2_map_device_fallback_pedestal(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.click = MagicMock() coordinator.send_wasd_direction = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_device.png") coordinator._core_owns_map_device_loop = MagicMock(return_value=False) coordinator.find_map_device_screen_pos = MagicMock(return_value=(536, 280)) coordinator.is_atlas_ui_open = MagicMock(return_value=True) with patch("ctypes.windll.user32.mouse_event"), patch("time.sleep"): res = coordinator.execute_stage2_map_device() # Fallback click bệ đá Map Device Tây Bắc tại Shoreline Hideout: (536, 280) coordinator.click.assert_any_call(536, 280, delay_after=2.5) def test_stage2_skips_when_core_owns_map_device(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.click = MagicMock() coordinator._core_owns_map_device_loop = MagicMock(return_value=True) res = coordinator.execute_stage2_map_device() assert res["status"] == "SUCCESS" assert res["skipped_core_owner"] is True coordinator.press_key.assert_not_called() coordinator.click.assert_not_called() def test_stage2_game_menu_guard_clicks_resume(): """INV-GAME-MENU-GUARD: Khi Game Menu bị mở ở Stage 2, tự động click RESUME GAME giải phóng menu.""" coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.click = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_device.png") coordinator._core_owns_map_device_loop = MagicMock(return_value=False) coordinator.find_map_device_screen_pos = MagicMock(return_value=(1150, 600)) coordinator.send_wasd_direction = MagicMock() coordinator.is_atlas_ui_open = MagicMock(return_value=True) with patch("src.common.is_game_menu_visible", side_effect=[True, False, False, False]), \ patch("ctypes.windll.user32.mouse_event"), patch("time.sleep"): res = coordinator.execute_stage2_map_device() assert res["status"] == "SUCCESS" # Click giải phóng nút RESUME GAME tại 50% W, 23% H expected_resume_x = int(coordinator.screen_w * 0.50) expected_resume_y = int(coordinator.screen_h * 0.23) coordinator.click.assert_any_call(expected_resume_x, expected_resume_y, delay_after=0.4) def test_dynamic_camera_zoom(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) with patch("ctypes.windll.user32.mouse_event") as mock_mouse_event: coordinator.dynamic_camera_zoom(zoom_in=True, steps=4, step_delay=0.001) assert mock_mouse_event.call_count == 4 mock_mouse_event.assert_called_with(0x0800, 0, 0, 120, 0) mock_mouse_event.reset_mock() coordinator.dynamic_camera_zoom(zoom_in=False, steps=4, step_delay=0.001) assert mock_mouse_event.call_count == 4 mock_mouse_event.assert_called_with(0x0800, 0, 0, -120, 0) def test_compute_tangent_slide_geometry(): coordinator = AutonomousMappingCoordinator() # Zero displacement returns (0, 0, 0) bx, by, hand = coordinator.compute_tangent_slide(0.0, 0.0, 0.0, 0.0, 10.0, 10.0) assert (bx, by, hand) == (0.0, 0.0, 0) # Heading straight along X: desired_dx=10, desired_dy=0, goal at (10, 10) (Y > 0) bx, by, hand = coordinator.compute_tangent_slide(0.0, 0.0, 10.0, 0.0, 10.0, 10.0, preferred_hand=0) # Left tangent CCW (+90 deg) is (0, 1), right tangent CW (-90 deg) is (0, -1) # Goal direction is (10, 10) -> normalized (0.707, 0.707). # Left tangent dot goal = (0)*(0.707) + (1)*(0.707) = 0.707 > -0.707 (right) # So left hand is preferred: hand == 1 assert hand == 1 # Blended vector should have positive Y (along left tangent) and positive X (forward) assert by > 0.0 assert bx > 0.0 # Test preferred_hand hysteresis: forcing right hand (-1) gives negative Y bx_r, by_r, hand_r = coordinator.compute_tangent_slide(0.0, 0.0, 10.0, 0.0, 10.0, 10.0, preferred_hand=-1) assert hand_r == -1 assert by_r < 0.0 assert bx_r > 0.0 def test_stage2_map_device_approaches_when_distant(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.click = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_device.png") coordinator.navigate_towards_with_tangent_sliding = MagicMock(return_value=True) coordinator._core_owns_map_device_loop = MagicMock(return_value=False) coordinator.is_atlas_ui_open = MagicMock(return_value=True) with patch("ctypes.windll.user32.mouse_event"): # Map device at (100, 200), player at (100, 100) -> Dist = 100 > 18.0u res = coordinator.execute_stage2_map_device( device_world_pos=(100.0, 200.0), player_world_pos=(100.0, 100.0), ) assert res["status"] == "SUCCESS" assert res["approached"] is True coordinator.navigate_towards_with_tangent_sliding.assert_called_once_with( 100.0, 100.0, 100.0, 200.0, interact_radius=18.0, max_duration_sec=8.0 ) def test_project_map_device_world_to_screen(): coordinator = AutonomousMappingCoordinator() # At exact player position (dist < 0.001), returns screen center cx, cy = coordinator.project_map_device_world_to_screen(100.0, 100.0, 100.0, 100.0) assert cx == 1280 assert cy == 720 # Distant device: Zoomed-in vs Zoomed-out should have greater pixel displacement when zoomed in px_zoom, py_zoom = coordinator.project_map_device_world_to_screen(120.0, 100.0, 100.0, 100.0, is_zoomed_in=True) px_norm, py_norm = coordinator.project_map_device_world_to_screen(120.0, 100.0, 100.0, 100.0, is_zoomed_in=False) # Distance from center should be greater when zoomed in dist_zoom = math.hypot(px_zoom - 1280, py_zoom - 720) dist_norm = math.hypot(px_norm - 1280, py_norm - 720) assert dist_zoom > dist_norm def test_navigate_towards_with_tangent_sliding(): coordinator = AutonomousMappingCoordinator() coordinator.send_wasd_direction = MagicMock() coordinator.get_player_world_pos = MagicMock(return_value=None) # If already within interact radius, returns True immediately without moving reached = coordinator.navigate_towards_with_tangent_sliding(100.0, 100.0, 105.0, 100.0, interact_radius=18.0) assert reached is True assert coordinator.send_wasd_direction.call_count == 0 def test_stage3_atlas_activation(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.find_reachable_atlas_node = MagicMock(return_value=(725, 969)) coordinator.selector.select_node = MagicMock(return_value=True) coordinator.is_inventory_open = MagicMock(return_value=False) coordinator.press_key = MagicMock() coordinator.ctrl_click = MagicMock() coordinator.click = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_activated.png") coordinator.find_traverse_button = MagicMock(return_value=(1075, 750)) with patch("time.sleep"): res = coordinator.execute_stage3_atlas_activation(waystone_slot=(0, 1731, 822, 65.5)) assert res["status"] == "SUCCESS" coordinator.find_reachable_atlas_node.assert_called_once() coordinator.selector.select_node.assert_called_once() coordinator.press_key.assert_called_once_with(0x49, 0x17, delay_after=0.5) coordinator.ctrl_click.assert_called_once_with(1731, 822, delay_after=0.5) coordinator.click.assert_any_call(725, 969, delay_after=0.8) coordinator.click.assert_any_call(1075, 750, delay_after=0.5) def test_stage3_inventory_not_reclosed_if_already_open(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.find_reachable_atlas_node = MagicMock(return_value=(725, 969)) coordinator.selector.select_node = MagicMock(return_value=True) coordinator.is_inventory_open = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.ctrl_click = MagicMock() coordinator.click = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_activated.png") coordinator.find_traverse_button = MagicMock(return_value=(1075, 750)) with patch("time.sleep"): res = coordinator.execute_stage3_atlas_activation(waystone_slot=(0, 1731, 822, 65.5)) assert res["status"] == "SUCCESS" coordinator.press_key.assert_not_called() coordinator.ctrl_click.assert_called_once_with(1731, 822, delay_after=0.5) def test_stage3_traverse_ocr_fail_closed(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.find_reachable_atlas_node = MagicMock(return_value=(725, 969)) coordinator.selector.select_node = MagicMock(return_value=True) coordinator.is_inventory_open = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.ctrl_click = MagicMock() coordinator.click = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_activated.png") coordinator.find_traverse_button = MagicMock(return_value=None) with patch("time.sleep"): res = coordinator.execute_stage3_atlas_activation(waystone_slot=(0, 1731, 822, 65.5)) assert res["status"] == "FAILED" assert "TRAVERSE" in res["error"] assert not any(c.args[:2] == (1075, 750) for c in coordinator.click.call_args_list) def test_stage2_atlas_ui_unverified_fail_closed(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.click = MagicMock() coordinator.send_wasd_direction = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_device.png") coordinator._core_owns_map_device_loop = MagicMock(return_value=False) coordinator.find_map_device_screen_pos = MagicMock(return_value=(1150, 600)) coordinator.is_atlas_ui_open = MagicMock(return_value=False) with patch("ctypes.windll.user32.mouse_event"), patch("time.sleep"): res = coordinator.execute_stage2_map_device() assert res["status"] == "FAILED" assert "INV-MAPDEVICE-VERIFY-UI" in res["error"] # Đã click mở Map Device nhưng UI không mở -> từ chối click search/ENDGAME coordinator.click.assert_any_call(1150, 600, delay_after=2.5) def test_stage4_map_entrance(): coordinator = AutonomousMappingCoordinator(screen_w=2560, screen_h=1440) coordinator.ensure_focus = MagicMock(return_value=True) coordinator.dismiss_modals = MagicMock() coordinator.click = MagicMock() coordinator.hover = MagicMock() coordinator.press_key = MagicMock() coordinator.send_wasd_direction = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_entered.png") coordinator.find_portal_screen_pos = MagicMock(return_value=(705, 450)) coordinator.log_watcher = None with patch("time.sleep"): res = coordinator.execute_stage4_map_entrance() assert res["status"] == "SUCCESS" assert res["zone_transitioned"] is True coordinator.dismiss_modals.assert_called_once() coordinator.click.assert_called_once_with(705, 450, delay_after=0.3) coordinator.press_key.assert_not_called() assert coordinator.send_wasd_direction.call_count == 2 def test_stage4_map_entrance_timeout_fails_closed(): coordinator = AutonomousMappingCoordinator(screen_w=2560, screen_h=1440) coordinator.ensure_focus = MagicMock(return_value=True) coordinator.dismiss_modals = MagicMock() coordinator.click = MagicMock() coordinator.hover = MagicMock() coordinator.press_key = MagicMock() coordinator.send_wasd_direction = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_entered.png") coordinator.find_portal_screen_pos = MagicMock(return_value=(705, 450)) mock_watcher = MagicMock() mock_zone = MagicMock() mock_zone.area_name = "Shoreline Hideout" mock_watcher.seed_latest_playable_zone.return_value = mock_zone coordinator.log_watcher = mock_watcher with patch("time.sleep"): res = coordinator.execute_stage4_map_entrance() assert res["status"] == "FAILED" assert res["zone_transitioned"] is False assert "Fail-closed" in res["error"] def test_stage5_map_clear(): coordinator = AutonomousMappingCoordinator(map_duration_sec=15) coordinator.ensure_focus = MagicMock(return_value=True) with patch("tools.autonomous_mapping_coordinator.AutonomousMapRunner") as MockRunner: mock_instance = MockRunner.return_value mock_instance.start.return_value = { "status": "COMPLETED", "death_detected": False, "captured_frames": 5, } res = coordinator.execute_stage5_map_clear() assert res["status"] == "SUCCESS" assert res["runner_result"]["death_detected"] is False MockRunner.assert_called_once_with( core_exe=str(Path("bin/AutoPOE2_Core.exe")), max_duration_sec=15, min_duration_sec=90.0, capture_interval_sec=2.5, portal_exit=False, max_es=4284, max_mana=897, max_ward=241, minimap_adapter=coordinator.minimap_nav_adapter, wasd_sender=coordinator.send_wasd_direction, ) def test_stage6_portal_return_with_runner(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.capture_step = MagicMock(return_value="captures/fake_returned.png") coordinator.log_watcher = None mock_runner = MagicMock() with patch("time.sleep"): res = coordinator.execute_stage6_portal_return(runner=mock_runner, map_zone="Deserted Map") assert res["status"] == "SUCCESS" assert res["returned_to_hideout"] is True mock_runner.execute_portal_exit.assert_called_once() def test_stage6_portal_return_timeout_fails_closed(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.capture_step = MagicMock(return_value="captures/fake_returned.png") mock_watcher = MagicMock() mock_zone = MagicMock() mock_zone.area_name = "Deserted Map" mock_watcher.seed_latest_playable_zone.return_value = mock_zone coordinator.log_watcher = mock_watcher with patch("time.sleep"): res = coordinator.execute_stage6_portal_return(runner=None, map_zone="Deserted Map") assert res["status"] == "FAILED" assert res["returned_to_hideout"] is False assert "Fail-closed" in res["error"] def test_stage6_portal_return_standalone(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.press_key = MagicMock() coordinator.click = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_returned.png") coordinator.log_watcher = None with patch("time.sleep"): res = coordinator.execute_stage6_portal_return(runner=None, map_zone="Deserted Map") assert res["status"] == "SUCCESS" assert res["returned_to_hideout"] is True # Kiểm tra phím '.' (VK 0xBE / scan 0x34) và click LMB vào cổng (CẤM phím F) coordinator.press_key.assert_called_once_with(0xBE, 0x34, delay_after=1.5) called_keys = [call[0][0] for call in coordinator.press_key.call_args_list] assert 0x46 not in called_keys, "CẤM TUYỆT ĐỐI gửi phím VK_F (0x46) để vào portal!" coordinator.click.assert_called_once_with(1280, 760, delay_after=2.5) def test_execute_pipeline_success(): coordinator = AutonomousMappingCoordinator(is_simulated=True) coordinator.check_active_portal_in_hideout = MagicMock(return_value=None) coordinator.execute_stage1_precheck = MagicMock(return_value={ "status": "SUCCESS", "selected_slot": (0, 1731, 822, 65.5), }) coordinator.execute_stage2_map_device = MagicMock(return_value={"status": "SUCCESS"}) coordinator.execute_stage3_atlas_activation = MagicMock(return_value={"status": "SUCCESS"}) coordinator.execute_stage4_map_entrance = MagicMock(return_value={"status": "SUCCESS"}) coordinator.execute_stage5_map_clear = MagicMock(return_value={ "status": "SUCCESS", "runner_result": {"death_detected": False}, "runner_instance": MagicMock(), }) coordinator.execute_stage6_portal_return = MagicMock(return_value={"status": "SUCCESS"}) report = coordinator.execute_pipeline() assert report["status"] == "SUCCESS" assert len(report["stages_completed"]) == 6 assert report["stages_completed"] == [ "STAGE1_PRECHECK", "STAGE2_MAP_DEVICE", "STAGE3_ATLAS_ACTIVATION", "STAGE4_MAP_ENTRANCE", "STAGE5_MAP_CLEAR", "STAGE6_PORTAL_RETURN", ] coordinator.execute_stage1_precheck.assert_called_once() coordinator.execute_stage2_map_device.assert_called_once() coordinator.execute_stage3_atlas_activation.assert_called_once() coordinator.execute_stage4_map_entrance.assert_called_once() coordinator.execute_stage5_map_clear.assert_called_once() coordinator.execute_stage6_portal_return.assert_called_once() def test_execute_pipeline_stage1_fail_halts(): coordinator = AutonomousMappingCoordinator(is_simulated=True) coordinator.check_active_portal_in_hideout = MagicMock(return_value=None) coordinator.execute_stage1_precheck = MagicMock(return_value={ "status": "FAILED", "error": "Stage 1 thất bại", }) coordinator.execute_stage2_map_device = MagicMock() report = coordinator.execute_pipeline() assert report["status"] == "FAILED" assert "Stage 1 thất bại" in report["error"] assert report["stages_completed"] == [] coordinator.execute_stage2_map_device.assert_not_called() def test_execute_pipeline_portal_bypass(): coordinator = AutonomousMappingCoordinator(is_simulated=True) coordinator.check_active_portal_in_hideout = MagicMock(return_value=(600, 300, "BURNING MONOLITH")) coordinator.execute_stage1_precheck = MagicMock() coordinator.execute_stage2_map_device = MagicMock() coordinator.execute_stage3_atlas_activation = MagicMock() coordinator.execute_stage4_map_entrance = MagicMock(return_value={"status": "SUCCESS", "entered_zone": "Burning Monolith"}) coordinator.execute_stage5_map_clear = MagicMock(return_value={ "status": "SUCCESS", "runner_result": {"death_detected": False}, "runner_instance": MagicMock(), }) coordinator.execute_stage6_portal_return = MagicMock(return_value={"status": "SUCCESS"}) report = coordinator.execute_pipeline() assert report["status"] == "SUCCESS" assert report["stages_completed"] == ["STAGE4_MAP_ENTRANCE", "STAGE5_MAP_CLEAR", "STAGE6_PORTAL_RETURN"] coordinator.execute_stage1_precheck.assert_not_called() coordinator.execute_stage2_map_device.assert_not_called() coordinator.execute_stage3_atlas_activation.assert_not_called() coordinator.execute_stage4_map_entrance.assert_called_once() def test_optical_map_device_locator_authentic_capture(): coordinator = AutonomousMappingCoordinator(screen_w=2560, screen_h=1440) img_path = "captures/20260913_215022_379_LIVE_VERIFICATION.png" if os.path.exists(img_path): mx, my = coordinator.find_map_device_screen_pos(img_path) assert (1800 <= mx <= 1890 and 400 <= my <= 550) or (mx in (536, 1120, 1175) and my in (230, 280, 645)) else: # Fallback test mx, my = coordinator.find_map_device_screen_pos(None) assert mx in (536, 1120, 1175, 1866) assert my in (230, 280, 645, 471) def test_optical_portal_locator_authentic_capture(): coordinator = AutonomousMappingCoordinator(screen_w=2560, screen_h=1440) img_path = "captures/20260913_215022_379_LIVE_VERIFICATION.png" if os.path.exists(img_path): px, py = coordinator.find_portal_screen_pos(img_path) assert 1200 <= px <= 1700 assert 280 <= py <= 600 else: # Fallback test px, py = coordinator.find_portal_screen_pos(None) assert px in (506, 705) assert py in (240, 450) def test_optical_reachable_atlas_node_locator(): coordinator = AutonomousMappingCoordinator(screen_w=2560, screen_h=1440) # Test authentic capture with green nodes img_path = "captures/20260914_055336_887_LIVE_CLEAR_SEARCH_TEST_CORRELATED.png" if os.path.exists(img_path): nx, ny = coordinator.find_reachable_atlas_node(img_path) assert 100 <= nx <= 2000 assert 300 <= ny <= 1200 else: # Fallback test nx, ny = coordinator.find_reachable_atlas_node(None) assert nx == 1043 assert ny == 810 def test_dismiss_modals_precheck(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.click = MagicMock() coordinator.press_key = MagicMock() coordinator.capture_step = MagicMock(return_value="") coordinator.dismiss_modals() assert coordinator.ensure_focus.call_count >= 1 assert coordinator.click.call_count == 2 coordinator.click.assert_any_call(1626, 44, delay_after=0.15) coordinator.click.assert_any_call(2530, 44, delay_after=0.15) coordinator.press_key.assert_called_once_with(0x20, 0x39, delay_after=0.2) def test_waystone_auto_crafter_integration_rare_skip(): coordinator = AutonomousMappingCoordinator() coordinator.ensure_focus = MagicMock(return_value=True) coordinator.dismiss_modals = MagicMock() coordinator.press_key = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_inv.png") coordinator.is_inventory_open = MagicMock(return_value=True) with patch("os.path.exists", return_value=True): coordinator.scanner.scan_full_inventory = MagicMock(return_value={ "occupied_count": 25, "is_full": False, }) coordinator.scanner.scan_all_waystones = MagicMock(return_value=[ (0, 0, 1731, 822, 104.2), ]) res = coordinator.execute_stage1_precheck(target_slot_idx=0) assert res["status"] == "SUCCESS" assert res["craft_action"] == "SKIP_CRAFT" assert res["selected_slot"] == (0, 0, 1731, 822, 104.2) def test_run_multi_map_loop_success(): coordinator = AutonomousMappingCoordinator() coordinator.recover_from_client_crash_if_needed = MagicMock(return_value=True) coordinator.start_3s_sentinel = MagicMock() coordinator.stop_3s_sentinel = MagicMock() success_report = { "status": "SUCCESS", "stages_completed": ["STAGE1_PRECHECK", "STAGE2_MAP_DEVICE", "STAGE3_ATLAS_ACTIVATION", "STAGE4_MAP_ENTRANCE", "STAGE5_MAP_CLEAR", "STAGE6_PORTAL_RETURN"], "stage1_data": {"waystones_found": [(0, 1731, 822, 104.2)]}, "runner_result": {"death_detected": False}, } coordinator.execute_pipeline = MagicMock(return_value=success_report) with patch("time.sleep"): res = coordinator.run_multi_map_loop(max_maps=2) assert res["status"] == "COMPLETED" assert res["total_maps_requested"] == 2 assert res["maps_completed"] == 2 assert res["exhausted"] is False assert len(res["runs"]) == 2 assert coordinator.execute_pipeline.call_count == 2 def test_run_multi_map_loop_stops_on_exhaustion(): coordinator = AutonomousMappingCoordinator() coordinator.recover_from_client_crash_if_needed = MagicMock(return_value=True) coordinator.start_3s_sentinel = MagicMock() coordinator.stop_3s_sentinel = MagicMock() success_report = { "status": "SUCCESS", "stages_completed": ["STAGE1_PRECHECK", "STAGE2_MAP_DEVICE", "STAGE3_ATLAS_ACTIVATION", "STAGE4_MAP_ENTRANCE", "STAGE5_MAP_CLEAR", "STAGE6_PORTAL_RETURN"], "stage1_data": {"waystones_found": [(0, 1731, 822, 104.2)]}, "runner_result": {"death_detected": False}, } exhausted_report = { "status": "FAILED", "stages_completed": [], "stage1_data": {"waystones_found": []}, "error": "Stage 1 thất bại: Khong tim thay Waystone nao trong Hang 1 (Row 0)! Fail-closed.", } coordinator.execute_pipeline = MagicMock(side_effect=[success_report, exhausted_report]) with patch("time.sleep"): res = coordinator.run_multi_map_loop(max_maps=3) assert res["status"] == "EXHAUSTED" assert res["total_maps_requested"] == 3 assert res["maps_completed"] == 1 assert res["exhausted"] is True assert len(res["runs"]) == 2 assert coordinator.execute_pipeline.call_count == 2 def test_stage0_stash_routine_success(): coordinator = AutonomousMappingCoordinator(is_simulated=True) coordinator.ensure_focus = MagicMock(return_value=True) coordinator.stash_orchestrator.run_full_stash_cycle = MagicMock(return_value={ "success": True, "stash_opened": True, "tabs_detected": 4, "items_deposited": 5, "waystone_withdrawn": False, "stash_closed": True, "error": "", }) res = coordinator.execute_stage0_stash_routine() assert res["status"] == "SUCCESS" assert res["items_deposited"] == 5 assert res["tabs_detected"] == 4 coordinator.stash_orchestrator.run_full_stash_cycle.assert_called_once() def test_execute_pipeline_with_stashing(): coordinator = AutonomousMappingCoordinator(is_simulated=True) coordinator.ensure_focus = MagicMock(return_value=True) coordinator.dismiss_modals = MagicMock() coordinator.check_active_portal_in_hideout = MagicMock(return_value=None) coordinator.execute_stage0_stash_routine = MagicMock(return_value={ "status": "SUCCESS", "items_deposited": 3, }) coordinator.execute_stage1_precheck = MagicMock(return_value={ "status": "SUCCESS", "selected_slot": (0, 0, 1731, 822, 65.5), }) coordinator.execute_stage2_map_device = MagicMock(return_value={"status": "SUCCESS"}) coordinator.execute_stage3_atlas_activation = MagicMock(return_value={"status": "SUCCESS"}) coordinator.execute_stage4_map_entrance = MagicMock(return_value={"status": "SUCCESS", "entered_zone": "The Western Enigma"}) coordinator.execute_stage5_map_clear = MagicMock(return_value={"status": "SUCCESS", "runner_instance": None, "runner_result": {}}) coordinator.execute_stage6_portal_return = MagicMock(return_value={"status": "SUCCESS"}) report = coordinator.execute_pipeline(target_slot_idx=0, enable_stashing=True, enable_post_stashing=False) assert report["status"] == "SUCCESS" assert "STAGE0_STASH" in report["stages_completed"] assert "STAGE1_PRECHECK" in report["stages_completed"] coordinator.execute_stage0_stash_routine.assert_called_once() def test_check_active_portal_rejects_burning_monolith_and_forbidden_nodes(): """INV-ATLAS-FARM-NODE: Cổng portal cũ của The Burning Monolith hoặc The Well of Souls phải bị từ chối 100%.""" coordinator = AutonomousMappingCoordinator(is_simulated=False) coordinator.capture_step = MagicMock(return_value="captures/fake_hideout.png") with patch("os.path.exists", return_value=True), \ patch("src.common.find_portal_label_by_ocr", return_value=[ (700, 450, "THE BURNING MONOLITH"), (800, 500, "THE WELL OF SOULS"), ]): res = coordinator.check_active_portal_in_hideout() assert res is None, "Phải từ chối toàn bộ portal thuộc Special Arenas (Monolith/Well of Souls)!" def test_check_active_portal_accepts_regular_farming_map(): """Portal của bản đồ cày thông thường hợp lệ (Deserted) được chấp nhận.""" coordinator = AutonomousMappingCoordinator(is_simulated=False) coordinator.capture_step = MagicMock(return_value="captures/fake_hideout.png") with patch("os.path.exists", return_value=True), \ patch("src.common.find_portal_label_by_ocr", return_value=[ (700, 450, "THE BURNING MONOLITH"), (800, 500, "DESERTED"), ]): res = coordinator.check_active_portal_in_hideout() assert res is not None assert res[2] == "DESERTED" def test_stage2_rejects_forbidden_portal_at_map_device(): """Stage 2: Nếu portal tại Map Device là The Burning Monolith, không được coi là portal_already_open.""" coordinator = AutonomousMappingCoordinator(is_simulated=False) coordinator.ensure_focus = MagicMock(return_value=True) coordinator._core_owns_map_device_loop = MagicMock(return_value=False) coordinator.get_player_world_pos = MagicMock(return_value=(100.0, 100.0)) coordinator.find_map_device_world_pos = MagicMock(return_value=(105.0, 105.0)) coordinator.find_map_device_screen_pos = MagicMock(return_value=(1175, 645)) coordinator.click = MagicMock() coordinator.dismiss_modals = MagicMock() coordinator.capture_step = MagicMock(return_value="captures/fake_device.png") coordinator.is_atlas_ui_open = MagicMock(return_value=False) with patch("src.common.find_portal_label_by_ocr", return_value=[(700, 450, "BURNING MONOLITH")]): res = coordinator.execute_stage2_map_device() assert not res.get("portal_already_open"), "Không được chấp thuận portal_already_open cho Burning Monolith!" def test_stage3_refuses_direct_traverse_on_forbidden_node(): """Stage 3: Dù có nút TRAVERSE nhưng text node là THE BURNING MONOLITH thì tuyệt đối CẤM bấm TRAVERSE.""" coordinator = AutonomousMappingCoordinator(is_simulated=False) coordinator.ensure_focus = MagicMock(return_value=True) coordinator.dismiss_modals = MagicMock() coordinator.find_reachable_atlas_node = MagicMock(return_value=(1000, 600)) coordinator.find_active_glowing_atlas_node = MagicMock(return_value=None) coordinator.capture_step = MagicMock(return_value="captures/fake_node.png") coordinator.click = MagicMock() coordinator.ctrl_click = MagicMock() coordinator.press_key = MagicMock() coordinator.is_inventory_open = MagicMock(return_value=True) # Giả lập nút TRAVERSE có sẵn coordinator.find_traverse_button = MagicMock(return_value=(1200, 800)) # Giả lập OCR trả về "THE BURNING MONOLITH" fake_ocr_res = {"lines": [{"text": "THE BURNING MONOLITH", "words": []}]} with patch("os.path.exists", return_value=True), \ patch("PIL.Image.open", return_value=MagicMock()), \ patch("winocr.recognize_pil_sync", return_value=fake_ocr_res): res = coordinator.execute_stage3_atlas_activation(waystone_slot=(0, 0, 1731, 822, 65.5)) # Kiểm tra xem bot có chuyển sang click node cày thông thường (1075, 779) không regular_clicks = [c for c in coordinator.click.call_args_list if c[0] == (int(1075 * coordinator.scale_x), int(779 * coordinator.scale_y))] assert len(regular_clicks) == 1, "Phải click chọn sang node cày thông thường thay vì direct-traverse vào Monolith!" # Bắt buộc nạp Waystone bằng ctrl_click vào node thường thay vì direct traverse sớm coordinator.ctrl_click.assert_called_once_with(1731, 822, delay_after=0.5)