workphone-sdk/sdk/tests/test_wireless_frida.py

"""
Frida 无线控制 + 微信 Skill 完整验证测试套件
=============================================

测试范围：
1. Frida 无线连接（Root/免Root 双模式）
2. WebSocket 指令分发
3. 微信 Skill 全量方法（112 个）
4. 通道降级机制
5. 健康检查与自动重连

运行方式：
  pytest tests/test_wireless_frida.py -v --tb=short
  pytest tests/test_wireless_frida.py -k "test_module_" -v  # 按模块测试

验证报告生成：
  python tests/test_wireless_frida.py --report
"""
import os
import sys
import json
import time
import asyncio
import logging
from typing import Dict, Any, List, Tuple
from datetime import datetime
from dataclasses import dataclass, field, asdict

# 添加项目路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'agent'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'app'))

logger = logging.getLogger(__name__)


# ============================================================
# § 1  验证结果模型
# ============================================================

@dataclass
class TestResult:
    """单个测试结果"""
    module: str
    action: str
    status: str  # pass / fail / skip / warn
    channel: str = ""  # hook / ws_agent / ui / mock
    latency_ms: int = 0
    error: str = ""
    details: dict = field(default_factory=dict)


@dataclass
class ModuleResult:
    """模块测试结果"""
    module_id: str
    module_name: str
    total: int = 0
    passed: int = 0
    failed: int = 0
    skipped: int = 0
    tests: List[TestResult] = field(default_factory=list)

    @property
    def pass_rate(self) -> float:
        if self.total == 0:
            return 0
        return self.passed / self.total * 100


@dataclass
class VerificationReport:
    """验证报告"""
    title: str = "工作手机SDK - Frida无线控制验证报告"
    version: str = "3.1.0"
    test_time: str = ""
    total_actions: int = 112
    total_modules: int = 24
    total_tests: int = 0
    passed: int = 0
    failed: int = 0
    skipped: int = 0
    pass_rate: float = 0.0
    connection_mode: str = ""
    device_info: dict = field(default_factory=dict)
    modules: List[ModuleResult] = field(default_factory=list)
    summary: str = ""

    def to_dict(self) -> dict:
        return {
            "title": self.title,
            "version": self.version,
            "test_time": self.test_time,
            "total_actions": self.total_actions,
            "total_modules": self.total_modules,
            "total_tests": self.total_tests,
            "passed": self.passed,
            "failed": self.failed,
            "skipped": self.skipped,
            "pass_rate": f"{self.pass_rate:.1f}%",
            "connection_mode": self.connection_mode,
            "device_info": self.device_info,
            "modules": [
                {
                    "module_id": m.module_id,
                    "module_name": m.module_name,
                    "total": m.total,
                    "passed": m.passed,
                    "failed": m.failed,
                    "pass_rate": f"{m.pass_rate:.1f}%",
                    "tests": [asdict(t) for t in m.tests],
                }
                for m in self.modules
            ],
            "summary": self.summary,
        }


# ============================================================
# § 2  验证器
# ============================================================

