e2hang
2a844e83a8
- EmotionEngine: 5状态马尔可夫情绪机 + 蒙特卡洛转移 - VectorMemory: TF-IDF向量记忆 + SQLite持久化 + RAG检索 - AgentBrain: Ollama/OpenAI/Dummy三后端LLM - BehaviorScheduler: 优先级/冷却/活跃度调度 - FastAPI服务器 + WebSocket实时推送 - perception: 键鼠监控 + 屏幕截图 - ui/pet_window: PySide6桌宠窗口 + 像素动画 - assets/pet: 5情绪各2帧像素艺术资源
292 lines
10 KiB
Python
292 lines
10 KiB
Python
"""
|
||
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)
|