test_plugin.py

# -*- coding: utf-8 -*-
"""
astrbot_plugin_skland_llm v2.0 测试脚本
- mock 全部网络请求，验证签到/查询/签名/设备ID缓存逻辑
- 运行: python test_plugin.py
"""
import asyncio
import json
import os
import shutil
import sys
import tempfile
import types

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

PASS = 0
FAIL = 0

def check(name, cond, detail=""):
    global PASS, FAIL
    if cond:
        PASS += 1
        print(f"  [PASS] {name}")
    else:
        FAIL += 1
        print(f"  [FAIL] {name}  {detail}")

# ==================== mock httpx ====================
import httpx

class FakeResponse:
    def __init__(self, data):
        self._data = data
    def json(self):
        return self._data

class FakeClient:
    responses = []
    calls = []
    def __init__(self, timeout=30.0):
        pass
    async def get(self, url, headers=None):
        FakeClient.calls.append(("GET", url))
        return self._match(url)
    async def post(self, url, headers=None, json=None):
        FakeClient.calls.append(("POST", url, json))
        return self._match(url)
    async def aclose(self):
        pass
    def _match(self, url):
        for key, resp in FakeClient.responses:
            if key in url:
                return FakeResponse(resp)
        raise Exception(f"未 mock 的 URL: {url}")

httpx.AsyncClient = FakeClient

# ==================== 注入假 astrbot 模块 ====================
astrbot = types.ModuleType("astrbot")
api = types.ModuleType("astrbot.api")
event = types.ModuleType("astrbot.api.event")
star = types.ModuleType("astrbot.api.star")
from unittest.mock import MagicMock
event.filter = MagicMock()
event.filter.llm_tool = MagicMock(return_value=lambda f: f)
event.filter.command = MagicMock(return_value=lambda f: f)
event.AstrMessageEvent = MagicMock()
star.Context = MagicMock()

class FakeStar:
    def __init__(self, context=None, *args, **kwargs):
        self.context = context

star.Star = FakeStar
star.register = MagicMock(return_value=lambda cls: cls)
api.logger = MagicMock()
api.AstrBotConfig = dict
api.event = event
api.star = star
astrbot.api = api
sys.modules["astrbot"] = astrbot
sys.modules["astrbot.api"] = api
sys.modules["astrbot.api.event"] = event
sys.modules["astrbot.api.star"] = star
sys.modules["astrbot.api.event.filter"] = event.filter

# 临时工作目录（隔离 data 缓存）
tmpdir = tempfile.mkdtemp(prefix="skland_test_")
print(f"测试工作目录: {tmpdir}")

from skland_api import SklandAPI, _DID_CACHE_DIR, _DID_CACHE_FILE
import main as plugin_main
# 让插件 data 目录指向临时目录
plugin_main._DATA_DIR = os.path.join(tmpdir, "data")
plugin_main._STATUS_FILE = os.path.join(plugin_main._DATA_DIR, "sign_in_status.json")
# 设备缓存也指向临时目录（重新绑定 skland_api 内的变量）
import skland_api
skland_api._DID_CACHE_DIR = os.path.join(tmpdir, "data")
skland_api._DID_CACHE_FILE = os.path.join(skland_api._DID_CACHE_DIR, "device_id.txt")

