1135 lines
39 KiB
Python
1135 lines
39 KiB
Python
"""消息管理九路由契约回归。
|
|
|
|
本文件只做离线契约测试：不连接真机、不执行真实消息写操作。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.routing import APIRoute
|
|
from pydantic import BaseModel, ValidationError
|
|
|
|
APP = Path(__file__).resolve().parents[1] / "app"
|
|
if str(APP) not in sys.path:
|
|
sys.path.insert(0, str(APP))
|
|
|
|
from routers import unified
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def payment_test_gate(monkeypatch):
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "filehelper,卡若,udbfnvtk,test-user")
|
|
unified._payment_rate_events.clear()
|
|
|
|
|
|
MESSAGE_ROUTES = {
|
|
"/message/send": unified.SendMessageRequest,
|
|
"/message/list": unified.GetMessagesRequest,
|
|
"/message/sync-since": unified.MessageSyncSinceRequest,
|
|
"/message/batch-send": unified.BatchSendMessageRequest,
|
|
"/comment/reply": unified.ReplyCommentRequest,
|
|
"/message/forward": unified.ForwardMessageRequest,
|
|
"/message/recall": unified.RecallMessageRequest,
|
|
"/message/send-card": unified.SendCardRequest,
|
|
"/message/voice": unified.SendVoiceMessageRequest,
|
|
}
|
|
|
|
REQUIRED_FIELDS = {
|
|
unified.SendMessageRequest: {"device_id", "platform", "to_id", "content"},
|
|
unified.GetMessagesRequest: {"device_id", "platform"},
|
|
unified.MessageSyncSinceRequest: {"device_id"},
|
|
unified.BatchSendMessageRequest: {"device_id", "platform", "to_ids", "content"},
|
|
unified.ReplyCommentRequest: {"device_id", "platform", "comment_id", "content"},
|
|
unified.ForwardMessageRequest: {"device_id", "platform", "to_id"},
|
|
unified.RecallMessageRequest: {"device_id"},
|
|
unified.SendCardRequest: {"device_id", "platform", "to_id", "card_wxid"},
|
|
unified.SendVoiceMessageRequest: {"device_id"},
|
|
}
|
|
|
|
VALID_REQUESTS = {
|
|
unified.SendMessageRequest: {
|
|
"device_id": "device-1",
|
|
"platform": "wechat",
|
|
"to_id": "filehelper",
|
|
"content": "hello",
|
|
},
|
|
unified.GetMessagesRequest: {"device_id": "device-1", "platform": "wechat"},
|
|
unified.MessageSyncSinceRequest: {"device_id": "device-1"},
|
|
unified.BatchSendMessageRequest: {
|
|
"device_id": "device-1",
|
|
"platform": "wechat",
|
|
"to_ids": ["filehelper"],
|
|
"content": "hello",
|
|
},
|
|
unified.ReplyCommentRequest: {
|
|
"device_id": "device-1",
|
|
"platform": "douyin",
|
|
"comment_id": "comment-1",
|
|
"content": "reply",
|
|
},
|
|
unified.ForwardMessageRequest: {
|
|
"device_id": "device-1",
|
|
"platform": "wechat",
|
|
"to_id": "filehelper",
|
|
"msg_svr_id": "10001",
|
|
},
|
|
unified.RecallMessageRequest: {
|
|
"device_id": "device-1",
|
|
"platform": "wechat",
|
|
"msg_svr_id": "10001",
|
|
},
|
|
unified.SendCardRequest: {
|
|
"device_id": "device-1",
|
|
"platform": "wechat",
|
|
"to_id": "filehelper",
|
|
"card_wxid": "wxid_card",
|
|
},
|
|
unified.SendVoiceMessageRequest: {
|
|
"device_id": "device-1",
|
|
"platform": "wechat",
|
|
"to_id": "filehelper",
|
|
},
|
|
}
|
|
|
|
|
|
def _route(path: str) -> APIRoute:
|
|
matches = [
|
|
route
|
|
for route in unified.router.routes
|
|
if isinstance(route, APIRoute)
|
|
and route.path == path
|
|
and "POST" in route.methods
|
|
]
|
|
assert len(matches) == 1, f"{path} 应且仅应注册一个POST路由"
|
|
return matches[0]
|
|
|
|
|
|
def _assert_invalid(model: type[BaseModel], patch: dict[str, Any]) -> None:
|
|
payload = dict(VALID_REQUESTS[model])
|
|
payload.update(patch)
|
|
with pytest.raises(ValidationError):
|
|
model.model_validate(payload)
|
|
|
|
|
|
@pytest.mark.parametrize("path", MESSAGE_ROUTES)
|
|
def test_nine_routes_have_explicit_response_model_and_error_contract(path: str):
|
|
route = _route(path)
|
|
assert route.response_model not in (None, dict)
|
|
assert {400, 401, 422, 503}.issubset(set(route.responses))
|
|
for status in (400, 401, 422, 503):
|
|
assert route.responses[status].get("description")
|
|
|
|
|
|
@pytest.mark.parametrize("model,required", REQUIRED_FIELDS.items())
|
|
def test_request_models_publish_required_fields_and_accept_valid_payload(
|
|
model: type[BaseModel], required: set[str]
|
|
):
|
|
fields = model.model_fields
|
|
assert required.issubset(fields)
|
|
assert {name for name, field in fields.items() if field.is_required()} == required
|
|
assert model.model_validate(VALID_REQUESTS[model])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("model", "patch"),
|
|
[
|
|
(unified.SendMessageRequest, {"device_id": ""}),
|
|
(unified.SendMessageRequest, {"to_id": ""}),
|
|
(unified.SendMessageRequest, {"content": ""}),
|
|
(unified.SendMessageRequest, {"timeout_seconds": 4}),
|
|
(unified.SendMessageRequest, {"timeout_seconds": 181}),
|
|
(unified.GetMessagesRequest, {"limit": 0}),
|
|
(unified.GetMessagesRequest, {"limit": unified.MAX_MESSAGE_PULL_LIMIT + 1}),
|
|
(unified.GetMessagesRequest, {"offset": -1}),
|
|
(unified.GetMessagesRequest, {"since_time": -1}),
|
|
(unified.MessageSyncSinceRequest, {"since_time": -1}),
|
|
(unified.MessageSyncSinceRequest, {"limit": 0}),
|
|
(unified.MessageSyncSinceRequest, {"limit": unified.MAX_MESSAGE_PULL_LIMIT + 1}),
|
|
(unified.MessageSyncSinceRequest, {"offset": -1}),
|
|
(unified.BatchSendMessageRequest, {"to_ids": []}),
|
|
(unified.BatchSendMessageRequest, {"to_ids": ["x"] * 101}),
|
|
(unified.BatchSendMessageRequest, {"content": ""}),
|
|
(unified.BatchSendMessageRequest, {"interval": 0.4}),
|
|
(unified.BatchSendMessageRequest, {"interval": 31}),
|
|
(unified.ReplyCommentRequest, {"comment_id": ""}),
|
|
(unified.ReplyCommentRequest, {"content": ""}),
|
|
(unified.ForwardMessageRequest, {"to_id": ""}),
|
|
(unified.RecallMessageRequest, {"device_id": ""}),
|
|
(unified.SendCardRequest, {"to_id": ""}),
|
|
(unified.SendCardRequest, {"card_wxid": ""}),
|
|
(unified.SendVoiceMessageRequest, {"device_id": ""}),
|
|
(unified.SendVoiceMessageRequest, {"duration": 0}),
|
|
(unified.SendVoiceMessageRequest, {"duration": 61}),
|
|
],
|
|
)
|
|
def test_request_model_constraints_reject_invalid_values(
|
|
model: type[BaseModel], patch: dict[str, Any]
|
|
):
|
|
_assert_invalid(model, patch)
|
|
|
|
|
|
def test_message_error_passthrough_helper_preserves_original_semantics():
|
|
response = unified._message_operation_response(
|
|
{
|
|
"code": 503,
|
|
"success": False,
|
|
"error_code": "frida_receiver_missing",
|
|
"error_message": "no_receiver_registered",
|
|
"retryable": True,
|
|
"retry_after_seconds": 12,
|
|
"channel_used": "hook",
|
|
"trace_id": "trace-message-contract-001",
|
|
"raw_rpc_receipt": {"rpc": "sendMessage", "implemented": False},
|
|
"readback": None,
|
|
},
|
|
action="send_message",
|
|
target_id="filehelper",
|
|
require_message_id=True,
|
|
)
|
|
|
|
assert response["code"] == 503
|
|
assert response["success"] is False
|
|
assert response["channel_used"] == "hook"
|
|
assert response["data"]["error_code"] == "frida_receiver_missing"
|
|
assert response["data"]["error_message"] == "no_receiver_registered"
|
|
assert response["data"]["retryable"] is True
|
|
assert response["data"]["retry_after_seconds"] == 12
|
|
assert response["trace_id"] == "trace-message-contract-001"
|
|
assert response["data"]["trace_id"] == "trace-message-contract-001"
|
|
assert response["data"]["raw_rpc_receipt"] == {"rpc": "sendMessage", "implemented": False}
|
|
assert response["data"]["readback"] is None
|
|
assert response["feedback"]["message"] == "no_receiver_registered"
|
|
|
|
|
|
def test_message_response_does_not_generate_trace_when_execution_lacks_evidence():
|
|
response = unified._message_operation_response(
|
|
{"code": 503, "success": False, "error_code": "hook_unavailable"},
|
|
action="send_message",
|
|
target_id="filehelper",
|
|
require_message_id=True,
|
|
)
|
|
assert response["trace_id"] == ""
|
|
assert response["data"]["raw_rpc_receipt"] is None
|
|
assert response["data"]["readback"] is None
|
|
|
|
|
|
def test_sync_since_is_strictly_incremental_sorted_and_deduplicated(monkeypatch):
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: True)
|
|
|
|
async def fake_execute(*args, **kwargs):
|
|
return {
|
|
"code": 200,
|
|
"_channel_used": "hook",
|
|
"data": {
|
|
"success": True,
|
|
"messages": [
|
|
{"message_id": "m2", "timestamp": 102, "content": "second"},
|
|
{"message_id": "boundary", "timestamp": 100, "content": "equal"},
|
|
{"message_id": "m1", "timestamp": 101, "content": "first"},
|
|
{"message_id": "m1", "timestamp": 101, "content": "duplicate"},
|
|
{"message_id": "old", "timestamp": 99, "content": "old"},
|
|
],
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
response = asyncio.run(
|
|
unified.sync_messages_since(
|
|
unified.MessageSyncSinceRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
since_time=100,
|
|
limit=10,
|
|
)
|
|
)
|
|
)
|
|
|
|
messages = response["data"]["messages"]
|
|
assert [item["message_id"] for item in messages] == ["m1", "m2"]
|
|
assert all(item["timestamp"] > 100 for item in messages)
|
|
assert response["data"]["count"] == 2
|
|
assert response["data"]["since_time"] == 100
|
|
assert response["data"]["next_since_time"] == 102
|
|
assert response["channel_used"] == "hook"
|
|
|
|
|
|
def test_ai_channel_does_not_generate_fake_message_id(monkeypatch):
|
|
async def pass_guard(*args, **kwargs):
|
|
return {"pass": True, "content": "hello"}
|
|
|
|
async def fake_agent(*args, **kwargs):
|
|
return {"success": True, "channel_used": "ai_agent"}
|
|
|
|
async def no_log(*args, **kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(unified, "_anti_ban_guard", pass_guard)
|
|
monkeypatch.setattr(unified, "_get_device_mode", lambda *args, **kwargs: "offline")
|
|
monkeypatch.setattr(unified, "_send_via_agent", fake_agent)
|
|
monkeypatch.setattr(unified.device_manager, "log_command", no_log)
|
|
monkeypatch.setattr(
|
|
unified.rate_limiter, "trip_silent_throttle", lambda *args, **kwargs: 60
|
|
)
|
|
|
|
response = asyncio.run(
|
|
unified.send_message(
|
|
unified.SendMessageRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
to_id="filehelper",
|
|
content="hello",
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["channel_used"] == "ai_agent"
|
|
assert response["data"]["message_id"] is None
|
|
assert response["data"]["verified"] is False
|
|
assert response["data"]["status"] == "accepted_unverified"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("handler", "request_obj", "action"),
|
|
[
|
|
(
|
|
unified.forward_message,
|
|
unified.ForwardMessageRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
to_id="filehelper",
|
|
),
|
|
"forward_message",
|
|
),
|
|
(
|
|
unified.recall_message,
|
|
unified.RecallMessageRequest(device_id="device-1", platform="wechat"),
|
|
"recall_message",
|
|
),
|
|
],
|
|
)
|
|
def test_wechat_forward_and_recall_require_msg_svr_id(
|
|
monkeypatch, handler, request_obj, action
|
|
):
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: True)
|
|
|
|
async def should_not_execute(*args, **kwargs):
|
|
pytest.fail("参数校验失败时不应下发设备动作")
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", should_not_execute)
|
|
response = asyncio.run(handler(request_obj))
|
|
|
|
assert response["code"] == 422
|
|
assert response["success"] is False
|
|
assert response["data"]["action"] == action
|
|
assert response["data"]["status"] == "validation_failed"
|
|
assert response["data"]["error_code"] == "validation_failed"
|
|
assert response["channel_used"] == "none"
|
|
|
|
|
|
def test_voice_rejects_empty_target_before_device_execution(monkeypatch):
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: True)
|
|
|
|
async def should_not_execute(*args, **kwargs):
|
|
pytest.fail("空目标时不应下发设备动作")
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", should_not_execute)
|
|
response = asyncio.run(
|
|
unified.send_voice_message(
|
|
unified.SendVoiceMessageRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
to_id=" ",
|
|
user_id=" ",
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["code"] == 422
|
|
assert response["success"] is False
|
|
assert response["data"]["status"] == "validation_failed"
|
|
assert response["data"]["error_code"] == "validation_failed"
|
|
assert response["channel_used"] == "none"
|
|
|
|
|
|
def test_batch_guard_returns_one_failure_item_per_target(monkeypatch):
|
|
async def blocked_guard(*args, **kwargs):
|
|
return {"pass": False, "reason": "新号期禁止高风险操作: batch_send"}
|
|
|
|
monkeypatch.setattr(unified, "_anti_ban_guard", blocked_guard)
|
|
response = asyncio.run(
|
|
unified.batch_send_message(
|
|
unified.BatchSendMessageRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
to_ids=["filehelper", "wxid_target"],
|
|
content="batch test",
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["code"] == 503
|
|
assert response["data"]["failed_count"] == 2
|
|
assert [item["to_id"] for item in response["data"]["failed"]] == [
|
|
"filehelper",
|
|
"wxid_target",
|
|
]
|
|
assert all(
|
|
item["error_code"] == "rate_limited"
|
|
for item in response["data"]["failed"]
|
|
)
|
|
|
|
|
|
def test_send_red_packet_defaults_to_verified_dry_run(monkeypatch):
|
|
monkeypatch.setattr(
|
|
unified,
|
|
"_check_device_online",
|
|
lambda *args, **kwargs: pytest.fail("演练模式不应连接设备"),
|
|
)
|
|
response = asyncio.run(
|
|
unified.send_red_packet(
|
|
unified.RedPacketRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
to_id="filehelper",
|
|
amount="0.01",
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["code"] == 200
|
|
assert response["success"] is True
|
|
assert response["data"]["verified"] is True
|
|
assert response["data"]["dry_run"] is True
|
|
assert response["data"]["confirm_required"] is True
|
|
assert response["channel_used"] == "dry_run"
|
|
|
|
|
|
def test_red_packet_real_failure_is_not_wrapped_as_success(monkeypatch):
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: True)
|
|
|
|
async def failed_execute(*args, **kwargs):
|
|
return {
|
|
"code": 200,
|
|
"success": False,
|
|
"error": "no_receiver_registered",
|
|
"_channel_used": "server/frida",
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", failed_execute)
|
|
response = asyncio.run(
|
|
unified.send_red_packet(
|
|
unified.RedPacketRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
to_id="filehelper",
|
|
amount="0.01",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["code"] == 503
|
|
assert response["success"] is False
|
|
assert response["data"]["error_message"] == "no_receiver_registered"
|
|
assert response["channel_used"] == "server/frida"
|
|
|
|
|
|
def test_send_red_packet_real_path_checks_hook_action(monkeypatch):
|
|
checked = {}
|
|
|
|
def fake_check(device_id, platform="wechat", action=""):
|
|
checked.update(device_id=device_id, platform=platform, action=action)
|
|
return True
|
|
|
|
async def fake_execute(*args, **kwargs):
|
|
return {
|
|
"code": 200,
|
|
"success": True,
|
|
"verified": True,
|
|
"payment_status": "prepared_waiting_pay_confirm",
|
|
"trace_id": "trace-send-red-packet",
|
|
"raw_rpc_receipt": {"err_type": 0, "err_code": 0},
|
|
"readback": {"message_id": "msg-red-packet"},
|
|
"_channel_used": "server/frida",
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_check_device_online", fake_check)
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
response = asyncio.run(
|
|
unified.send_red_packet(
|
|
unified.RedPacketRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
to_id="filehelper",
|
|
amount="0.01",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
)
|
|
)
|
|
|
|
assert checked["action"] == "send_red_packet"
|
|
assert response["code"] == 200
|
|
assert response["channel_used"] == "server/frida"
|
|
|
|
|
|
def test_receive_red_packet_defaults_to_verified_dry_run(monkeypatch):
|
|
monkeypatch.setattr(
|
|
unified,
|
|
"_check_device_online",
|
|
lambda *args, **kwargs: pytest.fail("演练模式不应连接设备"),
|
|
)
|
|
response = asyncio.run(
|
|
unified.receive_red_packet(
|
|
unified.RedPacketReceiveRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
msg_svr_id="msg-1",
|
|
from_id="卡若",
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["code"] == 200
|
|
assert response["success"] is True
|
|
assert response["data"]["dry_run"] is True
|
|
assert response["data"]["confirm_required"] is True
|
|
assert response["data"]["msg_svr_id"] == "msg-1"
|
|
assert response["channel_used"] == "dry_run"
|
|
|
|
|
|
def test_receive_red_packet_real_path_uses_msg_svr_id(monkeypatch):
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: True)
|
|
called = {}
|
|
|
|
async def fake_execute(device_id, platform, action, params, **kwargs):
|
|
called.update(
|
|
device_id=device_id,
|
|
platform=platform,
|
|
action=action,
|
|
params=params,
|
|
)
|
|
return {
|
|
"code": 200,
|
|
"success": True,
|
|
"verified": True,
|
|
"status": "received",
|
|
"msg_svr_id": params["msg_svr_id"],
|
|
"trace_id": "trace-receive-red-packet",
|
|
"raw_rpc_receipt": {"err_type": 0, "err_code": 0},
|
|
"readback": {"msg_svr_id": params["msg_svr_id"]},
|
|
"_channel_used": "server/frida",
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
response = asyncio.run(
|
|
unified.receive_red_packet(
|
|
unified.RedPacketReceiveRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
msg_svr_id="10435",
|
|
from_id="卡若",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
)
|
|
)
|
|
|
|
assert called["action"] == "receive_red_packet"
|
|
assert called["params"]["msg_svr_id"] == "10435"
|
|
assert response["code"] == 200
|
|
assert response["success"] is True
|
|
assert response["channel_used"] == "server/frida"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("handler", "action"),
|
|
[
|
|
(unified.receive_transfer_payment, "receive_transfer"),
|
|
(unified.reject_transfer_payment, "reject_transfer"),
|
|
],
|
|
)
|
|
def test_transfer_decision_defaults_to_verified_dry_run(
|
|
monkeypatch, handler, action
|
|
):
|
|
monkeypatch.setattr(
|
|
unified,
|
|
"_check_device_online",
|
|
lambda *args, **kwargs: pytest.fail("演练模式不应连接设备"),
|
|
)
|
|
response = asyncio.run(
|
|
handler(
|
|
unified.TransferDecisionRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
msg_svr_id="10436",
|
|
from_id="卡若",
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["code"] == 200
|
|
assert response["success"] is True
|
|
assert response["data"]["action"] == action
|
|
assert response["data"]["verified"] is True
|
|
assert response["data"]["dry_run"] is True
|
|
assert response["channel_used"] == "dry_run"
|
|
|
|
|
|
def test_reject_transfer_confirmed_result_uses_hook_only_channel(monkeypatch):
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: True)
|
|
|
|
async def fake_execute(device_id, platform, action, params, **kwargs):
|
|
assert action == "reject_transfer"
|
|
assert params["confirm"] is True
|
|
assert kwargs["hook_only"] is True
|
|
return {
|
|
"success": True,
|
|
"verified": True,
|
|
"status": "rejected",
|
|
"msg_svr_id": "10436",
|
|
"trace_id": "trace-reject-transfer",
|
|
"raw_rpc_receipt": {"err_type": 0, "err_code": 0},
|
|
"readback": {"msg_svr_id": "10436", "pay_subtype": "3"},
|
|
"_channel_used": "server/frida",
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
response = asyncio.run(
|
|
unified.reject_transfer_payment(
|
|
unified.TransferDecisionRequest(
|
|
device_id="device-1",
|
|
platform="wechat",
|
|
msg_svr_id="10436",
|
|
from_id="卡若",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
)
|
|
)
|
|
|
|
assert response["code"] == 200
|
|
assert response["success"] is True
|
|
assert response["data"]["status"] == "verified"
|
|
assert response["data"]["operation_status"] == "rejected"
|
|
assert response["channel_used"] == "server/frida"
|
|
|
|
|
|
def test_transfer_decisions_are_not_ui_only_actions():
|
|
from services.device_transport import WECHAT_U2_ONLY_ACTIONS
|
|
|
|
assert "send_red_packet" not in WECHAT_U2_ONLY_ACTIONS
|
|
assert "receive_red_packet" not in WECHAT_U2_ONLY_ACTIONS
|
|
assert "receive_transfer" not in WECHAT_U2_ONLY_ACTIONS
|
|
assert "reject_transfer" not in WECHAT_U2_ONLY_ACTIONS
|
|
|
|
|
|
def test_check_device_online_allows_server_frida_when_ws_offline(monkeypatch):
|
|
from services.device_transport import device_transport
|
|
|
|
monkeypatch.setattr(device_transport, "resolve_mode", lambda *args, **kwargs: "offline")
|
|
|
|
class DummyBridge:
|
|
@staticmethod
|
|
def enabled_for(device_id):
|
|
return device_id == "device-1"
|
|
|
|
import services.server_frida_bridge as bridge_module
|
|
|
|
monkeypatch.setattr(bridge_module, "server_frida_bridge", DummyBridge())
|
|
|
|
assert (
|
|
device_transport.check_device_online("device-1", "wechat", "receive_red_packet")
|
|
== "hook"
|
|
)
|
|
|
|
|
|
def test_transfer_hook_rpc_uses_internal_network_scene_only():
|
|
hook_path = (
|
|
Path(__file__).resolve().parents[1] / "agent" / "hook" / "wechat_hook_v2.js"
|
|
)
|
|
source = hook_path.read_text(encoding="utf-8")
|
|
start = source.index("function _executeTransferOperation")
|
|
end = source.index("// ============================================================", start)
|
|
implementation = source[start:end]
|
|
|
|
assert "com.tencent.mm.plugin.remittance.model.n0" in implementation
|
|
assert "/cgi-bin/mmpay-bin/transferoperation" in implementation
|
|
assert "cgi_func_id: 1691" in implementation
|
|
assert "_intentAction" not in implementation
|
|
assert "input tap" not in implementation
|
|
|
|
|
|
def test_red_packet_receive_hook_uses_internal_network_scenes_only():
|
|
hook_path = (
|
|
Path(__file__).resolve().parents[1] / "agent" / "hook" / "wechat_hook_v2.js"
|
|
)
|
|
source = hook_path.read_text(encoding="utf-8")
|
|
start = source.index("function _executeRedPacketReceive")
|
|
end = source.index("function _executeTransferOperation", start)
|
|
implementation = source[start:end]
|
|
|
|
assert "com.tencent.mm.plugin.luckymoney.model.u5" in implementation
|
|
assert "com.tencent.mm.plugin.luckymoney.model.o5" in implementation
|
|
assert "/cgi-bin/mmpay-bin/receivewxhb" in implementation
|
|
assert "/cgi-bin/mmpay-bin/openwxhb" in implementation
|
|
assert "cgi_func_id: 1581" in implementation
|
|
assert "cgi_func_id: 1685" in implementation
|
|
assert "_intentAction" not in implementation
|
|
assert "input tap" not in implementation
|
|
|
|
|
|
def test_red_packet_send_hook_uses_internal_prepare_scene_only():
|
|
hook_path = (
|
|
Path(__file__).resolve().parents[1] / "agent" / "hook" / "wechat_hook_v2.js"
|
|
)
|
|
source = hook_path.read_text(encoding="utf-8")
|
|
start = source.index("function _executeRedPacketPrepareSend")
|
|
end = source.index("function _executeTransferOperation", start)
|
|
implementation = source[start:end]
|
|
|
|
assert "com.tencent.mm.plugin.luckymoney.model.q5" in implementation
|
|
assert "/cgi-bin/mmpay-bin/requestwxhb" in implementation
|
|
assert "cgi_func_id: 1575" in implementation
|
|
assert "reqkey" in implementation
|
|
assert "send_msg_xml" in implementation
|
|
assert "_intentAction" not in implementation
|
|
assert "input tap" not in implementation
|
|
|
|
|
|
def test_hook_copies_are_identical_after_red_packet_update():
|
|
root = Path(__file__).resolve().parents[1]
|
|
canonical = (root / "agent" / "hook" / "wechat_hook_v2.js").read_bytes()
|
|
assert (root / "app" / "agent" / "hook" / "wechat_hook_v2.js").read_bytes() == canonical
|
|
assert (
|
|
root
|
|
/ "android-app"
|
|
/ "app"
|
|
/ "src"
|
|
/ "main"
|
|
/ "assets"
|
|
/ "wechat_hook_v2.js"
|
|
).read_bytes() == canonical
|
|
|
|
|
|
def test_message_list_online_check_uses_get_messages_action():
|
|
router_path = Path(__file__).resolve().parents[1] / "app" / "routers" / "unified.py"
|
|
source = router_path.read_text(encoding="utf-8")
|
|
start = source.index("async def get_messages(req: GetMessagesRequest)")
|
|
end = source.index('@router.post(\n "/message/sync-since"', start)
|
|
implementation = source[start:end]
|
|
|
|
assert '_check_device_online(req.device_id, req.platform.value, "get_messages")' in implementation
|
|
assert '"send_red_packet"' not in implementation
|
|
|
|
|
|
def test_incoming_payment_batch_reports_each_attempt_and_summary():
|
|
router_path = Path(__file__).resolve().parents[1] / "app" / "routers" / "unified.py"
|
|
source = router_path.read_text(encoding="utf-8")
|
|
start = source.index("async def receive_incoming_payment_batch")
|
|
end = source.index("# =============================================================================", start)
|
|
implementation = source[start:end]
|
|
|
|
assert '"receive_transfer"' in implementation
|
|
assert '"receive_red_packet"' in implementation
|
|
assert '"requested_count"' in implementation
|
|
assert '"success_count"' in implementation
|
|
assert '"failed_count"' in implementation
|
|
assert '"skipped_count"' in implementation
|
|
|
|
|
|
def test_wp_hl_06_payment_routes_and_aliases_registered():
|
|
from main import app
|
|
|
|
paths = app.openapi()["paths"]
|
|
expected = {
|
|
"/api/v3/payment/red-packet",
|
|
"/api/v3/payment/transfer",
|
|
"/api/v3/payment/transfer/receive",
|
|
"/api/v3/payment/receive-red-packet",
|
|
"/api/v3/payment/receive",
|
|
"/api/v3/payment/receive-payment",
|
|
"/api/v3/payment/payment-receive",
|
|
"/api/v3/payment/receive-incoming-batch",
|
|
}
|
|
assert expected.issubset(set(paths))
|
|
|
|
|
|
def test_wp_hl_06_transfer_defaults_to_dry_run_confirm_gate():
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
resp = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(device_id="device-1", to_id="test-user", amount="0.01")
|
|
))
|
|
assert resp["success"] is True
|
|
assert resp["channel_used"] == "dry_run"
|
|
assert resp["data"]["dry_run"] is True
|
|
assert resp["data"]["confirm_required"] is True
|
|
|
|
|
|
def test_wp_hl_06_transfer_accepts_payment_password_without_echo():
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
resp = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(
|
|
device_id="device-1",
|
|
to_id="test-user",
|
|
amount="0.01",
|
|
payment_password="123456",
|
|
)
|
|
))
|
|
text = str(resp)
|
|
assert resp["success"] is True
|
|
assert resp["data"]["password_present"] is True
|
|
assert "123456" not in text
|
|
|
|
|
|
def test_wp_hl_06_transfer_forwards_payment_password_masked(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "test-user")
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: "hook")
|
|
captured = {}
|
|
|
|
async def fake_execute(device_id, platform, action, params, **kwargs):
|
|
captured.update(params)
|
|
return {
|
|
"code": 202,
|
|
"success": False,
|
|
"verified": True,
|
|
"channel_used": "frida_rpc",
|
|
"trace_id": "trace-password",
|
|
"raw_rpc_receipt": {"ok": True},
|
|
"prepared": True,
|
|
"payment_confirm_required": True,
|
|
"status": "payment_confirm_required",
|
|
"error_code": "payment_confirm_required",
|
|
"readback": {"found": False},
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
resp = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(
|
|
device_id="device-password",
|
|
to_id="test-user",
|
|
amount="0.01",
|
|
pay_password="123456",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert captured["payment_password"] == "123456"
|
|
assert captured["password_present"] is True
|
|
assert resp["code"] == 202
|
|
assert "123456" not in str(resp)
|
|
|
|
|
|
def test_wp_hl_06_send_transfer_requires_password_before_rpc(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "test-user")
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not touch device")))
|
|
resp = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(
|
|
device_id="device-password-required",
|
|
to_id="test-user",
|
|
amount="0.01",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert resp["success"] is False
|
|
assert resp["code"] == 422
|
|
assert resp["data"]["error_code"] == "payment_password_required"
|
|
|
|
|
|
def test_wp_hl_06_payment_amount_guard_rejects_large_real_action(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "test-user")
|
|
resp = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(
|
|
device_id="device-1",
|
|
to_id="test-user",
|
|
amount="2.01",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert resp["success"] is False
|
|
assert resp["data"]["error_code"] == "payment_amount_out_of_range"
|
|
|
|
|
|
def test_wp_hl_06_payment_target_guard_requires_whitelist(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "test-user")
|
|
resp = asyncio.run(unified.send_red_packet(
|
|
unified.RedPacketRequest(
|
|
device_id="device-1",
|
|
platform=unified.Platform.WECHAT,
|
|
to_id="not-in-whitelist",
|
|
amount="0.01",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert resp["success"] is False
|
|
assert resp["code"] == 403
|
|
assert resp["data"]["error_code"] == "payment_test_target_required"
|
|
|
|
|
|
def test_wp_hl_06_payment_rate_limit_blocks_fourth_real_action(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "test-user")
|
|
monkeypatch.setenv("WP_PAYMENT_RATE_MAX_CALLS", "3")
|
|
monkeypatch.setenv("WP_PAYMENT_RATE_WINDOW_SECONDS", "60")
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: "hook")
|
|
|
|
async def fake_execute(*args, **kwargs):
|
|
return {
|
|
"code": 200,
|
|
"success": True,
|
|
"verified": True,
|
|
"channel_used": "frida_rpc",
|
|
"trace_id": "trace-rate",
|
|
"raw_rpc_receipt": {"ok": True},
|
|
"readback": {"ok": True},
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
for _ in range(3):
|
|
ok = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(
|
|
device_id="device-rate",
|
|
to_id="test-user",
|
|
amount="0.01",
|
|
payment_password="123456",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert ok["success"] is True
|
|
blocked = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(
|
|
device_id="device-rate",
|
|
to_id="test-user",
|
|
amount="0.01",
|
|
payment_password="123456",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert blocked["success"] is False
|
|
assert blocked["code"] == 429
|
|
assert blocked["data"]["error_code"] == "rate_limited"
|
|
|
|
|
|
def test_wp_hl_06_payment_success_requires_trace_raw_receipt_and_readback(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "test-user")
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: "hook")
|
|
|
|
async def fake_execute(*args, **kwargs):
|
|
return {
|
|
"code": 200,
|
|
"success": True,
|
|
"verified": True,
|
|
"channel_used": "frida_rpc",
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
resp = asyncio.run(unified.send_transfer(
|
|
unified.TransferRequest(
|
|
device_id="device-evidence",
|
|
to_id="test-user",
|
|
amount="0.01",
|
|
payment_password="123456",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert resp["success"] is False
|
|
assert resp["code"] == 503
|
|
assert resp["data"]["error_code"] == "payment_readback_missing"
|
|
assert set(resp["data"]["missing_evidence"]) >= {"trace_id", "raw_rpc_receipt", "message_or_ledger_readback"}
|
|
|
|
|
|
def test_wp_hl_06_payment_actions_have_no_u2_fallback():
|
|
from services.device_transport import WECHAT_U2_ONLY_ACTIONS
|
|
|
|
for action in [
|
|
"payment_receive",
|
|
"receive_payment",
|
|
"receive_red_packet",
|
|
"receive_transfer",
|
|
"send_red_packet",
|
|
"send_transfer",
|
|
"transfer",
|
|
]:
|
|
assert action not in WECHAT_U2_ONLY_ACTIONS
|
|
|
|
|
|
def test_wp_hl_06_transfer_hook_has_kinda_password_autofill_path():
|
|
hook_path = Path(__file__).resolve().parents[1] / "agent" / "hook" / "wechat_hook_v2.js"
|
|
source = hook_path.read_text(encoding="utf-8")
|
|
assert "_startKindaSnsTransferPay" in source
|
|
assert "KindaPwdInputViewImpl" in source
|
|
assert "setOnEndEnterPasswordCallback" in source
|
|
assert "inspectKindaPaymentAutofill" in source
|
|
assert "kinda_autofill_attempted" in source
|
|
|
|
|
|
def test_wp_hl_06_payment_hook_capabilities_are_explicit_frida_status():
|
|
hook_path = Path(__file__).resolve().parents[1] / "agent" / "hook" / "wechat_hook_v2.js"
|
|
source = hook_path.read_text(encoding="utf-8")
|
|
send_start = source.index("sendTransfer: function")
|
|
send_end = source.index("getWalletBalance: function", send_start)
|
|
receive_start = source.index("receivePayment: function", send_end)
|
|
receive_end = source.index("autoRegister: function", receive_start)
|
|
send_impl = source[send_start:send_end]
|
|
receive_impl = source[receive_start:receive_end]
|
|
|
|
assert "_executeTransferPrepareSend" in source
|
|
assert "wechat_internal_transfer_prepare" in send_impl
|
|
assert "send_transfer_scene_constructor_unresolved" in source
|
|
assert "com.tencent.mm.plugin.remittance.model.v" in source
|
|
assert "model_v_int_8strings_scene_user_fee_desc_feeType_receiver_request_ext" in source
|
|
assert "/cgi-bin/mmpay-bin/transfer" in source
|
|
assert "receive_payment_internal_rpc_not_ready" in receive_impl
|
|
assert "capability_unavailable" in receive_impl
|
|
assert "frida_rpc" in send_impl
|
|
assert "frida_rpc" in receive_impl
|
|
assert "_intentAction" not in send_impl
|
|
assert "_intentAction" not in receive_impl
|
|
|
|
|
|
def test_wp_hl_06_transfer_confirm_endpoint_dry_run_masks_password():
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
resp = asyncio.run(unified.confirm_prepared_transfer(
|
|
unified.TransferConfirmRequest(
|
|
device_id="device-1",
|
|
req_key="sns_tf_test_req_key",
|
|
to_id="test-user",
|
|
payment_password="123456",
|
|
)
|
|
))
|
|
assert resp["success"] is True
|
|
assert resp["channel_used"] == "dry_run"
|
|
assert resp["data"]["confirm_required"] is True
|
|
assert resp["data"]["password_present"] is True
|
|
assert "123456" not in str(resp)
|
|
|
|
|
|
def test_wp_hl_06_transfer_confirm_forwards_to_hook_without_echo(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setenv("WP_PAYMENT_TEST_TARGETS", "test-user")
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: "hook")
|
|
captured = {}
|
|
|
|
async def fake_execute(device_id, platform, action, params, **kwargs):
|
|
captured["action"] = action
|
|
captured.update(params)
|
|
return {
|
|
"code": 202,
|
|
"success": False,
|
|
"verified": True,
|
|
"channel_used": "frida_rpc",
|
|
"trace_id": "trace-confirm",
|
|
"raw_rpc_receipt": {"started": True},
|
|
"prepared": True,
|
|
"payment_confirm_required": True,
|
|
"status": "payment_confirm_required",
|
|
"error_code": "payment_confirm_required",
|
|
"readback": {"found": False},
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
resp = asyncio.run(unified.confirm_prepared_transfer(
|
|
unified.TransferConfirmRequest(
|
|
device_id="device-confirm",
|
|
req_key="sns_tf_test_req_key",
|
|
to_id="test-user",
|
|
transfer_id="tid",
|
|
transaction_id="txid",
|
|
pay_password="123456",
|
|
confirm=True,
|
|
dry_run=False,
|
|
)
|
|
))
|
|
assert captured["action"] == "confirm_prepared_transfer"
|
|
assert captured["payment_password"] == "123456"
|
|
assert captured["password_present"] is True
|
|
assert resp["code"] == 202
|
|
assert "123456" not in str(resp)
|
|
|
|
|
|
def test_wp_hl_06_transfer_confirm_hook_mapping_and_rpc_export():
|
|
from pathlib import Path
|
|
from agent.hook import hook_executor
|
|
|
|
assert hook_executor.ACTION_TO_RPC["confirm_prepared_transfer"] == "confirmPreparedTransfer"
|
|
assert hook_executor.ACTION_ALIASES["transfer_confirm"] == "confirm_prepared_transfer"
|
|
hook_path = Path(__file__).resolve().parents[1] / "agent" / "hook" / "wechat_hook_v2.js"
|
|
source = hook_path.read_text(encoding="utf-8")
|
|
assert "function _confirmPreparedTransfer" in source
|
|
assert "confirmPreparedTransfer: function" in source
|
|
assert "wechat_internal_transfer_kinda_confirm" in source
|
|
assert "confirm_prepared_transfer_kinda_autofill" in source
|
|
|
|
|
|
def test_wp_hl_06_transfer_readback_endpoint_is_read_only(monkeypatch):
|
|
import asyncio
|
|
from routers import unified
|
|
|
|
monkeypatch.setattr(unified, "_check_device_online", lambda *args, **kwargs: "hook")
|
|
captured = {}
|
|
|
|
async def fake_execute(device_id, platform, action, params, **kwargs):
|
|
captured.update({"action": action, "params": params, "kwargs": kwargs})
|
|
return {
|
|
"success": False,
|
|
"found": False,
|
|
"verified": True,
|
|
"stage": "outgoing_transfer_message_missing",
|
|
"trace_id": "trace-readback",
|
|
"raw_rpc_receipt": {"found": False},
|
|
"_channel_used": "server/frida",
|
|
}
|
|
|
|
monkeypatch.setattr(unified, "_execute_skill", fake_execute)
|
|
resp = asyncio.run(unified.readback_transfer(
|
|
unified.TransferReadbackRequest(
|
|
device_id="device-readback",
|
|
to_id="test-user",
|
|
req_key="req-1",
|
|
transfer_id="transfer-1",
|
|
transaction_id="txn-1",
|
|
)
|
|
))
|
|
assert resp["code"] == 202
|
|
assert resp["success"] is False
|
|
assert captured["action"] == "inspect_transfer_readback"
|
|
assert captured["kwargs"]["hook_only"] is True
|
|
assert "payment_password" not in captured["params"]
|