"""AutoPOE2 - Unit Tests for Frontier Exploration Engine & Monster Density Heatmap. ================================================================================== Kiểm thử tích hợp chuyên sâu cho Stage 5 In-Map Autonomous Exploration: 1. `MonsterDensityHeatmap`: Phân bố không gian 2D, gom cụm bãi quái (pack clustering), đếm kills, thẩm định bãi quái hợp lệ `is_clearing_legitimate(min_required_packs=4)`. 2. `FrontierExplorationEngine`: Điều phối 3 chế độ (FRONTIER_EXPLORE, COMBAT_ENGAGE, BOSS_COMBAT), tích hợp BossArenaDetector và MinimapNavigationAdapter. 3. Rào chắn CẤM THOÁT SỚM (Anti-Premature-Exit Interlock): Kháng triệt để lỗ hổng "đánh 1 con quái ở cửa map xong đứng yên 25s kết luận map clear". 4. Kiểm thử hiệu năng trên frame thực tế (captures/*.png) < 10.0 ms. Tuân thủ nghiêm ngặt: - Rule 1: Python 3.11 Cold Path Tier 2. - Rule 4: Module hóa độc lập trần < 500 dòng. - Rule 5: 100% test case kiểm thử thực nghiệm trên artifact thật. - Bất biến: INV-MAP-LEGITIMATE-CLEAR, INV-WASD-NO-BLIND-LMB. """ from __future__ import annotations import os import time import numpy as np import pytest from src.assistant_tool.navigation.frontier_exploration_engine import ( FrontierExplorationEngine, InMapOperationMode, ExplorationStepDecision, ) from src.assistant_tool.navigation.map_clear_evaluator import MapClearEvaluator from src.assistant_tool.navigation.monster_density_heatmap import ( MonsterDensityHeatmap, MonsterPackEncounter, ) from src.assistant_tool.vision.boss_arena_detector import ( ArenaGateDetection, BossArenaDetector, BossArenaState, BossHealthBarDetection, ) REAL_CAPTURE_PATH = r"c:\Projects\AutoPOE2\captures\20260910_054356_319_PERIODIC_ANALYSIS.png" class TestMonsterDensityHeatmap: """Kiểm thử tính năng quản trị bản đồ nhiệt mật độ quái vật 2D.""" def test_spatial_sectors_and_pack_grouping(self): """Kiểm thử phân ô sector 100u x 100u và gom cụm bãi quái trong bán kính 80u.""" heatmap = MonsterDensityHeatmap(sector_size=100.0, min_pack_monsters=2, pack_merge_radius=80.0) # Quái vật ở sector (1, 1): Tọa độ (120, 120) heatmap.record_telemetry_frame(player_x=120.0, player_y=120.0, monsters_near=5, now_sec=1.0) assert heatmap.total_packs_count == 1 assert heatmap.visited_sector_count == 1 assert (1, 1) in heatmap.visited_sectors # Bước tiếp theo ở (130, 125): Cách vị trí cũ ~11.18u (< 80u) -> Gộp vào bãi cũ heatmap.record_telemetry_frame(player_x=130.0, player_y=125.0, monsters_near=4, now_sec=2.0) assert heatmap.total_packs_count == 1 assert heatmap.packs[0].current_count == 4 # Di chuyển sang sector (3, 3): Tọa độ (350, 350) -> Cách > 300u -> Tạo bãi quái số 2 heatmap.record_telemetry_frame(player_x=350.0, player_y=350.0, monsters_near=6, now_sec=3.0) assert heatmap.total_packs_count == 2 assert heatmap.visited_sector_count == 2 assert (3, 3) in heatmap.visited_sectors def test_kill_tracking_and_pack_cleared(self): """Kiểm thử tính toán số quái vật bị tiêu diệt và đánh dấu bãi dọn sạch.""" heatmap = MonsterDensityHeatmap(min_pack_monsters=2, pack_merge_radius=80.0) # Gặp bãi quái 8 con ở (100, 100) heatmap.record_telemetry_frame(player_x=100.0, player_y=100.0, monsters_near=8, now_sec=1.0) assert heatmap.total_monsters_encountered == 8 assert heatmap.total_monsters_eliminated == 0 # Sau đòn đánh, còn 3 con sống heatmap.record_telemetry_frame(player_x=100.0, player_y=100.0, monsters_near=3, now_sec=2.0) assert heatmap.total_monsters_eliminated == 5 # 8 - 3 = 5 # Dọn sạch bãi: monsters_near = 0 heatmap.record_telemetry_frame(player_x=100.0, player_y=100.0, monsters_near=0, now_sec=3.0) assert heatmap.total_monsters_eliminated == 8 # 5 + 3 = 8 assert heatmap.cleared_packs_count == 1 assert heatmap.packs[0].is_cleared is True def test_is_clearing_legitimate_thresholds(self): """Kiểm thử logic thẩm định tính hợp lệ của quá trình dọn quái.""" heatmap = MonsterDensityHeatmap(pack_merge_radius=80.0) # Chưa gặp bãi quái nào -> Từ chối is_legit, reason = heatmap.is_clearing_legitimate(min_required_packs=4) assert is_legit is False assert "0/4 bãi" in reason # Gặp 1 bãi quái và diệt 3 con -> Vẫn từ chối (chưa đủ 4 bãi hoặc 25 kills) heatmap.record_telemetry_frame(player_x=100.0, player_y=100.0, monsters_near=3, now_sec=1.0) heatmap.record_telemetry_frame(player_x=100.0, player_y=100.0, monsters_near=0, now_sec=2.0) is_legit, reason = heatmap.is_clearing_legitimate(min_required_packs=4) assert is_legit is False assert "1/4 bãi" in reason # Trường hợp A: Diệt bãi lớn đạt >= 25 kills -> Hợp lệ dù mới có 1-2 bãi heatmap.total_monsters_eliminated = 26 is_legit, reason = heatmap.is_clearing_legitimate(min_required_packs=4) assert is_legit is True assert ">= 25 quái" in reason # Trường hợp B: Đạt >= 4 bãi quái độc lập heatmap.total_monsters_eliminated = 15 coords = [(100, 100), (300, 100), (500, 100), (700, 100)] for i, (cx, cy) in enumerate(coords): heatmap.record_telemetry_frame(player_x=float(cx), player_y=float(cy), monsters_near=4, now_sec=10.0 + i) heatmap.record_telemetry_frame(player_x=float(cx), player_y=float(cy), monsters_near=0, now_sec=10.5 + i) assert heatmap.total_packs_count >= 4 is_legit, reason = heatmap.is_clearing_legitimate(min_required_packs=4) assert is_legit is True assert "Đã dọn sạch" in reason def test_optical_boss_status_tracking(self): """Kiểm thử tích hợp dữ liệu thị giác về Boss bản đồ.""" heatmap = MonsterDensityHeatmap() # Phát hiện thanh máu Boss 100% heatmap.update_optical_boss_status(boss_bar_detected=True, health_ratio=1.0) assert heatmap.boss_detected is True assert heatmap.boss_slain_optical is False # Boss bị đánh tụt máu xuống 10% heatmap.update_optical_boss_status(boss_bar_detected=True, health_ratio=0.10) assert heatmap.boss_slain_optical is False # Thanh máu Boss tụt về 0.01 -> Hạ gục trực tiếp heatmap.update_optical_boss_status(boss_bar_detected=True, health_ratio=0.01) assert heatmap.boss_slain_optical is True # Test kịch bản thanh máu biến mất sau khi máu xuống thấp (< 0.25) heatmap.reset() heatmap.update_optical_boss_status(boss_bar_detected=True, health_ratio=0.80) heatmap.update_optical_boss_status(boss_bar_detected=True, health_ratio=0.15) assert heatmap.boss_slain_optical is False # Camera không còn thấy thanh máu Boss (boss_bar_detected = False) heatmap.update_optical_boss_status(boss_bar_detected=False) assert heatmap.boss_slain_optical is True class TestFrontierExplorationEngine: """Kiểm thử động cơ điều phối thám hiểm sương mù liên tục FrontierExplorationEngine.""" def test_exploration_mode_transitions_with_monsters(self): """Kiểm thử chuyển trạng thái mượt mà giữa FRONTIER_EXPLORE và COMBAT_ENGAGE.""" engine = FrontierExplorationEngine(target_coverage=0.80, min_required_packs=4) # Tạo khung hình giả định 1080p (toàn đen = sương mù dày) dummy_frame = np.zeros((1080, 1920, 3), dtype=np.uint8) # Nhịp 1: Xung quanh an toàn (monsters_near = 0) -> FRONTIER_EXPLORE decision1 = engine.process_frame_and_decide( frame_input=dummy_frame, player_x=100.0, player_y=100.0, monsters_near=0, ) assert decision1.mode == InMapOperationMode.FRONTIER_EXPLORE assert decision1.packs_encountered == 0 # Nhịp 2: Gặp đàn quái vật (monsters_near = 5) -> Chuyển ngay sang COMBAT_ENGAGE decision2 = engine.process_frame_and_decide( frame_input=dummy_frame, player_x=120.0, player_y=120.0, monsters_near=5, ) assert decision2.mode == InMapOperationMode.COMBAT_ENGAGE assert decision2.packs_encountered == 1 assert "Stutter-Step Combat" in decision2.reason # Nhịp 3: Quái vật đã bị tiêu diệt (monsters_near = 0) -> Trở về FRONTIER_EXPLORE decision3 = engine.process_frame_and_decide( frame_input=dummy_frame, player_x=125.0, player_y=120.0, monsters_near=0, ) assert decision3.mode == InMapOperationMode.FRONTIER_EXPLORE def test_boss_gate_and_boss_combat_modes(self): """Kiểm thử phát hiện cổng sương mù và chuyển chế độ đánh Boss.""" engine = FrontierExplorationEngine(target_coverage=0.80, min_required_packs=4) dummy_frame = np.zeros((1080, 1920, 3), dtype=np.uint8) # Mock BossArenaDetector state: Cổng sương mù xuất hiện engine.boss_detector.process_frame = lambda frame: BossArenaState( gate_detected=True, gate_info=ArenaGateDetection(detected=True, confidence=0.85, bounding_box=(800, 400, 1100, 600)), boss_bar_active=False, boss_bar_info=BossHealthBarDetection(detected=False), ) decision = engine.process_frame_and_decide(dummy_frame, player_x=500.0, player_y=500.0, monsters_near=0) assert decision.mode == InMapOperationMode.BOSS_PRE_BUFF assert "tiền buff" in decision.reason # Mock BossArenaDetector state: Bước vào đấu trường, thanh máu Boss xuất hiện engine.boss_detector.process_frame = lambda frame: BossArenaState( gate_detected=False, gate_info=ArenaGateDetection(detected=False), boss_bar_active=True, boss_bar_info=BossHealthBarDetection(detected=True, health_ratio=0.75, is_phased=False), ) decision_boss = engine.process_frame_and_decide(dummy_frame, player_x=600.0, player_y=600.0, monsters_near=0) assert decision_boss.mode == InMapOperationMode.BOSS_COMBAT assert "Giao tranh với Boss" in decision_boss.reason # Mock Boss chuyển pha bất tử (is_phased = True) -> BOSS_KITE engine.boss_detector.process_frame = lambda frame: BossArenaState( gate_detected=False, gate_info=ArenaGateDetection(detected=False), boss_bar_active=True, boss_bar_info=BossHealthBarDetection(detected=True, health_ratio=0.50, is_phased=True), ) decision_kite = engine.process_frame_and_decide(dummy_frame, player_x=600.0, player_y=600.0, monsters_near=0) assert decision_kite.mode == InMapOperationMode.BOSS_KITE assert "Kiting thả diều né đòn" in decision_kite.reason class TestAntiPrematureExitInterlock: """Kiểm thử rào chắn cấm thoát sớm: Loại bỏ 100% hiện tượng đánh 1 quái rồi kết luận map clear.""" def test_rejects_single_monster_at_entrance_even_with_distance_and_time(self): """Kịch bản thực tế của phản hồi User: Nhân vật đánh 1 bãi quái ở cửa map, đi vòng vòng đạt đủ 1200u, đứng yên 25s, quá 90s: BẮT BUỘC KHÔNG ĐƯỢC KẾT LUẬN MAP CLEAR! """ heatmap = MonsterDensityHeatmap() evaluator = MapClearEvaluator( min_duration_sec=90.0, target_coverage=0.80, min_traverse_units=1200.0, min_required_packs=4, density_heatmap=heatmap, ) engine = FrontierExplorationEngine( target_coverage=0.80, min_required_packs=4, density_heatmap=heatmap, ) dummy_frame = np.zeros((1080, 1920, 3), dtype=np.uint8) # 1. Đánh 1 bãi quái duy nhất ở cửa map (100, 100) lúc t=5s evaluator.update_telemetry(pos_x=100.0, pos_y=100.0, monsters_near=3, coverage_ratio=0.05, now_sec=5.0) engine.process_frame_and_decide(dummy_frame, player_x=100.0, player_y=100.0, monsters_near=3) # 2. Dọn xong bãi quái đó (monsters_near = 0) evaluator.update_telemetry(pos_x=110.0, pos_y=100.0, monsters_near=0, coverage_ratio=0.08, now_sec=10.0) engine.process_frame_and_decide(dummy_frame, player_x=110.0, player_y=100.0, monsters_near=0) # 3. Đi qua đi lại quanh lối vào để tích lũy quãng đường đạt 1300u for step in range(7): evaluator.update_telemetry(pos_x=100.0 + step * 200.0, pos_y=100.0, monsters_near=0, coverage_ratio=0.15, now_sec=15.0 + step * 5) assert evaluator.total_distance_traversed >= 1200.0 # 4. Đứng yên 25s lúc t=95s (> 90s tối thiểu) is_cleared, reason = evaluator.evaluate_clear(elapsed_sec=95.0, now_sec=95.0) # KHẲNG ĐỊNH: Bị từ chối hoàn toàn do chưa mở sương (15% < 80%) và mới đánh 1 bãi quái (< 4 bãi) assert is_cleared is False assert reason == "" # Engine vẫn tiếp tục thâm nhập Frontier sương mù, không bao giờ hoàn tất map decision = engine.process_frame_and_decide(dummy_frame, player_x=1300.0, player_y=100.0, monsters_near=0) assert decision.mode == InMapOperationMode.FRONTIER_EXPLORE assert decision.mode != InMapOperationMode.MAP_COMPLETED def test_legitimate_full_clear_pipeline(self): """Kịch bản hoàn tất bản đồ chuẩn chỉnh: - Tuần tra qua 5 bãi quái khác nhau, tiêu diệt tổng cộng 22 quái vật. - Quãng đường đạt > 1200u. - Độ phủ sương mù Minimap đạt 82% (>= 80%). - Thời gian trong map 115s (>= 90s). -> Cả MapClearEvaluator và FrontierExplorationEngine cùng công nhận hoàn tất map! """ heatmap = MonsterDensityHeatmap(pack_merge_radius=80.0) evaluator = MapClearEvaluator( min_duration_sec=90.0, target_coverage=0.80, min_traverse_units=1200.0, min_required_packs=4, density_heatmap=heatmap, ) engine = FrontierExplorationEngine( target_coverage=0.80, min_required_packs=4, density_heatmap=heatmap, ) dummy_frame = np.zeros((1080, 1920, 3), dtype=np.uint8) # Di chuyển tuần tra qua 5 sector khác nhau và dọn từng bãi quái waypoints = [ (100.0, 100.0), (350.0, 150.0), (600.0, 200.0), (850.0, 300.0), (1100.0, 400.0), (1350.0, 500.0), ] t = 10.0 cov = 0.10 for i, (wx, wy) in enumerate(waypoints): # Đến điểm, phát hiện quái evaluator.update_telemetry(pos_x=wx, pos_y=wy, monsters_near=4, coverage_ratio=cov, now_sec=t) engine.process_frame_and_decide(dummy_frame, player_x=wx, player_y=wy, monsters_near=4) t += 5.0 cov += 0.10 # Tiêu diệt sạch quái ở điểm đó evaluator.update_telemetry(pos_x=wx + 10.0, pos_y=wy + 10.0, monsters_near=0, coverage_ratio=cov, now_sec=t) engine.process_frame_and_decide(dummy_frame, player_x=wx + 10.0, player_y=wy + 10.0, monsters_near=0) t += 10.0 assert evaluator.total_distance_traversed >= 1200.0 assert heatmap.total_packs_count >= 5 assert heatmap.cleared_packs_count >= 4 # Cập nhật độ phủ đạt 83% (> 80%) tại t=115s evaluator.update_telemetry(pos_x=1400.0, pos_y=500.0, monsters_near=0, coverage_ratio=0.83, now_sec=115.0) is_cleared, reason = evaluator.evaluate_clear(elapsed_sec=115.0, now_sec=115.0) assert is_cleared is True assert "Minimap đã mở sương đạt" in reason assert ">= 80%" in reason assert ">= 1200u" in reason # Giả lập adapter báo coverage 83% cho engine engine.nav_adapter.explorer.coverage_ratio = 0.83 engine.fog_explorer.coverage_ratio = 0.83 engine.nav_adapter.process_fullscreen_frame = lambda f: None # Keep coverage_ratio from explorer decision = engine.process_frame_and_decide(dummy_frame, player_x=1400.0, player_y=500.0, monsters_near=0) assert decision.mode == InMapOperationMode.MAP_COMPLETED assert "Đạt độ phủ sương mù" in decision.reason class TestPerformanceOnRealArtifacts: """Kiểm thử thời gian thực thi trên ảnh chụp game thực tế.""" @pytest.mark.skipif(not os.path.exists(REAL_CAPTURE_PATH), reason="Artifact ảnh thực tế không tồn tại") def test_process_frame_latency_under_tier2_budget(self): """Xử lý khung hình thực tế 1080p phải hoàn tất trong Tier 2 budget (< 35ms, tức > 28 FPS).""" from PIL import Image frame_arr = np.array(Image.open(REAL_CAPTURE_PATH)) engine = FrontierExplorationEngine(target_coverage=0.80, min_required_packs=4) # Khởi động bộ nhớ (warmup) engine.process_frame_and_decide(frame_arr, player_x=100.0, player_y=100.0, monsters_near=0) # Đo đạc 10 vòng lặp latencies = [] for _ in range(10): t0 = time.perf_counter() decision = engine.process_frame_and_decide(frame_arr, player_x=150.0, player_y=150.0, monsters_near=0) latencies.append((time.perf_counter() - t0) * 1000.0) avg_latency = float(np.mean(latencies)) print(f"\n[PERF] FrontierExplorationEngine process_frame latency: {avg_latency:.2f} ms") assert avg_latency < 35.0, f"Latency {avg_latency:.2f} ms vượt quá ngân sách Tier 2 35.0 ms!" assert decision.mode in ( InMapOperationMode.FRONTIER_EXPLORE, InMapOperationMode.COMBAT_ENGAGE, InMapOperationMode.BOSS_KITE, InMapOperationMode.BOSS_COMBAT, )