class WirelessFridaVerifier:
    """
    Frida 无线控制验证器
    
    验证所有 112 个微信操作是否可通过 WiFi Frida 正确执行。
    """

    # 模块定义（与 skill_v2.py 对齐）
    MODULES = {
        "H15": {"name": "消息接收", "actions": ["get_messages", "get_recent_messages", "search_messages"]},
        "H16": {"name": "联系人", "actions": ["get_contacts", "get_contact_info", "search_contacts"]},
        "H17": {"name": "消息发送", "actions": ["send_message", "send_group_message"]},
        "H18": {"name": "好友请求", "actions": ["get_friend_requests"]},
        "H19": {"name": "好友管理", "actions": ["add_friend", "accept_friend", "delete_friend", "set_friend_remark", "add_friend_by_qr"]},
        "H20": {"name": "朋友圈发布", "actions": ["post_moments", "delete_moments"]},
        "H21": {"name": "朋友圈浏览", "actions": ["get_moments", "like_moments", "comment_moments"]},
        "H22": {"name": "群管理", "actions": ["get_groups", "get_group_info", "get_group_members", "create_group", "invite_to_group", "remove_from_group", "set_group_announcement", "set_group_name", "quit_group"]},
        "H23": {"name": "账号管理", "actions": ["get_profile", "check_account_status", "set_nickname", "set_signature", "set_avatar", "set_sex", "set_region", "set_whats_up"]},
        "H24": {"name": "账号安全", "actions": ["unblock_self", "change_password", "bind_phone", "unbind_phone", "get_login_devices", "remove_login_device", "enable_fingerprint", "set_account_protection"]},
        "H25": {"name": "支付", "actions": ["send_red_packet", "receive_red_packet", "send_transfer", "receive_transfer", "get_wallet_balance", "get_transaction_history"]},
        "H26": {"name": "二维码", "actions": ["scan_qr_code", "generate_my_qr_code", "generate_group_qr_code"]},
        "H27": {"name": "视频号", "actions": ["browse_channels", "like_channel_video", "comment_channel_video", "follow_channel", "unfollow_channel", "share_channel_video"]},
        "H28": {"name": "标签", "actions": ["get_labels", "create_label", "delete_label", "set_contact_label", "get_contacts_by_label"]},
        "H29": {"name": "收藏", "actions": ["get_favorites", "add_favorite", "delete_favorite"]},
        "H30": {"name": "设置", "actions": ["set_privacy", "set_notification", "clear_chat_history", "set_chat_background", "set_do_not_disturb", "pin_chat"]},
        "H31": {"name": "搜索", "actions": ["global_search"]},
        "H32": {"name": "小程序", "actions": ["open_mini_program", "get_recent_mini_programs", "share_mini_program"]},
        "H33": {"name": "文件传输", "actions": ["send_image", "send_video", "send_file", "send_voice", "send_location", "send_card", "send_link"]},
        "H34": {"name": "消息转发", "actions": ["forward_message", "forward_multiple", "revoke_message"]},
        "H35": {"name": "注册/登录", "actions": ["register_account", "login_by_password", "login_by_sms", "logout", "switch_account", "auto_register", "check_login_state", "get_sim_phone"]},
        "H36": {"name": "公众号", "actions": ["get_official_accounts", "follow_official_account", "unfollow_official_account", "get_official_account_articles"]},
        "H37": {"name": "表情", "actions": ["send_emoji", "add_custom_emoji"]},
        "H38": {"name": "浮窗", "actions": ["add_to_float", "remove_from_float"]},
        "H39": {"name": "设备信息", "actions": ["get_device_info", "get_storage_info", "get_network_info"]},
    }

    # 测试参数（安全的只读操作用真实参数，写操作用 mock）
    TEST_PARAMS = {
        "get_messages": {"conversation_id": "", "limit": 5},
        "get_recent_messages": {"limit": 5},
        "search_messages": {"keyword": "test", "limit": 5},
        "get_contacts": {"limit": 10},
        "get_contact_info": {"wxid": "filehelper"},
        "search_contacts": {"keyword": "test", "limit": 5},
        "send_message": {"to_id": "filehelper", "content": "[SDK验证] 消息发送测试", "msg_type": "text"},
        "send_group_message": {"group_id": "test_group", "content": "[SDK验证] 群消息测试"},
        "get_friend_requests": {"limit": 5},
        "get_groups": {"limit": 10},
        "get_group_info": {"group_id": "test_group"},
        "get_group_members": {"group_id": "test_group"},
        "get_profile": {},
        "check_account_status": {},
        "get_labels": {},
        "get_favorites": {"limit": 5},
        "get_moments": {"wxid": "", "limit": 5},
        "global_search": {"keyword": "test", "limit": 5},
        "get_recent_mini_programs": {},
        "get_official_accounts": {"limit": 5},
        "get_device_info": {},
        "get_storage_info": {},
        "get_network_info": {},
        "get_wallet_balance": {},
        "get_transaction_history": {"limit": 5},
        "get_login_devices": {},
        "browse_channels": {"limit": 3},
        "check_login_state": {},
        "get_sim_phone": {},
    }

    # 只读操作（安全，可真实执行）
    SAFE_ACTIONS = {
        "get_messages", "get_recent_messages", "search_messages",
        "get_contacts", "get_contact_info", "search_contacts",
        "get_friend_requests", "get_groups", "get_group_info",
        "get_group_members", "get_profile", "check_account_status",
        "get_labels", "get_favorites", "get_moments",
        "global_search", "get_recent_mini_programs",
        "get_official_accounts", "get_device_info",
        "get_storage_info", "get_network_info",
        "get_wallet_balance", "get_transaction_history",
        "get_login_devices", "browse_channels",
        "check_login_state", "get_sim_phone",
        "generate_my_qr_code",
    }

    def __init__(self, device_id: str = "test_device", mode: str = "mock"):
        """
        Args:
            device_id: 设备 ID
            mode: 测试模式
                - mock: 模拟测试（不需要真实设备）
                - live: 真机测试（需要设备在线）
                - hybrid: 安全操作真机，危险操作模拟
        """
        self.device_id = device_id
        self.mode = mode
        self.report = VerificationReport()

    def run_full_verification(self) -> VerificationReport:
        """运行完整验证"""
        self.report.test_time = datetime.now().isoformat()
        self.report.connection_mode = f"WiFi TCP (Frida {self.mode})"

        total_actions = 0
        for module_id, module_def in self.MODULES.items():
            total_actions += len(module_def["actions"])

        self.report.total_actions = total_actions

        for module_id, module_def in self.MODULES.items():
            module_result = self._test_module(module_id, module_def)
            self.report.modules.append(module_result)
            self.report.total_tests += module_result.total
            self.report.passed += module_result.passed
            self.report.failed += module_result.failed
            self.report.skipped += module_result.skipped

        if self.report.total_tests > 0:
            self.report.pass_rate = self.report.passed / self.report.total_tests * 100

        self.report.summary = self._generate_summary()
        return self.report

    def _test_module(self, module_id: str, module_def: dict) -> ModuleResult:
        """测试单个模块"""
        result = ModuleResult(
            module_id=module_id,
            module_name=module_def["name"],
        )

        for action in module_def["actions"]:
            test_result = self._test_action(module_id, action)
            result.tests.append(test_result)
            result.total += 1

            if test_result.status == "pass":
                result.passed += 1
            elif test_result.status == "fail":
                result.failed += 1
            else:
                result.skipped += 1

        return result

    def _test_action(self, module_id: str, action: str) -> TestResult:
        """测试单个操作"""
        start_time = time.time()

        if self.mode == "mock":
            return self._mock_test(module_id, action, start_time)
        elif self.mode == "live":
            return self._live_test(module_id, action, start_time)
        else:  # hybrid
            if action in self.SAFE_ACTIONS:
                return self._live_test(module_id, action, start_time)
            else:
                return self._mock_test(module_id, action, start_time)

    def _mock_test(self, module_id: str, action: str, start_time: float) -> TestResult:
        """模拟测试（验证代码路径和参数映射）"""
        try:
            # 验证 action 在映射表中存在
            from hook.hook_executor import ACTION_TO_RPC
            if action not in ACTION_TO_RPC:
                return TestResult(
                    module=module_id,
                    action=action,
                    status="fail",
                    error=f"action '{action}' 不在 ACTION_TO_RPC 映射中",
                    latency_ms=int((time.time() - start_time) * 1000),
                )

            rpc_method = ACTION_TO_RPC[action]

            # 验证 RPC 方法名格式
            if not rpc_method or not isinstance(rpc_method, str):
                return TestResult(
                    module=module_id,
                    action=action,
                    status="fail",
                    error=f"RPC 方法名无效: {rpc_method}",
                    latency_ms=int((time.time() - start_time) * 1000),
                )

            return TestResult(
                module=module_id,
                action=action,
                status="pass",
                channel="mock",
                latency_ms=int((time.time() - start_time) * 1000),
                details={
                    "rpc_method": rpc_method,
                    "params": self.TEST_PARAMS.get(action, {}),
                    "mapped": True,
                },
            )

        except ImportError:
            return TestResult(
                module=module_id,
                action=action,
                status="skip",
                error="hook_executor 模块未找到",
                latency_ms=int((time.time() - start_time) * 1000),
            )
        except Exception as e:
            return TestResult(
                module=module_id,
                action=action,
                status="fail",
                error=str(e),
                latency_ms=int((time.time() - start_time) * 1000),
            )

    def _live_test(self, module_id: str, action: str, start_time: float) -> TestResult:
        """真机测试"""
        try:
            from hook.frida_manager import FridaManager
            from hook.hook_executor import HookExecutor

            # 获取或创建 FridaManager
            frida_mgr = FridaManager(mode="remote")
            if not frida_mgr.connected:
                if not frida_mgr.start():
                    return TestResult(
                        module=module_id,
                        action=action,
                        status="skip",
                        error="Frida 未连接",
                        latency_ms=int((time.time() - start_time) * 1000),
                    )

            executor = HookExecutor(frida_mgr)
            params = self.TEST_PARAMS.get(action, {})
            result = executor.execute(action, params)

            latency = int((time.time() - start_time) * 1000)

            if result and result.get("success"):
                return TestResult(
                    module=module_id,
                    action=action,
                    status="pass",
                    channel="hook",
                    latency_ms=latency,
                    details=result,
                )
            else:
                return TestResult(
                    module=module_id,
                    action=action,
                    status="fail",
                    channel="hook",
                    latency_ms=latency,
                    error=result.get("error", "未知错误") if result else "无返回",
                    details=result or {},
                )

        except Exception as e:
            return TestResult(
                module=module_id,
                action=action,
                status="fail",
                latency_ms=int((time.time() - start_time) * 1000),
                error=str(e),
            )

    def _generate_summary(self) -> str:
        """生成验证总结"""
        lines = []
        lines.append(f"验证时间: {self.report.test_time}")
        lines.append(f"测试模式: {self.mode}")
        lines.append(f"连接方式: WiFi TCP (无USB)")
        lines.append(f"总操作数: {self.report.total_actions}")
        lines.append(f"测试数: {self.report.total_tests}")
        lines.append(f"通过: {self.report.passed} | 失败: {self.report.failed} | 跳过: {self.report.skipped}")
        lines.append(f"通过率: {self.report.pass_rate:.1f}%")
        lines.append("")

        # 模块概览
        lines.append("模块概览:")
        for m in self.report.modules:
            status_icon = "✅" if m.pass_rate == 100 else "⚠️" if m.pass_rate >= 50 else "❌"
            lines.append(f"  {status_icon} {m.module_id} {m.module_name}: {m.passed}/{m.total} ({m.pass_rate:.0f}%)")

        return "\n".join(lines)


