d6bb3f1acc
- Hook通道全量实现: frida_manager(三模式usb/gadget/remote)、hook_executor(30动作)、event_reporter - 微信Hook脚本 v2.1: wechat_hook_v2.js 消息/联系人/好友/群管理/朋友圈 完整RPC - 真机联调验证: ADB通道启动微信/截图/UI树/点击/发消息 全链路通过 - 54项单元测试全绿: test_frida_manager(14) + test_hook_executor(20) + test_hook_module_api(20) - SDK服务端 116 个API端点全部在线 - 开发文档/进度表更新: Phase 2 进度 90%, 整体 88% Made-with: Cursor
235 lines
9.3 KiB
Python
235 lines
9.3 KiB
Python
"""
|
|
Hook 模块管理 API 集成测试 (H32)
|
|
测试 hook_module_service 的 CRUD、脚本管理和事件总线。
|
|
使用临时目录，不依赖 MongoDB 或真实文件。
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import json
|
|
import asyncio
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, AsyncMock
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
|
|
|
|
from services.hook_module_service import HookModuleService
|
|
|
|
|
|
def run_async(coro):
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
loop = None
|
|
if loop and loop.is_running():
|
|
import concurrent.futures
|
|
with concurrent.futures.ThreadPoolExecutor() as pool:
|
|
return pool.submit(asyncio.run, coro).result()
|
|
return asyncio.run(coro)
|
|
|
|
|
|
class TestHookModuleServiceCRUD(unittest.TestCase):
|
|
"""模块 CRUD 测试"""
|
|
|
|
def setUp(self):
|
|
self.svc = HookModuleService()
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
self.svc.data_dir = Path(self.tmpdir) / "hook"
|
|
self.svc.scripts_dir = self.svc.data_dir / "scripts"
|
|
self.svc.modules_file = self.svc.data_dir / "modules.json"
|
|
self.svc.events_file = self.svc.data_dir / "events.jsonl"
|
|
self.svc.device_state_file = self.svc.data_dir / "device_modules.json"
|
|
self.svc._ensure_store()
|
|
|
|
def test_create_module(self):
|
|
module = run_async(self.svc.upsert_module({
|
|
"module_id": "wechat_v2",
|
|
"name": "微信Hook",
|
|
"version": "2.1.0",
|
|
"description": "完整微信Hook模块",
|
|
"scopes": ["com.tencent.mm"],
|
|
"capabilities": ["send_message", "get_contacts"],
|
|
}))
|
|
self.assertEqual(module["module_id"], "wechat_v2")
|
|
self.assertEqual(module["version"], "2.1.0")
|
|
self.assertTrue(module["enabled"])
|
|
|
|
def test_get_module(self):
|
|
run_async(self.svc.upsert_module({"module_id": "test1", "name": "Test"}))
|
|
result = run_async(self.svc.get_module("test1"))
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result["name"], "Test")
|
|
|
|
def test_get_nonexistent_module(self):
|
|
result = run_async(self.svc.get_module("nonexistent"))
|
|
self.assertIsNone(result)
|
|
|
|
def test_list_modules(self):
|
|
run_async(self.svc.upsert_module({"module_id": "m1", "name": "M1", "enabled": True}))
|
|
run_async(self.svc.upsert_module({"module_id": "m2", "name": "M2", "enabled": False}))
|
|
all_modules = run_async(self.svc.list_modules())
|
|
self.assertEqual(len(all_modules), 2)
|
|
enabled = run_async(self.svc.list_modules(enabled=True))
|
|
self.assertEqual(len(enabled), 1)
|
|
self.assertEqual(enabled[0]["module_id"], "m1")
|
|
|
|
def test_list_modules_by_scope(self):
|
|
run_async(self.svc.upsert_module({"module_id": "wc", "name": "WC", "scopes": ["com.tencent.mm"]}))
|
|
run_async(self.svc.upsert_module({"module_id": "dy", "name": "DY", "scopes": ["com.ss.android.ugc.aweme"]}))
|
|
wechat = run_async(self.svc.list_modules(scope="com.tencent.mm"))
|
|
self.assertEqual(len(wechat), 1)
|
|
self.assertEqual(wechat[0]["module_id"], "wc")
|
|
|
|
def test_update_module(self):
|
|
run_async(self.svc.upsert_module({"module_id": "upd", "name": "V1", "version": "1.0.0"}))
|
|
updated = run_async(self.svc.upsert_module({"module_id": "upd", "name": "V2", "version": "2.0.0"}))
|
|
self.assertEqual(updated["name"], "V2")
|
|
self.assertEqual(updated["version"], "2.0.0")
|
|
|
|
def test_delete_module(self):
|
|
run_async(self.svc.upsert_module({"module_id": "del", "name": "Del"}))
|
|
ok = run_async(self.svc.delete_module("del"))
|
|
self.assertTrue(ok)
|
|
result = run_async(self.svc.get_module("del"))
|
|
self.assertIsNone(result)
|
|
|
|
def test_delete_nonexistent(self):
|
|
ok = run_async(self.svc.delete_module("nope"))
|
|
self.assertFalse(ok)
|
|
|
|
def test_set_scope(self):
|
|
run_async(self.svc.upsert_module({"module_id": "sc", "name": "SC", "scopes": ["a"]}))
|
|
updated = run_async(self.svc.set_scope("sc", ["a", "b", "c"]))
|
|
self.assertEqual(updated["scopes"], ["a", "b", "c"])
|
|
|
|
def test_set_enabled(self):
|
|
run_async(self.svc.upsert_module({"module_id": "en", "name": "EN", "enabled": True}))
|
|
updated = run_async(self.svc.set_enabled("en", False))
|
|
self.assertFalse(updated["enabled"])
|
|
|
|
|
|
class TestHookModuleServiceScripts(unittest.TestCase):
|
|
"""脚本管理测试"""
|
|
|
|
def setUp(self):
|
|
self.svc = HookModuleService()
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
self.svc.data_dir = Path(self.tmpdir) / "hook"
|
|
self.svc.scripts_dir = self.svc.data_dir / "scripts"
|
|
self.svc.modules_file = self.svc.data_dir / "modules.json"
|
|
self.svc.events_file = self.svc.data_dir / "events.jsonl"
|
|
self.svc.device_state_file = self.svc.data_dir / "device_modules.json"
|
|
self.svc._ensure_store()
|
|
|
|
def test_save_and_list_scripts(self):
|
|
content = b"console.log('hello');"
|
|
saved = run_async(self.svc.save_script("test_script", content))
|
|
self.assertEqual(saved["script_id"], "test_script")
|
|
self.assertIn("sha256:", saved["hash"])
|
|
scripts = run_async(self.svc.list_scripts())
|
|
self.assertEqual(len(scripts), 1)
|
|
self.assertEqual(scripts[0]["script_id"], "test_script")
|
|
|
|
def test_get_script_path(self):
|
|
run_async(self.svc.save_script("s1", b"var x = 1;"))
|
|
path = run_async(self.svc.get_script_path("s1"))
|
|
self.assertIsNotNone(path)
|
|
self.assertTrue(path.exists())
|
|
none_path = run_async(self.svc.get_script_path("nonexistent"))
|
|
self.assertIsNone(none_path)
|
|
|
|
|
|
class TestHookModuleServiceEvents(unittest.TestCase):
|
|
"""事件总线测试"""
|
|
|
|
def setUp(self):
|
|
self.svc = HookModuleService()
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
self.svc.data_dir = Path(self.tmpdir) / "hook"
|
|
self.svc.scripts_dir = self.svc.data_dir / "scripts"
|
|
self.svc.modules_file = self.svc.data_dir / "modules.json"
|
|
self.svc.events_file = self.svc.data_dir / "events.jsonl"
|
|
self.svc.device_state_file = self.svc.data_dir / "device_modules.json"
|
|
self.svc._ensure_store()
|
|
|
|
def test_add_and_list_events(self):
|
|
run_async(self.svc.add_event({
|
|
"event_type": "message_received",
|
|
"device_id": "dev1",
|
|
"platform": "wechat",
|
|
"payload": {"content": "hello"},
|
|
}))
|
|
run_async(self.svc.add_event({
|
|
"event_type": "friend_request",
|
|
"device_id": "dev1",
|
|
"platform": "wechat",
|
|
}))
|
|
all_events = run_async(self.svc.list_events())
|
|
self.assertEqual(len(all_events), 2)
|
|
|
|
def test_filter_events_by_type(self):
|
|
run_async(self.svc.add_event({"event_type": "msg", "device_id": "d1"}))
|
|
run_async(self.svc.add_event({"event_type": "friend", "device_id": "d1"}))
|
|
msgs = run_async(self.svc.list_events(event_type="msg"))
|
|
self.assertEqual(len(msgs), 1)
|
|
|
|
def test_filter_events_by_device(self):
|
|
run_async(self.svc.add_event({"event_type": "a", "device_id": "d1"}))
|
|
run_async(self.svc.add_event({"event_type": "b", "device_id": "d2"}))
|
|
d1_events = run_async(self.svc.list_events(device_id="d1"))
|
|
self.assertEqual(len(d1_events), 1)
|
|
|
|
def test_event_limit(self):
|
|
for i in range(10):
|
|
run_async(self.svc.add_event({"event_type": f"ev_{i}", "device_id": "d"}))
|
|
limited = run_async(self.svc.list_events(limit=3))
|
|
self.assertEqual(len(limited), 3)
|
|
|
|
|
|
class TestHookModuleServiceDeviceState(unittest.TestCase):
|
|
"""设备模块状态测试"""
|
|
|
|
def setUp(self):
|
|
self.svc = HookModuleService()
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
self.svc.data_dir = Path(self.tmpdir) / "hook"
|
|
self.svc.scripts_dir = self.svc.data_dir / "scripts"
|
|
self.svc.modules_file = self.svc.data_dir / "modules.json"
|
|
self.svc.events_file = self.svc.data_dir / "events.jsonl"
|
|
self.svc.device_state_file = self.svc.data_dir / "device_modules.json"
|
|
self.svc._ensure_store()
|
|
|
|
def test_update_device_probe(self):
|
|
run_async(self.svc.update_device_probe("dev1", {
|
|
"supports_hook": True,
|
|
"root_status": True,
|
|
"frida_version": "16.2.1",
|
|
}))
|
|
state = run_async(self.svc.get_device_modules("dev1"))
|
|
self.assertTrue(state["supports_hook"])
|
|
self.assertEqual(state["frida_version"], "16.2.1")
|
|
|
|
def test_deploy_script_to_device(self):
|
|
result = run_async(self.svc.deploy_script("wechat_v2", ["dev1", "dev2"]))
|
|
self.assertEqual(len(result["deployed"]), 2)
|
|
state = run_async(self.svc.get_device_modules("dev1"))
|
|
self.assertEqual(len(state["modules"]), 1)
|
|
self.assertEqual(state["modules"][0]["module_id"], "wechat_v2")
|
|
|
|
def test_reload_device_modules(self):
|
|
run_async(self.svc.deploy_script("m1", ["dev1"]))
|
|
result = run_async(self.svc.reload_device_modules("dev1", ["m1"]))
|
|
self.assertEqual(result["reloaded"], ["m1"])
|
|
|
|
def test_device_logs(self):
|
|
run_async(self.svc.add_device_log("dev1", "m1", "Hook loaded"))
|
|
run_async(self.svc.add_device_log("dev1", "m1", "Message intercepted"))
|
|
logs = run_async(self.svc.get_device_logs("dev1", "m1"))
|
|
self.assertEqual(len(logs), 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|