test_send_msg.py
5.57 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
#!/usr/bin/env python3
"""
发送消息测试 — 验证本地 `am broadcast` 触发 LSPosed 模块发送消息。
在 Ubuntu proot 容器内执行。
"""
import subprocess, sys, json, time, os
CONFIG_PATH = "config.json"
REPORT = {"test": "send_msg", "steps": {}, "overall": None}
def log_step(name, ok, detail=""):
status = "[OK]" if ok else "[FAIL]"
print(f" {status} {name}")
if detail:
print(f" {detail}")
REPORT["steps"][name] = {"ok": ok, "detail": detail}
return ok
def send_via_am_broadcast(to_user, content, am_path="/system/bin/am", to_user_display_name=None):
"""Send message via Android am broadcast."""
cmd = [
am_path, "broadcast",
"-a", "com.hermes.wechat.SEND_MSG",
"--es", "toUser", to_user,
"--es", "toUserDisplayName", to_user_display_name or to_user,
"--es", "content", content
]
r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
return r.returncode == 0, r.stdout, r.stderr
def send_via_content_provider(to_user, content):
"""Alternative: use content insert (if provider is available)."""
cmd = [
"/system/bin/am", "startservice",
# ContentProvider insert via am is not directly supported,
# but we can use content command if available
]
# Try using 'content' command from Termux or system
for content_path in ["/data/data/com.termux/files/usr/bin/content", "/system/bin/content"]:
if os.path.exists(content_path):
cmd = [
content_path, "insert",
"--uri", "content://com.hermes.wechat.xposed.provider/send",
"--bind", f"toUser:s:{to_user}",
"--bind", f"content:s:{content}"
]
r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
if r.returncode == 0:
return True, r.stdout, r.stderr
return False, "", "content command not found"
def check_lsposed_log():
"""Check LSPosed logs for HermesWX entries."""
try:
# Try multiple log file patterns
for pattern in ["modules_*.log", "verbose_*.log"]:
r = subprocess.run(
["su", "-c", f"cat /data/adb/lspd/log/{pattern} 2>/dev/null | grep HermesWX | tail -5"],
capture_output=True, text=True, timeout=5
)
if r.stdout.strip():
return r.stdout.strip()
return ""
except Exception:
return ""
def check_logcat():
"""Check logcat for HermesWX."""
try:
r = subprocess.run(
["su", "-c", "logcat -d -s HermesWX:D | tail -5"],
capture_output=True, text=True, timeout=5
)
return r.stdout.strip()
except Exception:
return ""
def main():
print("=" * 60)
print("发送消息测试")
print("=" * 60)
config = {}
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH) as f:
config = json.load(f)
all_ok = True
am_path = config.get("send", {}).get("am_path", "/system/bin/am")
test_user = config.get("auto_reply", {}).get("contacts", ["filehelper"])[0]
display_names = config.get("contact_display_names", {})
test_user_display = display_names.get(test_user, test_user)
test_content = f"本地发送测试 {time.strftime('%H:%M:%S')}"
# Step 1: Check am exists
print("\n[Step 1] 检查 am 命令")
ok = os.path.exists(am_path)
all_ok &= log_step("am_exists", ok, f"路径: {am_path}")
# Step 2: Send via broadcast
print(f"\n[Step 2] 发送测试消息 to={test_user} (search={test_user_display})")
print(f" 内容: {test_content}")
ok, stdout, stderr = send_via_am_broadcast(test_user, test_content, am_path, test_user_display)
detail = f"returncode=0" if ok else f"stderr: {stderr[:200]}"
all_ok &= log_step("send_broadcast", ok, detail)
# Step 3: Wait and check logs
if ok:
print("\n[Step 3] 等待 3 秒后检查日志...")
time.sleep(3)
lsposed_log = check_lsposed_log()
logcat_log = check_logcat()
has_hermes = "HermesWX" in (lsposed_log + logcat_log)
has_received = "Broadcast received" in (lsposed_log + logcat_log)
has_result = "Send result:" in (lsposed_log + logcat_log)
log_detail = []
if lsposed_log:
log_detail.append(f"LSPosed: {lsposed_log[:150]}")
if logcat_log:
log_detail.append(f"logcat: {logcat_log[:150]}")
all_ok &= log_step("lsposed_log", has_hermes, " | ".join(log_detail) if log_detail else "无日志")
all_ok &= log_step("broadcast_received", has_received, "LSPosed 模块是否收到广播")
all_ok &= log_step("send_result", has_result, "消息是否成功提交")
# Step 4: Alternative method test (content provider)
print("\n[Step 4] 备用方式: content provider")
ok, stdout, stderr = send_via_content_provider(test_user, f"provider test {time.strftime('%H:%M:%S')}")
all_ok &= log_step("send_provider", ok, f"stdout: {stdout[:100]}" if ok else f"stderr: {stderr[:100]}")
# Summary
REPORT["overall"] = "PASS" if all_ok else "FAIL"
REPORT["timestamp"] = subprocess.check_output(["date", "-Iseconds"]).decode().strip()
with open("test_report_send_msg.json", "w") as f:
json.dump(REPORT, f, indent=2, ensure_ascii=False)
print("\n" + "=" * 60)
print(f"结果: {REPORT['overall']}")
print("=" * 60)
print("\n请手动检查微信，确认消息是否已发送。")
print("详细报告: test_report_send_msg.json")
return 0 if all_ok else 1
if __name__ == "__main__":
sys.exit(main())