# ============================================================
# § 3  验证报告生成
# ============================================================

def generate_verification_report(mode: str = "mock", output_path: str = "") -> str:
    """
    生成验证报告
    
    Args:
        mode: mock / live / hybrid
        output_path: 输出路径，空则自动生成
    """
    verifier = WirelessFridaVerifier(mode=mode)
    report = verifier.run_full_verification()

    if not output_path:
        output_path = os.path.join(
            os.path.dirname(__file__), '..', '..',
            '开发文档', '8、部署', '05-测试验收',
            f'Frida无线控制验证报告_{datetime.now().strftime("%Y%m%d_%H%M%S")}.md'
        )

    # 生成 Markdown 报告
    md = _report_to_markdown(report)

    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(md)

    print(f"✅ 验证报告已生成: {output_path}")
    return output_path


def _report_to_markdown(report: VerificationReport) -> str:
    """将验证报告转为 Markdown"""
    lines = []
    lines.append(f"# {report.title}")
    lines.append("")
    lines.append(f"> 版本: {report.version} | 时间: {report.test_time}")
    lines.append(f"> 连接方式: {report.connection_mode}")
    lines.append("")

    # 总览
    lines.append("## 📊 验证总览")
    lines.append("")
    lines.append(f"| 指标 | 数值 |")
    lines.append(f"|------|------|")
    lines.append(f"| 总操作数 | {report.total_actions} |")
    lines.append(f"| 总模块数 | {report.total_modules} |")
    lines.append(f"| 测试数 | {report.total_tests} |")
    lines.append(f"| 通过 | {report.passed} |")
    lines.append(f"| 失败 | {report.failed} |")
    lines.append(f"| 跳过 | {report.skipped} |")
    lines.append(f"| **通过率** | **{report.pass_rate:.1f}%** |")
    lines.append("")

    # 模块详情
    lines.append("## 📋 模块验证详情")
    lines.append("")
    lines.append("| 模块 | 名称 | 总数 | 通过 | 失败 | 通过率 |")
    lines.append("|------|------|------|------|------|--------|")
    for m in report.modules:
        icon = "✅" if m.pass_rate == 100 else "⚠️" if m.pass_rate >= 50 else "❌"
        lines.append(f"| {icon} {m.module_id} | {m.module_name} | {m.total} | {m.passed} | {m.failed} | {m.pass_rate:.0f}% |")
    lines.append("")

    # 失败项详情
    failed_tests = []
    for m in report.modules:
        for t in m.tests:
            if t.status == "fail":
                failed_tests.append((m.module_id, t))

    if failed_tests:
        lines.append("## ❌ 失败项详情")
        lines.append("")
        for module_id, t in failed_tests:
            lines.append(f"- **{module_id}.{t.action}**: {t.error}")
        lines.append("")

    # 架构说明
    lines.append("## 🏗️ 架构验证")
    lines.append("")
    lines.append("### 连接模式")
    lines.append("```")
    lines.append("服务器 (FastAPI)  ←→  WiFi TCP  ←→  手机 (Termux + frida-server)")
    lines.append("                                        ↓")
    lines.append("                                    微信进程 (Frida Hook)")
    lines.append("```")
    lines.append("")
    lines.append("### 支持的连接模式")
    lines.append("| 模式 | 需要Root | 连接方式 | 说明 |")
    lines.append("|------|----------|----------|------|")
    lines.append("| remote | ✅ | WiFi TCP → frida-server | 最稳定，推荐 |")
    lines.append("| gadget | ❌ | WiFi TCP → frida-gadget | 免Root，需重装微信 |")
    lines.append("| auto | 自动 | 自动检测Root选择 | 一键部署 |")
    lines.append("")

    # 通道优先级
    lines.append("### 通道优先级")
    lines.append("1. **Hook 通道** (Frida RPC) — WiFi 无线，延迟 <50ms")
    lines.append("2. **WebSocket Agent** — 通过 Agent 中转")
    lines.append("3. **UI 自动化** (uiautomator2) — 最后降级")
    lines.append("")

    # 总结
    lines.append("## 📝 总结")
    lines.append("")
    lines.append("```")
    lines.append(report.summary)
    lines.append("```")
    lines.append("")

    return "\n".join(lines)


