"""Bộ kiểm thử đơn vị cho thư viện dùng chung src/common/ và các bản vá P0 Python. Kiểm tra: 1. src/common/win32_window: attach_to_default_desktop, find_poe2_window, get_client_rect, get_dpi_scale. 2. src/common/win32_process: is_poe2_running, get_poe2_pid, find_process_by_name. 3. src/common/screen_capture: capture_rect_bitblt, capture_window_bitblt, capture_window_print. 4. src/common/audio_feedback: beep_bot_toggle, beep_panic, beep_notification. 5. src/common/item_types: ItemData, ItemRarity, AffixData, AffixType. 6. src/common/logging_util: get_logger, log_exception. 7. CoreShmBridge Telemetry: npcs_count, area_code, area_seed, read_limit 256. 8. Các trường hợp biên win32_window và win32_process trong src/common. """ import io import logging import mmap import os import struct import sys import unittest sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) import src.common as common from src.assistant_tool.core_shm_bridge import ( CoreTelemetrySnapshot, PlayerTelemetry, ) class TestCommonLibrary(unittest.TestCase): """Kiểm chứng hoạt động của toàn bộ các module trong src/common/.""" def test_win32_window_functions(self): # 1. attach_to_default_desktop không được ném ngoại lệ desk = common.attach_to_default_desktop() self.assertTrue(desk is not None or desk is None) # 2. find_poe2_window trả về int hwnd = common.find_poe2_window(["NonExistentWindowTitle12345"]) self.assertIsInstance(hwnd, int) # 3. get_client_rect trả về 4-tuple int rect = common.get_client_rect(hwnd) self.assertEqual(len(rect), 4) for val in rect: self.assertIsInstance(val, int) # 4. get_dpi_scale trả về float >= 1.0 khi hwnd = 0 scale = common.get_dpi_scale(0) self.assertIsInstance(scale, float) self.assertGreaterEqual(scale, 0.5) def test_win32_process_functions(self): # 1. find_process_by_name với tên tiến trình Python hiện tại current_pid = os.getpid() # Tìm python hoặc python.exe pids = common.find_process_by_name("python.exe") if not pids: pids = common.find_process_by_name("python") self.assertIsInstance(pids, list) # 2. is_poe2_running trả về bool running = common.is_poe2_running() self.assertIsInstance(running, bool) # 3. get_poe2_pid trả về int hoặc None poe_pid = common.get_poe2_pid() self.assertTrue(poe_pid is None or isinstance(poe_pid, int)) def test_screen_capture_safe_cleanup(self): # Test capture_rect_bitblt với kích thước không hợp lệ -> trả về None an toàn img_invalid = common.capture_rect_bitblt(0, 0, -10, -10) self.assertIsNone(img_invalid) # Test capture_window_bitblt với hwnd = 0 -> trả về None img_hwnd0 = common.capture_window_bitblt(0) self.assertIsNone(img_hwnd0) # Test capture_window_print với hwnd = 0 -> trả về None img_print0 = common.capture_window_print(0) self.assertIsNone(img_print0) def test_audio_feedback_non_blocking(self): # Kiểm tra gọi hàm không gây crash hoặc block main thread common.beep_bot_toggle(True) common.beep_bot_toggle(False) common.beep_panic() common.beep_notification() def test_item_types(self): # 1. Khởi tạo AffixData affix1 = common.AffixData( text="+75 to maximum Life", affix_type=common.AffixType.PREFIX, tier=1, name="Vigorous", category="life", values=[75.0], ) self.assertEqual(affix1.tier, 1) self.assertEqual(affix1.affix_type, common.AffixType.PREFIX) # 2. Khởi tạo ItemData item = common.ItemData( name="Doom Star", base_type="Expert Vaal Cuirass", item_class="Body Armours", rarity=common.ItemRarity.RARE, item_level=84, quality=20, identified=True, sockets=6, explicits=[affix1], ) self.assertTrue(item.is_rare) self.assertFalse(item.is_normal) self.assertFalse(item.is_unique) self.assertEqual(item.total_affixes_count, 1) def test_logging_util(self): logger = common.get_logger("TestLogger", level=logging.DEBUG) self.assertEqual(logger.level, logging.DEBUG) # Kiểm thử log_exception không crash try: raise ValueError("Test error for logging") except ValueError as err: common.log_exception(logger, "Ngữ cảnh kiểm thử unit test", err) class TestP0PythonFixes(unittest.TestCase): """Kiểm chứng các bản sửa lỗi P0 phía Python.""" def test_core_telemetry_snapshot_area_and_npcs_fields(self): """Kiểm chứng CoreTelemetrySnapshot lưu trữ đầy đủ area_code, area_seed và npcs_count.""" snap = CoreTelemetrySnapshot( is_online=True, snapshot_id=42, timestamp_ms=1000, player=PlayerTelemetry(current_hp=500, max_hp=500), entity_count=100, monsters_count=40, items_count=20, portals_count=2, npcs_count=15, area_name="Omen Ridge", area_level=72, area_code="Act3_Town_01", area_seed=987654321, ) self.assertEqual(snap.npcs_count, 15) self.assertEqual(snap.area_code, "Act3_Town_01") self.assertEqual(snap.area_seed, 987654321) def test_common_win32_window_edge_cases(self): """Kiểm chứng các trường hợp biên của win32_window trong src/common.""" # 1. get_client_rect với hwnd = 0 rect = common.get_client_rect(0) self.assertEqual(rect, (0, 0, 0, 0)) # 2. get_dpi_scale với hwnd = 0 scale = common.get_dpi_scale(0) self.assertGreaterEqual(scale, 0.5) # 3. find_poe2_window với danh sách rỗng hwnd = common.find_poe2_window([]) self.assertIsInstance(hwnd, int) def test_common_win32_process_edge_cases(self): """Kiểm chứng các trường hợp biên của win32_process trong src/common.""" # 1. Tìm tiến trình chắc chắn không tồn tại pids = common.find_process_by_name("non_existent_fake_process_9999.exe") self.assertEqual(pids, []) # 2. get_poe2_pid và is_poe2_running nhất quán running = common.is_poe2_running() pid = common.get_poe2_pid() if running: self.assertIsNotNone(pid) else: self.assertIsNone(pid) def test_common_win32_input(self): """Kiểm chứng các hàm và hằng số Win32 Input trong src/common.""" # 1. Kiểm tra tọa độ chuột pos = common.get_cursor_pos() self.assertIsInstance(pos, tuple) self.assertEqual(len(pos), 2) self.assertIsInstance(pos[0], int) self.assertIsInstance(pos[1], int) # 2. Kiểm tra các hằng số phím chuẩn SSoT self.assertEqual(common.VK_W, 0x57) self.assertEqual(common.VK_A, 0x41) self.assertEqual(common.VK_S, 0x53) self.assertEqual(common.VK_D, 0x44) self.assertEqual(common.VK_SPACE, 0x20) self.assertEqual(common.VK_I, 0x49) self.assertEqual(common.VK_U, 0x55) self.assertEqual(common.VK_OEM_PERIOD, 0xBE) # 3. Kiểm tra tính khả dụng của các hàm điều khiển self.assertTrue(callable(common.click)) self.assertTrue(callable(common.hover)) self.assertTrue(callable(common.ctrl_click)) self.assertTrue(callable(common.press_key)) self.assertTrue(callable(common.dismiss_modals)) self.assertTrue(callable(common.send_wasd_direction)) self.assertTrue(callable(common.mouse_wheel)) def test_common_vision_ocr(self): """Kiểm chứng các hàm thị giác máy tính và WinOCR trong src/common.""" from PIL import Image # 1. Biên None an toàn (Fail-closed) self.assertFalse(common.is_atlas_ui_open(None)) self.assertFalse(common.is_inventory_open(None)) self.assertIsNone(common.find_button_by_text(None, "TRAVERSE")) self.assertEqual(common.find_color_clusters(None, lambda r, g, b: True), []) # 2. Fallback Map Device an toàn khi không có ảnh pos = common.find_map_device_screen_pos(None, fallback_pos=(1175, 645)) self.assertEqual(pos, (1175, 645)) # 2b. Fallback Portal quanh Map Device thật portal_pos_fallback = common.find_portal_screen_pos(None, fallback_pos=(705, 450)) self.assertEqual(portal_pos_fallback, (705, 450)) portal_pos_relative = common.find_portal_screen_pos(None) self.assertEqual(portal_pos_relative, (506, 240)) portal_pos_custom_md = common.find_portal_screen_pos(None, map_device_pos=(518, 572)) self.assertEqual(portal_pos_custom_md, (488, 532)) # 3. Thuật toán phân cụm màu (Color Clustering) trên ảnh PIL nhân tạo test_img = Image.new("RGB", (200, 200), color=(0, 0, 0)) # Tạo 25 điểm màu xanh lá cây quanh tâm (100, 100) for dy in range(-2, 3): for dx in range(-2, 3): test_img.putpixel((100 + dx, 100 + dy), (0, 255, 0)) clusters = common.find_color_clusters( test_img, color_predicate=lambda r, g, b: g > 200 and r < 50 and b < 50, step=1, cluster_radius_x=10, cluster_radius_y=10, min_cluster_size=15, ) self.assertEqual(len(clusters), 1) cx, cy, count = clusters[0] self.assertEqual(cx, 100) self.assertEqual(cy, 100) self.assertEqual(count, 25) def test_background_input_functions(self): # 1. client_to_screen_pos fallback khi hwnd = 0 x, y = common.client_to_screen_pos(0, 150, 250) self.assertEqual(x, 150) self.assertEqual(y, 250) # 2. post_message_click trả về False khi hwnd = 0 res = common.post_message_click(0, 100, 100) self.assertFalse(res) # 3. post_message_ctrl_click trả về False khi hwnd = 0 res = common.post_message_ctrl_click(0, 100, 100) self.assertFalse(res) # 4. post_message_hover trả về False khi hwnd = 0 res = common.post_message_hover(0, 100, 100) self.assertFalse(res) # 5. post_message_key trả về False khi hwnd = 0 res = common.post_message_key(0, 0x20) self.assertFalse(res) # 6. snap_window_to_half trả về False khi hwnd = 0 res = common.snap_window_to_half(0) self.assertFalse(res) def test_fast_path_optical_sensors_and_modal_dismiss(self): """Kiểm thử Fast-Path (< 0.5ms) cho is_inventory_open, find_map_device_screen_pos và dismiss_modals.""" import time from unittest.mock import patch from PIL import Image # 1. is_inventory_open: Fail-closed khi None self.assertFalse(common.is_inventory_open(None)) # 2. is_inventory_open Fast-Path trên artifact thật 1440p (< 1.0ms) f_1440 = "captures/pipeline_run/20260913_215645_161_STAGE1_INVENTORY_OPEN.png" if os.path.exists(f_1440): im_1440 = Image.open(f_1440).convert("RGB") im_1440.load() common.is_inventory_open(im_1440) # Warmup t0 = time.perf_counter() self.assertTrue(common.is_inventory_open(im_1440)) dt_ms = (time.perf_counter() - t0) * 1000.0 self.assertLess(dt_ms, 2.0, f"Fast-Path quá chậm: {dt_ms:.2f}ms") # 3. is_inventory_open Fast-Path trên artifact thật 891p windowed (< 1.0ms) f_891 = "captures/pipeline_run/20260917_043805_828_STAGE1_INVENTORY_OPEN.png" if os.path.exists(f_891): im_891 = Image.open(f_891).convert("RGB") im_891.load() t0 = time.perf_counter() self.assertTrue(common.is_inventory_open(im_891)) dt_ms = (time.perf_counter() - t0) * 1000.0 self.assertLess(dt_ms, 2.0, f"Fast-Path 891p quá chậm: {dt_ms:.2f}ms") # 4. is_inventory_open trả về False khi túi đồ đóng f_closed = "captures/pipeline_run/20260913_215718_063_STAGE2_HIDEOUT_SCAN.png" if os.path.exists(f_closed): self.assertFalse(common.is_inventory_open(f_closed)) # 5. find_map_device_screen_pos với fast_path=True trên bệ đá Shoreline Hideout f_stage2 = "captures/pipeline_run/20260918_152632_757_STAGE2_MAP_DEVICE_VIEW.png" if os.path.exists(f_stage2): im_s2 = Image.open(f_stage2).convert("RGB") im_s2.load() t0 = time.perf_counter() pos = common.find_map_device_screen_pos(im_s2, fast_path=True) dt_ms = (time.perf_counter() - t0) * 1000.0 self.assertEqual(pos, (249, 173)) self.assertLess(dt_ms, 2.0, f"find_map_device_screen_pos fast_path quá chậm: {dt_ms:.2f}ms") # 5b. find_map_device_screen_pos trên artifact 153346_454 (OCR MAP DEVICE) và 161010_809 (bệ đá không nhãn) f_stage2_454 = "captures/pipeline_run/20260918_153346_454_STAGE2_MAP_DEVICE_VIEW.png" if os.path.exists(f_stage2_454): pos_454 = common.find_map_device_screen_pos(f_stage2_454) self.assertEqual(pos_454, (306, 196)) pos_454_fast = common.find_map_device_screen_pos(f_stage2_454, fast_path=True) self.assertEqual(pos_454_fast, (249, 173)) f_stage2_809 = "captures/pipeline_run/20260918_161010_809_STAGE2_MAP_DEVICE_VIEW.png" if os.path.exists(f_stage2_809): pos_809 = common.find_map_device_screen_pos(f_stage2_809) self.assertEqual(pos_809, (249, 173)) pos_809_fast = common.find_map_device_screen_pos(f_stage2_809, fast_path=True) self.assertEqual(pos_809_fast, (249, 173)) # 6. dismiss_modals tích hợp is_center_modal_present modal_img = Image.new("RGB", (200, 200), (20, 20, 20)) self.assertTrue(common.is_center_modal_present(modal_img)) with patch("src.common.win32_input.press_key") as mock_press: common.dismiss_modals(img=modal_img, is_simulated=True) vks = [call.args[0] for call in mock_press.call_args_list] self.assertIn(common.VK_ESCAPE, vks) self.assertIn(common.VK_SPACE, vks) normal_img = Image.new("RGB", (200, 200), (120, 100, 80)) self.assertFalse(common.is_center_modal_present(normal_img)) with patch("src.common.win32_input.press_key") as mock_press_normal: common.dismiss_modals(img=normal_img, is_simulated=True) vks_normal = [call.args[0] for call in mock_press_normal.call_args_list] self.assertNotIn(common.VK_ESCAPE, vks_normal) self.assertIn(common.VK_SPACE, vks_normal) if __name__ == "__main__": unittest.main()