# ==================== mock 网络响应 ====================
def setup_mock_responses():
    FakeClient.responses = [
        ("deviceprofile/v4", {"code": 1100, "detail": {"deviceId": "TESTDID999"}}),
        ("oauth2/v2/grant", {"status": 0, "data": {"code": "AUTHCODE123"}}),
        ("generate_cred_by_code", {"code": 0, "data": {"token": "CREDTOKEN123", "cred": "CRED123"}}),
        ("player/binding", {"code": 0, "data": {"list": [
            {"appCode": "arknights", "bindingList": [
                {"gameName": "明日方舟", "nickName": "测试博士", "channelName": "官服", "uid": "1001", "gameId": 1}
            ]},
            {"appCode": "endfield", "bindingList": [
                {"nickName": "测试管理员", "channelName": "官服", "roles": [
                    {"nickname": "角色A", "roleId": "2001", "serverId": "srv1"},
                    {"nickname": "角色B", "roleId": "2002", "serverId": "srv1"}
                ]}
            ]}
        ]}}),
        ("api/v1/game/attendance", {"code": 0, "data": {"awards": [{"resource": {"name": "龙门币"}, "count": 200}]}}),
        ("endfield/attendance", {"code": 0, "data": {
            "awardIds": [{"id": "a1"}],
            "resourceInfoMap": {"a1": {"name": "招募凭证", "count": 1}}
        }}),
    ]
    FakeClient.calls = []

# ==================== 测试用例 ====================

async def test_signature():
    print("\n[1] 签名算法确定性")
    api = SklandAPI()
    s1, h1 = api._generate_signature("tok", "/api/v1/test", "query=1", "DID")
    s2, h2 = api._generate_signature("tok", "/api/v1/test", "query=1", "DID")
    check("相同输入 → 相同签名", s1 == s2 and h1 == h2)
    check("签名长度 32 (md5)", len(s1) == 32)

def test_des():
    print("\n[2] DES/AES/tn 基础函数")
    api = SklandAPI()
    enc = api._des_encrypt(b"key12345", b"hello")
    check("DES 加密输出 bytes", isinstance(enc, bytes) and len(enc) % 8 == 0)
    aes = api._aes_encrypt(b"testdata", b"0123456789abcdef")
    check("AES 加密输出 hex", isinstance(aes, str) and len(aes) > 0)
    tn = api._get_tn({"b": "2", "a": 3})
    check("tn 生成可运行", isinstance(tn, str))

def test_device_id_cache():
    print("\n[3] 设备 ID 持久化")
    setup_mock_responses()
    api = SklandAPI()
    did1 = asyncio.run(api.get_device_id())
    check("首次生成 deviceId", did1 == "BTESTDID999")
    check("缓存文件已写入", os.path.exists(skland_api._DID_CACHE_FILE))
    FakeClient.calls = []
    api2 = SklandAPI()
    did2 = asyncio.run(api2.get_device_id())
    check("二次调用使用缓存（无网络请求）", did2 == did1 and len(FakeClient.calls) == 0)

async def test_full_sign_in():
    print("\n[4] 完整签到流程（mock 网络）")
    setup_mock_responses()
    api = SklandAPI()
    results, nickname = await api.do_full_sign_in("USERTOKEN")
    check("返回昵称", nickname == "测试博士", str(nickname))
    check("明日方舟签到成功", any(r.game == "明日方舟" and r.success for r in results))
    ak = [r for r in results if r.game == "明日方舟"][0]
    check("明日方舟奖励解析", "龙门币" in ak.awards[0], str(ak.awards))
    ef = [r for r in results if r.game == "终末地"]
    check("终末地双角色各签一次", len(ef) == 2, str(len(ef)))
    check("终末地奖励解析", all(r.success and r.awards for r in ef))
    # 验证请求次数
    post_urls = [c[1] for c in FakeClient.calls if c[0] == "POST"]
    check("终末地角色请求数=2", sum(1 for u in post_urls if "endfield" in u) == 2)

async def test_already_signed():
    print("\n[5] 重复签到错误处理")
    setup_mock_responses()
    # 覆盖明日方舟签到响应为"已签到"错误
    for i, (k, v) in enumerate(FakeClient.responses):
        if k == "api/v1/game/attendance":
            FakeClient.responses[i] = (k, {"code": 400, "message": "请勿重复签到"})
    api = SklandAPI()
    results, nickname = await api.do_full_sign_in("USERTOKEN")
    ak = [r for r in results if r.game == "明日方舟"][0]
    check("明日方舟标记为失败", not ak.success)
    check("错误信息包含已签到关键词", any(w in ak.error for w in ["重复", "已签到"]), ak.error)