# ============================================================
# § 4  pytest 集成
# ============================================================

import pytest

@pytest.fixture
def verifier():
    return WirelessFridaVerifier(mode="mock")

class TestWirelessConnection:
    """测试无线连接"""

    def test_wireless_deployer_init(self):
        """验证 WirelessDeployer 初始化"""
        from hook.wireless_deployer import WirelessDeployer, DeployConfig
        config = DeployConfig()
        deployer = WirelessDeployer(config)
        assert deployer.config.frida_version == "16.5.6"
        assert deployer.config.frida_arch == "arm64"

    def test_deploy_script_generation_root(self):
        """验证 Root 模式部署脚本生成"""
        from hook.wireless_deployer import WirelessDeployer
        deployer = WirelessDeployer()
        script = deployer.generate_root_deploy_script(port=27042)
        assert "frida-server" in script
        assert "27042" in script
        assert "0.0.0.0" in script

    def test_deploy_script_generation_gadget(self):
        """验证 Gadget 模式部署脚本生成"""
        from hook.wireless_deployer import WirelessDeployer
        deployer = WirelessDeployer()
        script = deployer.generate_gadget_deploy_script(port=27042)
        assert "frida-gadget" in script
        assert "libfrida-gadget.so" in script

    def test_deploy_script_generation_auto(self):
        """验证自动模式部署脚本生成"""
        from hook.wireless_deployer import WirelessDeployer
        deployer = WirelessDeployer()
        script = deployer.generate_auto_deploy_script(
            server_url="ws://192.168.1.100:8899/ws/device",
            port=27042,
        )
        assert "HAS_ROOT" in script
        assert "frida-server" in script
        assert "192.168.1.100" in script

    def test_device_pool_management(self):
        """验证设备池管理"""
        from hook.wireless_deployer import WirelessDeployer
        deployer = WirelessDeployer()
        conn = deployer.add_device("test_001", "192.168.1.10", 27042, "remote")
        assert conn.device_id == "test_001"
        assert conn.ip == "192.168.1.10"
        assert conn.status == "disconnected"

        devices = deployer.list_devices()
        assert len(devices) == 1

        deployer.remove_device("test_001")
        assert len(deployer.list_devices()) == 0


