""" test_perception.py ================== perception 模块单元测试。 运行 ---- cd /Users/e2hang/hermes/code/ezvibe python -m perception.test_perception """ import sys import time import threading from pathlib import Path # 确保项目根在 sys.path _ROOT = Path(__file__).parent.parent.resolve() sys.path.insert(0, str(_ROOT)) import unittest from perception import ( ActivityDetector, ActivityLevel, KeyboardMouseMonitor, ScreenCapture, ) # ============================================================================ # ActivityDetector Tests # ============================================================================ class TestActivityDetector(unittest.TestCase): """ActivityDetector 滑动窗口统计测试。""" def test_record_single_event(self): """记录一次事件，活跃度 > 0。""" det = ActivityDetector(window_seconds=300.0, idle_timeout_seconds=60.0) ts = time.monotonic() det.record("keyboard", timestamp=ts) self.assertGreater(det.get_activity(), 0.0) def test_record_multiple_events(self): """连续记录多次事件。""" det = ActivityDetector(window_seconds=10.0, idle_timeout_seconds=5.0) now = time.monotonic() for i in range(10): det.record("keyboard", timestamp=now + i * 0.1) self.assertGreater(det.get_activity(), 0.0) def test_empty_is_idle(self): """无事件时 is_idle 返回 True。""" det = ActivityDetector(window_seconds=300.0, idle_timeout_seconds=60.0) self.assertTrue(det.is_idle()) def test_recent_event_not_idle(self): """最近有事件时 is_idle 返回 False。""" det = ActivityDetector(window_seconds=300.0, idle_timeout_seconds=60.0) det.record("mouse", timestamp=time.monotonic()) self.assertFalse(det.is_idle()) def test_old_event_becomes_idle(self): """超过 idle_timeout 的事件导致 is_idle。""" det = ActivityDetector(window_seconds=10.0, idle_timeout_seconds=0.5) old_ts = time.monotonic() - 2.0 # 2 秒前 det.record("keyboard", timestamp=old_ts) time.sleep(0.6) self.assertTrue(det.is_idle()) def test_prune_clears_old_events(self): """超出 window_seconds 的事件被清除。""" det = ActivityDetector(window_seconds=1.0, idle_timeout_seconds=60.0) old_ts = time.monotonic() - 5.0 det.record("keyboard", timestamp=old_ts) # prune 只在 record/get_activity 内触发，先调用 get_activity det.get_activity() self.assertEqual(len(det._events), 0) # 已过期被 prune def test_get_activity_capped_at_one(self): """活跃度不超过 1.0。""" det = ActivityDetector(window_seconds=0.5, idle_timeout_seconds=60.0) now = time.monotonic() for i in range(100): det.record("keyboard", timestamp=now + i * 0.001) self.assertLessEqual(det.get_activity(), 1.0) def test_is_highly_engaged_with_many_events(self): """大量事件触发 is_highly_engaged。""" det = ActivityDetector(window_seconds=10.0, idle_timeout_seconds=60.0) now = time.monotonic() for i in range(30): det.record("keyboard", timestamp=now + i * 0.05) self.assertTrue(det.is_highly_engaged()) def test_is_highly_engaged_false_when_idle(self): """空闲时（无事件）不触发 is_highly_engaged。""" det = ActivityDetector(window_seconds=300.0, idle_timeout_seconds=60.0) # 无事件状态，is_idle=True，is_highly_engaged 必须 False self.assertTrue(det.is_idle()) self.assertFalse(det.is_highly_engaged()) def test_activity_level_enum_values(self): """验证 ActivityLevel 枚举值存在。""" det = ActivityDetector(window_seconds=300.0, idle_timeout_seconds=60.0) self.assertIsInstance(det.get_activity_level(), ActivityLevel) # 空状态 → IDLE self.assertEqual(det.get_activity_level(), ActivityLevel.IDLE) # ============================================================================ # KeyboardMouseMonitor Tests # ============================================================================ class TestKeyboardMouseMonitor(unittest.TestCase): """KeyboardMouseMonitor 测试（Dummy 模式）。""" def setUp(self): self.monitor = KeyboardMouseMonitor( window_seconds=10.0, idle_timeout_seconds=5.0, use_dummy=True, # 不启动 pynput ) def tearDown(self): self.monitor.stop() def test_start_stop_idempotent(self): """start/stop 可重入幂等。""" self.monitor.start() self.monitor.start() # 重复 start 不报错 self.monitor.stop() self.monitor.stop() # 重复 stop 不报错 def test_pause_resume(self): """暂停期间不记录事件。""" self.monitor.start() self.monitor.pause() time.sleep(0.1) act_before = self.monitor.get_activity() # 手动注入事件（通过内部 API） self.monitor._activity.record("keyboard") self.monitor._activity.record("mouse") self.monitor.pause() self.monitor.resume() # 暂停期间不应记录（我们直接调用了内部方法，所以这里主要测 API） self.assertIsInstance(act_before, float) def test_get_activity_returns_float(self): """get_activity 返回 0.0~1.0 float。""" self.monitor.start() act = self.monitor.get_activity() self.assertIsInstance(act, float) self.assertGreaterEqual(act, 0.0) self.assertLessEqual(act, 1.0) def test_is_idle_initially_true(self): """初始状态电脑空闲。""" self.monitor.start() self.assertTrue(self.monitor.is_idle()) def test_is_highly_engaged_false_initially(self): """初始状态非高度专注。""" self.monitor.start() self.assertFalse(self.monitor.is_highly_engaged()) def test_get_activity_level_initially_idle(self): """初始活跃度等级为 IDLE。""" self.monitor.start() self.assertEqual(self.monitor.get_activity_level(), ActivityLevel.IDLE) def test_record_events_increases_activity(self): """直接记录事件后活跃度上升。""" self.monitor.start() now = time.monotonic() for i in range(20): self.monitor._activity.record("keyboard", timestamp=now + i * 0.05) self.assertGreater(self.monitor.get_activity(), 0.0) self.assertTrue(self.monitor.is_highly_engaged()) def test_record_events_clears_idle(self): """记录事件后 is_idle 变 False。""" self.monitor.start() self.monitor._activity.record("keyboard") self.assertFalse(self.monitor.is_idle()) def test_get_event_count(self): """get_event_count 返回正确数量。""" self.monitor.start() now = time.monotonic() for i in range(5): self.monitor._activity.record("keyboard", timestamp=now + i * 0.1) count = self.monitor.get_event_count(window_seconds=10.0) self.assertEqual(count, 5) # ============================================================================ # ScreenCapture Tests # ============================================================================ class TestScreenCapture(unittest.TestCase): """ScreenCapture 测试（可能因平台/依赖不可用）。""" def test_init_loads_mss(self): """初始化成功（mss 可能可用或不可用）。""" cap = ScreenCapture() # 不崩溃即通过 def test_is_available_returns_bool(self): """is_available 返回布尔值。""" cap = ScreenCapture() self.assertIsInstance(cap.is_available(), bool) def test_get_monitors_returns_list(self): """get_monitors 返回列表。""" cap = ScreenCapture() result = cap.get_monitors() self.assertIsInstance(result, list) def test_capture_returns_image_or_none(self): """capture 返回 PIL.Image 或 None。""" cap = ScreenCapture() result = cap.capture(monitor=0) if cap.is_available(): self.assertIsNotNone(result) else: self.assertIsNone(result) def test_capture_region_returns_image_or_none(self): """capture_region 返回 PIL.Image 或 None。""" cap = ScreenCapture() result = cap.capture_region(0, 0, 100, 100) if cap.is_available(): self.assertIsNotNone(result) else: self.assertIsNone(result) def test_extract_text_returns_string(self): """extract_text 返回字符串（OCR 可能不可用）。""" cap = ScreenCapture() # 构造一个假的 RGB 图像（不依赖 PIL，直接测试 OCR 返回） text = cap.extract_text(None) # OCR 对 None 返回空字符串 self.assertIsInstance(text, str) # ============================================================================ # Global Monitor Tests # ============================================================================ class TestGlobalMonitor(unittest.TestCase): """全局单例 monitor 测试。""" def setUp(self): from perception import stop_global_monitor stop_global_monitor() # 确保干净状态 def tearDown(self): from perception import stop_global_monitor stop_global_monitor() def test_get_global_monitor_returns_same_instance(self): """多次调用返回同一实例。""" from perception import get_global_monitor m1 = get_global_monitor() m2 = get_global_monitor() self.assertIs(m1, m2) def test_global_monitor_works(self): """全局 monitor 基本功能正常。""" from perception import get_global_monitor m = get_global_monitor() self.assertIsInstance(m, KeyboardMouseMonitor) self.assertTrue(m._running) act = m.get_activity() self.assertGreaterEqual(act, 0.0) # ============================================================================ # Main # ============================================================================ if __name__ == "__main__": print("Running perception tests...") # 打印环境信息 print(f" Python: {sys.version.split()[0]}") print(f" Platform: {sys.platform}") from perception.keyboard_mouse_monitor import _check_pynput print(f" pynput available: {_check_pynput()}") unittest.main(verbosity=2)