def test_main_plugin():
    print("\n[6] 插件主逻辑（token 安全 + 只读查询）")
    # 实例化插件（config 无 token）
    plugin = plugin_main.SklandLLMPlugin(None, {})
    # token 缺失 → 错误提示
    try:
        plugin._get_token()
        check("无 token 时抛出异常", False)
    except ValueError as e:
        check("无 token 时抛出异常", "未配置 token" in str(e))
    # 配置 token
    plugin.config = {"token": "  MYTOKEN  "}
    check("token 去除空白", plugin._get_token() == "MYTOKEN")
    # 签到记录 + 只读查询
    from skland_api import SignInResult
    results = [
        SignInResult(success=True, game="明日方舟", nickname="测试博士", channel="官服", awards=["龙门币x200"]),
        SignInResult(success=True, game="终末地", nickname="角色A", channel="官服", awards=["招募凭证x1"]),
    ]
    plugin._record_sign_in(results, "测试博士")
    status = plugin._load_status()
    today = __import__("datetime").date.today().isoformat()
    check("签到记录已保存", today in status and len(status[today]["games"]) == 2)
    # 查询（不触发网络）
    FakeClient.calls = []
    async def run_query():
        return await plugin.skland_query_status(None)
    out = asyncio.run(run_query())
    check("查询返回已签到信息", "今日已签到" in out and "明日方舟" in out)
    check("查询未触发网络请求", len(FakeClient.calls) == 0)
    # 模拟新的一天（清空记录）→ 未签到提示
    plugin_main._STATUS_FILE = os.path.join(tmpdir, "data", "nonexist_" + str(os.getpid()) + ".json")
    async def run_query2():
        return await plugin.skland_query_status(None)
    out2 = asyncio.run(run_query2())
    check("无记录时提示未签到（只读）", "尚未签到" in out2)
    # 签到工具：config 无 token 时的提示
    plugin.config = {"token": ""}
    async def run_sign():
        return await plugin.skland_sign_in(None)
    out3 = asyncio.run(run_sign())
    check("签到工具无 token 提示配置", "未配置 token" in out3)

async def test_metadata_consistency():
    print("\n[7] 版本号/作者一致性")
    import yaml  # 可能没有 pyyaml，直接文本检查
    meta = open("metadata.yaml", encoding="utf-8").read()
    check("metadata 版本 2.0.0", "version: 2.0.0" in meta)
    check("metadata 作者 PRTS168", "author: PRTS168" in meta)
    src = open("main.py", encoding="utf-8").read()
    check("main.py 版本 2.0.0", "2.0.0" in src and "PRTS168" in src)
    check("main.py 无对话 token 参数", "token: str = None" not in src and "token=None" not in src)

async def test_sensitive_scan():
    print("\n[8] 敏感信息扫描")
    import re
    sensitive = []
    for fn in ["main.py", "skland_api.py", "README.md", "metadata.yaml", "_conf_schema.json", "requirements.txt"]:
        content = open(fn, encoding="utf-8", errors="ignore").read()
        if re.search(r"token\s*[:=]\s*['\"][A-Za-z0-9]{20,}", content):
            sensitive.append(f"{fn}: 疑似硬编码 token")
        if "13304850322" in content or "CMCC-502" in content:
            sensitive.append(f"{fn}: 疑似固件泄露信息")
    check("无硬编码敏感信息", not sensitive, str(sensitive))

if __name__ == "__main__":
    print("========== skland 插件 v2.0 测试 ==========")
    asyncio.run(test_signature())
    test_des()
    test_device_id_cache()
    asyncio.run(test_full_sign_in())
    asyncio.run(test_already_signed())
    test_main_plugin()
    asyncio.run(test_metadata_consistency())
    asyncio.run(test_sensitive_scan())
    print(f"\n========== 结果: {PASS} 通过, {FAIL} 失败 ==========")
    shutil.rmtree(tmpdir, ignore_errors=True)
    sys.exit(1 if FAIL else 0)