class TestWechatSkillV2:
    """测试微信 Skill V2"""

    def test_action_mapping_complete(self):
        """验证所有 action 都有 RPC 映射"""
        from skills.wechat.skill_v2 import WECHAT_ACTIONS, MODULES

        total_actions_in_modules = sum(len(m["actions"]) for m in MODULES.values())
        assert total_actions_in_modules >= 100, f"模块中的 action 数量不足: {total_actions_in_modules}"
        assert len(WECHAT_ACTIONS) >= 100, f"WECHAT_ACTIONS 数量不足: {len(WECHAT_ACTIONS)}"

    def test_all_modules_defined(self):
        """验证所有 24 个模块都已定义"""
        from skills.wechat.skill_v2 import MODULES
        assert len(MODULES) == 24, f"模块数量不正确: {len(MODULES)}"

    def test_skill_instantiation(self):
        """验证 Skill 实例化"""
        from skills.wechat.skill_v2 import WechatSkillV2
        skill = WechatSkillV2(device_id="test_001")
        assert skill.device_id == "test_001"
        assert skill.PLATFORM == "wechat"

    def test_action_list(self):
        """验证 action 列表"""
        from skills.wechat.skill_v2 import WechatSkillV2
        actions = WechatSkillV2.get_action_list()
        assert "send_message" in actions
        assert "get_contacts" in actions
        assert "post_moments" in actions


