""" Unit & Integration Test cho Control Center GUI, ScreenCapturer & CoreController """ import json import os import sys if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from src.assistant_tool.control_center import ControlCenterApp from src.assistant_tool.core_controller import CoreController from src.assistant_tool.screen_capturer import ScreenCapturer def test_screen_capturer(): print("[Test 1] ScreenCapturer...") cap = ScreenCapturer(output_dir="tests/test_fixtures_real_captures") mock_telem = {"hp": "100/100", "xyz": "100.0, 100.0, 0.0"} img_path = cap.capture_screenshot("UNIT_TEST_CAPTURE", mock_telem) assert img_path is not None, "Capture failed" assert os.path.exists(img_path), "Image file not found" assert img_path.endswith(".png"), "Image should be PNG" json_path = img_path.replace(".png", ".json") assert os.path.exists(json_path), "Metadata JSON not found" with open(json_path, "r", encoding="utf-8") as f: meta = json.load(f) assert meta["event"] == "UNIT_TEST_CAPTURE" assert meta["telemetry"]["hp"] == "100/100" print(f" -> Capture saved: {img_path} + {json_path} OK") # Clean up test capture files to prevent test folder bloating try: if os.path.exists(img_path): os.remove(img_path) if os.path.exists(json_path): os.remove(json_path) except Exception: pass def test_core_controller_cmd_generation(): print("[Test 2] CoreController...") cc = CoreController() cfg = { "auto_dodge": True, "burst_hp": 25, "smart_flask": True, "flask_hp": 50, "kiting": True, "emergency_logout": True, "auto_loot": True, "loot_radius": 35.0, "auto_combo": True, "auto_quest": True, "move_mode": "mouse" } assert len(cfg) > 0 # Check that is_running is False initially assert not cc.is_running() print(" -> CoreController initial state OK") def test_control_center_gui_lifecycle(): print("[Test 3] ControlCenterApp GUI Lifecycle...") app = ControlCenterApp() app.update() # Check move_mode auto-detection and radio buttons assert hasattr(app, "rb_move_mouse") assert hasattr(app, "rb_move_wasd") assert hasattr(app, "chk_auto_detect_move") assert hasattr(app, "lbl_move_badge") assert app.var_move_mode.get() in ("wasd", "mouse"), "Move mode must be either wasd or mouse" app.var_move_mode.set("mouse") assert app.var_move_mode.get() == "mouse" app.var_move_mode.set("wasd") assert app.var_move_mode.get() == "wasd" # Check button texts assert "BẮT ĐẦU" in app.btn_toggle_bot.cget("text") assert "Định Vị" in app.btn_recalibrate.cget("text") assert "Panic" in app.btn_panic.cget("text") # Check sliders & checkboxes assert app.chk_dodge.get() == 1 assert app.chk_flask.get() == 1 assert app.chk_loot.get() == 1 assert app.chk_combo.get() == 1 assert app.chk_quest.get() == 1 # Verify Act 3 Quest Selection in Dropdowns act_values = app.cmb_quest_act.cget("values") assert any("Act 3" in v for v in act_values), "Act 3 must be available in cmb_quest_act" app._on_quest_act_changed("Act 3 (The Sandswept Marsh & Phrey Jungle)") act3_zones = app.cmb_quest_zone.cget("values") assert "Sandswept Marsh (Waypoint)" in act3_zones assert "Phrey Jungle (Boss / Transition)" in act3_zones assert app.var_quest_zone.get() == "Sandswept Marsh (Waypoint)" # Simulate slider movements app.sld_burst.set(28) app._on_burst_slider_change(28) assert "28%" in app.lbl_burst.cget("text") app.sld_flask_hp.set(60) app._on_flask_slider_change(60) assert "60%" in app.lbl_flask_hp.cget("text") # Simulate log insertion app._append_log("Test Log Line 123") app.update() assert "Test Log Line 123" in app.log_textbox.get("1.0", "end") # Stop core_bridge worker thread to avoid async overwrite race condition during test assertions app.core_bridge.stop() # Check Radar telemetry snapshot update from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry snap = CoreTelemetrySnapshot( is_online=True, player=PlayerTelemetry(current_hp=179, max_hp=181, current_mana=144, max_mana=144), entity_count=12, monsters_count=7, items_count=4, portals_count=1, area_name="Cemetery of the Eternals", area_level=7 ) app._on_telemetry_snapshot(snap) app.update() radar_str = app.lbl_radar.cget("text") assert "7 Quái" in radar_str and "4 Đồ rơi" in radar_str and "1 Cổng" in radar_str, f"Radar string invalid: {radar_str}" assert "179 / 181" in app.lbl_hp.cget("text") assert "144 / 144" in app.lbl_mana.cget("text") # Verify TASK-OPT-01: _build_telemetry_dict uses self.core_bridge without AttributeError app.core_bridge.last_snapshot = snap telem_dict = app._build_telemetry_dict("MANUAL_ANALYSIS") assert telem_dict["current_hp"] == 179, f"Expected 179, got {telem_dict.get('current_hp')}" assert telem_dict["max_hp"] == 181, f"Expected 181, got {telem_dict.get('max_hp')}" assert telem_dict["current_mana"] == 144, f"Expected 144, got {telem_dict.get('current_mana')}" # Verify Item Triage & WhatIsBetter integration assert hasattr(app, "btn_test_appraisal") assert hasattr(app, "btn_test_triage") assert hasattr(app, "lbl_triage_status") assert hasattr(app, "lbl_item_eval") # Simulate button clicks app._on_test_item_appraisal_clicked() app._on_test_item_triage_clicked() app.update() log_content = app.log_textbox.get("1.0", "end") assert "WhatIsBetter" in log_content assert "ItemTriage" in log_content # Close safely app.on_closing() print(" -> ControlCenterApp GUI controls, sliders, logging, and Radar telemetry OK") def test_controller_anti_flicker_and_lifecycle_cooldowns(): """Kiểm tra cờ _starting, chống taskkill khi đang start, cooldown 15s mở client và 5s start core.""" from unittest.mock import MagicMock, patch from src.assistant_tool.autonomous_lifecycle import AutonomousLifecycleManager # 1. Kiểm tra CoreController starting guard cc = CoreController() assert hasattr(cc, "_starting") assert not cc._starting cc._starting = True assert cc.start({}) is True # Trả về True ngay lập tức chống trùng lặp cc._starting = False # 2. Kiểm tra _kill_orphan_core_processes không taskkill nếu đang starting hoặc running with patch("subprocess.run") as mock_run: with patch.object(cc, "is_running", return_value=True): cc._kill_orphan_core_processes() mock_run.assert_not_called() with patch.object(cc, "is_running", return_value=False): cc._starting = True cc._kill_orphan_core_processes() mock_run.assert_not_called() # 3. Kiểm tra AutonomousLifecycleManager cooldowns mock_ctrl = MagicMock() mock_ctrl.is_running.return_value = False alm = AutonomousLifecycleManager(core_controller=mock_ctrl, config={"auto_launch_client": True}) assert hasattr(alm, "last_launch_attempt") assert hasattr(alm, "last_core_start_attempt") # Cooldown mở game client 15s with patch("src.assistant_tool.client_launcher.Poe2ClientLauncher.is_client_running", return_value=False), \ patch("src.assistant_tool.client_launcher.Poe2ClientLauncher.launch_client"): alm._evaluate_and_act() t1 = alm.last_launch_attempt assert t1 > 0 alm._evaluate_and_act() assert alm.last_launch_attempt == t1 # Không bị gọi dồn dập # Cooldown gọi core_controller.start 5s with patch("src.assistant_tool.client_launcher.Poe2ClientLauncher.is_client_running", return_value=True): alm.last_core_start_attempt = 0.0 alm._evaluate_and_act() t2 = alm.last_core_start_attempt assert t2 > 0 assert mock_ctrl.start.call_count == 1 alm._evaluate_and_act() assert alm.last_core_start_attempt == t2 # Bị chặn bởi cooldown 5s assert mock_ctrl.start.call_count == 1 print(" -> Anti-flicker starting guard, kill guard, and lifecycle cooldowns OK") def test_chaos_inoculation_ci_support(): """Kiểm tra toàn diện hỗ trợ Chaos Inoculation (CI Keystone): 1. CoreController: parse 'ci', 'chaos inoculation', '1' -> max_hp = 1 và truyền cờ --ci, --max-hp 1. 2. ControlCenter: _get_expected_max_hp() trả về 1 khi nhập 'ci' hoặc 'chaos inoculation'. 3. ControlCenter: is_valid_vitals() chấp nhận max_hp == 1. 4. ControlCenter: _on_telemetry_snapshot() hiển thị '❤️ HP: {cur_hp} / {max_hp} (CI)' khi VERIFIED. 5. OpticalHPSensor: chấp nhận m == 1 trong detect_vitals_from_screen OCR parsing. """ from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry from src.assistant_tool.optical_hp_sensor import OpticalHPSensor, OpticalVitals # 1. CoreController CLI args generation cc = CoreController() cmd_ci1 = cc.build_cmd({"expected_max_hp": "ci"}) assert "--ci" in cmd_ci1, "Must contain --ci flag" assert "--max-hp" in cmd_ci1, "Must contain --max-hp flag" idx1 = cmd_ci1.index("--max-hp") assert cmd_ci1[idx1 + 1] == "1", "max-hp value must be '1'" cmd_ci2 = cc.build_cmd({"expected_max_hp": "chaos inoculation"}) assert "--ci" in cmd_ci2 assert cmd_ci2[cmd_ci2.index("--max-hp") + 1] == "1" cmd_ci3 = cc.build_cmd({"expected_max_hp": "1"}) assert "--ci" in cmd_ci3 assert cmd_ci3[cmd_ci3.index("--max-hp") + 1] == "1" # 2. ControlCenter: is_valid_vitals() assert ControlCenterApp.is_valid_vitals(1, 1) is True, "CI 1/1 must be valid" assert ControlCenterApp.is_valid_vitals(1, 0) is True, "CI 0/1 (dead) must be valid" assert ControlCenterApp.is_valid_vitals(1, 2) is False, "cur > max must be invalid" assert ControlCenterApp.is_valid_vitals(2, 2) is False, "< 50 and != 1 must be invalid" assert ControlCenterApp.is_valid_vitals(49, 49) is False, "49 must be invalid" assert ControlCenterApp.is_valid_vitals(50, 50) is True, "50 must be valid" assert ControlCenterApp.is_valid_vitals(778, 778) is False, "UI marker 778 must be invalid" # 3. ControlCenter: _get_expected_max_hp() & UI Label app = ControlCenterApp() app.update() # Input 'ci' app.entry_max_hp.delete(0, "end") app.entry_max_hp.insert(0, "ci") assert app._get_expected_max_hp() == 1 # Input 'CI' (case-insensitive) app.entry_max_hp.delete(0, "end") app.entry_max_hp.insert(0, "CI") assert app._get_expected_max_hp() == 1 # Input 'chaos inoculation' app.entry_max_hp.delete(0, "end") app.entry_max_hp.insert(0, "Chaos Inoculation") assert app._get_expected_max_hp() == 1 # Input 'Auto' app.entry_max_hp.delete(0, "end") app.entry_max_hp.insert(0, "Auto") assert app._get_expected_max_hp() == 0 # Test verified CI HUD text app.entry_max_hp.delete(0, "end") app.entry_max_hp.insert(0, "ci") snap_ci = CoreTelemetrySnapshot( is_online=True, player=PlayerTelemetry(current_hp=1, max_hp=1, current_mana=100, max_mana=100), entity_count=0, monsters_count=0, items_count=0, portals_count=0, area_name="CI Hideout", area_level=1 ) app._on_telemetry_snapshot(snap_ci) app.update() assert app.lbl_hp.cget("text") == "❤️ HP: 1 / 1 (CI)", f"Got: {app.lbl_hp.cget('text')}" app.on_closing() # 4. OpticalHPSensor: OCR parsing logic allows m == 1 txt_l = "Life 1/1\nEnergy Shield 4500/4500" pairs_l = [("1", "1"), ("4500", "4500")] v = OpticalVitals() life_tuple = (int(pairs_l[0][0]), int(pairs_l[0][1])) cur, m = life_tuple if (50 <= m <= 30000) or m == 1: v.current_hp = cur v.max_hp = m v.hp_ratio = cur / max(1, m) v.is_detected = True assert v.is_detected is True assert v.current_hp == 1 assert v.max_hp == 1 assert v.hp_ratio == 1.0 print(" -> Chaos Inoculation (CI) support in CoreController, ControlCenter & OpticalHPSensor OK") def test_character_state_and_telemetry_ui_display(): """Kiểm tra phân tích Realtime Character State, Map, Quái vật & Kế hoạch di chuyển trên UI.""" from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry # 1. Kiểm tra tĩnh parse_character_state với từng bitmask icon, text, color = ControlCenterApp.parse_character_state(1) assert icon == "🏃" and text == "ĐANG DI CHUYỂN" icon, text, color = ControlCenterApp.parse_character_state(2) assert icon == "🌀" and text == "LƯỚT NÉ" icon, text, color = ControlCenterApp.parse_character_state(4) assert icon == "🧘" and text == "ĐỨNG YÊN" icon, text, color = ControlCenterApp.parse_character_state(8) assert icon == "💀" and text == "TỬ TRẬN" icon, text, color = ControlCenterApp.parse_character_state(16) assert icon == "🛡️" and "GRACE PERIOD" in text # 2. Kiểm tra cập nhật UI qua _on_telemetry_snapshot app = ControlCenterApp() app.update() snap = CoreTelemetrySnapshot( is_online=True, player=PlayerTelemetry( current_hp=500, max_hp=503, current_mana=344, max_mana=344, movement_flags=1 ), entity_count=10, monsters_count=5, items_count=2, portals_count=1, area_name="The Mud Flats", area_level=12, monsters_alive_total=24, entities_monster_count=5 ) app._on_telemetry_snapshot(snap) app.update() # Kiểm tra nhãn Trạng thái nhân vật assert hasattr(app, "lbl_char_state") assert app.lbl_char_state.cget("text") == "🏃 Trạng thái: [ĐANG DI CHUYỂN]" # Kiểm tra nhãn Map assert hasattr(app, "lbl_area") assert app.lbl_area.cget("text") == "Khu vực: The Mud Flats (Lv 12)" # Kiểm tra nhãn đếm quái vật assert hasattr(app, "lbl_monsters") assert app.lbl_monsters.cget("text") == "👾 Quái vật: 5 quái gần (24 tổng)" # Kiểm tra nhãn kế hoạch di chuyển assert hasattr(app, "lbl_movement_plan") plan_text = app.lbl_movement_plan.cget("text") assert "Kế hoạch:" in plan_text # Thử đổi trạng thái sang LƯỚT NÉ (bit 1 = 2) snap.player.movement_flags = 2 app._on_telemetry_snapshot(snap) app.update() assert app.lbl_char_state.cget("text") == "🌀 Trạng thái: [LƯỚT NÉ]" # Thử đổi trạng thái sang TỬ TRẬN (bit 3 = 8) snap.player.movement_flags = 8 app._on_telemetry_snapshot(snap) app.update() assert app.lbl_char_state.cget("text") == "💀 Trạng thái: [TỬ TRẬN]" # Thử đổi trạng thái sang GRACE PERIOD (bit 4 = 16) snap.player.movement_flags = 16 app._on_telemetry_snapshot(snap) app.update() assert app.lbl_char_state.cget("text") == "🛡️ Trạng thái: [GRACE PERIOD (BẤT TỬ ĐẦU MAP)]" app.on_closing() print(" -> Character State & Telemetry UI display OK") def test_proactive_auto_calibration(): """Kiểm tra cơ chế Chủ động tự điều chỉnh Vitals (Proactive Auto-Calibration): - Chu kỳ kiểm tra 1.5s - 2.0s. - Gửi opcode 98 send_recalibrate_vitals khi shm_max_hp == 0 hoặc desync. - Tự động điền optical_max_hp (hoặc 'CI' nếu max_hp == 1) vào entry_max_hp. - Ghi log Proactive Auto-Detect. """ from unittest.mock import MagicMock from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry from src.assistant_tool.optical_hp_sensor import OpticalVitals app = ControlCenterApp() app.update() # Mock send_recalibrate_vitals trên core_bridge app.core_bridge.send_recalibrate_vitals = MagicMock(return_value=True) # 1. Trường hợp shm_max_hp == 0 (Core chưa khóa được máu người chơi) app.core_bridge.last_snapshot = CoreTelemetrySnapshot( is_online=True, player=PlayerTelemetry(current_hp=0, max_hp=0, current_mana=0, max_mana=0) ) app.optical_sensor.last_detected_vitals = OpticalVitals( current_hp=500, max_hp=503, max_mana=344, max_shield=100, is_detected=True ) app._last_optical_crosscheck = 0.0 app._periodic_task() app.update() # Xác minh gửi recalibrate với thông số từ optical sensor app.core_bridge.send_recalibrate_vitals.assert_called_with(503, 344, 100) assert app.entry_max_hp.get() == "503" log_content = app.log_textbox.get("1.0", "end") assert "[Proactive Auto-Detect]" in log_content assert "HP=503, Mana=344, ES=100" in log_content # 2. Trường hợp Chaos Inoculation (max_hp == 1) app.core_bridge.send_recalibrate_vitals.reset_mock() app.core_bridge.last_snapshot = CoreTelemetrySnapshot( is_online=True, player=PlayerTelemetry(current_hp=0, max_hp=0, current_mana=0, max_mana=0) ) app.optical_sensor.last_detected_vitals = OpticalVitals( current_hp=1, max_hp=1, max_mana=200, max_shield=3000, is_detected=True ) app._last_optical_crosscheck = 0.0 app._periodic_task() app.update() app.core_bridge.send_recalibrate_vitals.assert_called_with(1, 200, 3000) assert app.entry_max_hp.get() == "CI" app.on_closing() print(" -> Proactive Auto-Calibration test OK") def test_ward_display_and_mod_filter_gui_integration(): """Kiểm tra tích hợp hiển thị Ward, Mod Filter Banner và phương thức recalibrate_vitals.""" from unittest.mock import MagicMock from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry app = ControlCenterApp() try: app.update() # 1. Kiểm tra tồn tại thành phần Mod Filter trên UI assert hasattr(app, "mod_filter_card") assert hasattr(app, "lbl_mod_filter_title") assert hasattr(app, "lbl_mod_safety_badge") assert hasattr(app, "lbl_mod_report") assert hasattr(app, "btn_eval_mods") # 2. Kiểm tra đánh giá Mod qua button / method # Test mod Brick (Reflect) brick_text = "Monsters reflect 18% of Physical Damage\n40% increased Monster Pack Size" r_brick = app._on_evaluate_map_mods_clicked(sample_text=brick_text) assert r_brick.is_brick is True assert "BRICK" in app.lbl_mod_safety_badge.cget("text") # Test mod An toàn (Reward) safe_text = "80% increased Quantity of Items found in this Area\n25% increased Monster Pack Size" r_safe = app._on_evaluate_map_mods_clicked(sample_text=safe_text) assert r_safe.is_brick is False assert "AN TOÀN" in app.lbl_mod_safety_badge.cget("text") # 3. Kiểm tra nhãn hiển thị ES & Ward: 🛡️ ES: {cur_es}/{max_es} | 🎯 Ward: {cur_ward}/{max_ward} assert hasattr(app, "lbl_vitals_es_ward") snap = CoreTelemetrySnapshot( is_online=True, player=PlayerTelemetry( current_hp=1, max_hp=1, current_es=2843, max_es=2843, current_mana=350, max_mana=350, current_ward=826, max_ward=826 ) ) app._on_telemetry_snapshot(snap) app.update() es_ward_text = app.lbl_vitals_es_ward.cget("text") assert "2843/2843" in es_ward_text and "826/826" in es_ward_text assert "🛡️ ES:" in es_ward_text and "🎯 Ward:" in es_ward_text # 4. Kiểm tra phương thức recalibrate_vitals gửi đầy đủ 3 tham số (max_hp, max_mana, max_es) app.core_bridge.send_recalibrate_vitals = MagicMock(return_value=True) res = app.recalibrate_vitals(max_hp=1, max_mana=350, max_es=2843) assert res is True app.core_bridge.send_recalibrate_vitals.assert_called_once_with(1, 350, 2843) finally: app.on_closing() print(" -> Ward display, Mod Filter GUI, and recalibrate_vitals test OK") def test_hud_area_contract_and_world_map_gate(): """HUD khu vực = SHM playable hoặc seed Client.txt; Act 2 không ghi đè; U chỉ khi làng đã xác thực.""" from unittest.mock import MagicMock from src.assistant_tool.config import Config from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry app = ControlCenterApp() app.update() try: geo = app.geometry().split("+")[0] w, h = (int(x) for x in geo.split("x")[:2]) assert w >= Config.CONTROL_CENTER_MIN_WIDTH assert h >= Config.CONTROL_CENTER_MIN_HEIGHT from src.assistant_tool import ui_theme as theme for key in theme.REQUIRED_THEME_KEYS: assert key in Config.THEME assert "#10b981" in str(app.btn_toggle_bot.cget("fg_color")) app.lifecycle_manager.current_area_name = "Hive Fortress" app.lifecycle_manager.current_area_code = "ChayulaLeague_Tower" app.lifecycle_manager.has_real_map_log = True app.lifecycle_manager.is_in_town = False snap = CoreTelemetrySnapshot( is_online=True, player=PlayerTelemetry(current_hp=1, max_hp=1, current_mana=96, max_mana=97), area_name="", area_level=77, ) app._on_telemetry_snapshot(snap) app.update() assert "Hive Fortress" in app.lbl_area.cget("text") assert "💧 Mana: 96 / 97" in app.lbl_mana.cget("text") snap.area_name = "Act 2" app._on_telemetry_snapshot(snap) app.update() assert "Hive Fortress" in app.lbl_area.cget("text") assert "Act 2" not in app.lbl_area.cget("text") app.core_bridge.send_trigger_world_map_travel = MagicMock(return_value=True) app._on_trigger_quest_travel_clicked() assert not app.core_bridge.send_trigger_world_map_travel.called app.lifecycle_manager.is_in_town = True app.lifecycle_manager.current_area_name = "Shoreline Hideout" app._on_trigger_quest_travel_clicked() assert app.core_bridge.send_trigger_world_map_travel.called finally: app.on_closing() print(" -> HUD area contract + World Map U gate OK") if __name__ == "__main__": test_screen_capturer() test_core_controller_cmd_generation() test_control_center_gui_lifecycle() test_controller_anti_flicker_and_lifecycle_cooldowns() test_chaos_inoculation_ci_support() test_character_state_and_telemetry_ui_display() test_proactive_auto_calibration() test_ward_display_and_mod_filter_gui_integration() test_hud_area_contract_and_world_map_gate() print("\n✅ TẤT CẢ TEST CONTROL CENTER ĐỀU VƯỢT QUA 100%!")