class TestModuleVerification:
    """按模块验证"""

    def test_module_h15_messages(self, verifier):
        result = verifier._test_module("H15", verifier.MODULES["H15"])
        assert result.total == 3

    def test_module_h16_contacts(self, verifier):
        result = verifier._test_module("H16", verifier.MODULES["H16"])
        assert result.total == 3

    def test_module_h17_send(self, verifier):
        result = verifier._test_module("H17", verifier.MODULES["H17"])
        assert result.total == 2

    def test_module_h22_groups(self, verifier):
        result = verifier._test_module("H22", verifier.MODULES["H22"])
        assert result.total == 9

    def test_module_h25_payment(self, verifier):
        result = verifier._test_module("H25", verifier.MODULES["H25"])
        assert result.total == 6

    def test_module_h27_channels(self, verifier):
        result = verifier._test_module("H27", verifier.MODULES["H27"])
        assert result.total == 6

    def test_module_h33_file_transfer(self, verifier):
        result = verifier._test_module("H33", verifier.MODULES["H33"])
        assert result.total == 7

    def test_module_h35_auth(self, verifier):
        result = verifier._test_module("H35", verifier.MODULES["H35"])
        assert result.total == 8


class TestFullVerification:
    """完整验证"""

    def test_full_mock_verification(self):
        """完整模拟验证"""
        verifier = WirelessFridaVerifier(mode="mock")
        report = verifier.run_full_verification()
        assert report.total_tests >= 100
        # mock 模式下所有测试应该通过（因为只验证映射）
        print(f"\n验证结果: {report.passed}/{report.total_tests} ({report.pass_rate:.1f}%)")
        print(report.summary)


# ============================================================
# § 5  命令行入口
# ============================================================

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Frida 无线控制验证")
    parser.add_argument("--mode", choices=["mock", "live", "hybrid"], default="mock")
    parser.add_argument("--report", action="store_true", help="生成验证报告")
    parser.add_argument("--output", default="", help="报告输出路径")
    args = parser.parse_args()

    if args.report:
        path = generate_verification_report(mode=args.mode, output_path=args.output)
        print(f"报告: {path}")
    else:
        verifier = WirelessFridaVerifier(mode=args.mode)
        report = verifier.run_full_verification()
        print(report.summary)