fnos-agent-checkin/test_backend.py

#!/usr/bin/env python3
"""Local dev tests for the AgentCheckin backend (no fnOS required).

Runs with AGENTCHECKIN_DEV=1 so all data lands in a throwaway devdata dir.
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile

BACKEND = os.path.join(os.path.dirname(os.path.abspath(__file__)), "AgentCheckin", "app", "backend")
DEVROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "AgentCheckin", "devdata")

ENV = dict(os.environ)
ENV["AGENTCHECKIN_DEV"] = "1"
ENV["PYTHONPATH"] = BACKEND

PASS = []
FAIL = []


def check(name, cond, extra=""):
    if cond:
        PASS.append(name)
        print("  PASS %s" % name)
    else:
        FAIL.append(name)
        print("  FAIL %s %s" % (name, extra))


def run_py(code, env=None):
    e = dict(ENV)
    if env:
        e.update(env)
    r = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True,
                       env=e, cwd=os.path.dirname(os.path.abspath(__file__)))
    return r.returncode, r.stdout, r.stderr


def cgi(method, uri, body=None, query=""):
    """Invoke api.py the way the CGI bridge does."""
    env = dict(ENV)
    env["REQUEST_METHOD"] = method
    env["REQUEST_URI"] = uri
    env["QUERY_STRING"] = query
    payload = b""
    if body is not None:
        payload = json.dumps(body).encode()
        env["CONTENT_LENGTH"] = str(len(payload))
    r = subprocess.run([sys.executable, os.path.join(BACKEND, "api.py")],
                       input=payload, capture_output=True, env=env)
    out = r.stdout.decode("utf-8", "replace")
    # strip CGI headers
    if "\r\n\r\n" in out:
        out = out.split("\r\n\r\n", 1)[1]
    try:
        return json.loads(out)
    except ValueError:
        return {"_raw": out, "_stderr": r.stderr.decode("utf-8", "replace")[-500:]}


def main():
    shutil.rmtree(DEVROOT, ignore_errors=True)
    print("== 1. common / storage ==")
    rc, out, err = run_py(
        "import common; print(common.DATA_DIR); print(common.today_str()); print(common.mask_token('abcdefghijklmnopqrstuvwxyz'))")
    check("common imports", rc == 0, err[-300:])
    if rc == 0:
        lines = out.strip().splitlines()
        check("dev data dir", lines[0].endswith("devdata"), lines)
        check("mask token", lines[2].startswith("abcdef") and lines[2].endswith("wxyz"), lines)

    print("== 2. engine account management ==")
    rc, out, err = run_py(
        "import engine, common;"
        "a = engine.add_trae('测试 TRAE', 'eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyUmVnaW9uIjp7InJlZ2lvbiI6IkNOIn19.sig');"
        "b = engine.add_workbuddy_manual('测试 WB', 'fake-access-token-123', 'fake-refresh', 'codebuddy.cn');"
        "print(a['id'], b['id']);"
        "print(len(engine.load_accounts()))")
    check("add accounts", rc == 0 and "2" in out, err[-300:])
    ids = out.split() if rc == 0 else []
    trae_id, wb_id = (ids[0], ids[1]) if len(ids) >= 2 else ("", "")

    print("== 3. dashboard masks tokens ==")
    rc, out, err = run_py(
        "import engine, json;"
        "d = engine.dashboard();"
        "s = json.dumps(d, ensure_ascii=False);"
        "print('fake-access-token-123' in s, 'fake-refresh' in s);"
        "print(len(d['accounts']))")
    check("dashboard no raw tokens", rc == 0 and out.startswith("False False"), out + err[-200:])
    check("dashboard account count", "2" in out, out)

    print("== 4. CGI API routing ==")
    r = cgi("GET", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/state")
    check("GET /api/state", r.get("today") is not None and len(r.get("accounts", [])) == 2, r)
    r = cgi("GET", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/accounts")
    check("GET /api/accounts", len(r.get("accounts", [])) == 2, r)
    r = cgi("POST", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/accounts/add_trae",
            {"name": "API 添加", "access_token": "tok2"})
    check("POST add_trae", r.get("ok") is True, r)
    r = cgi("POST", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/accounts/update",
            {"id": r["account"]["id"], "enabled": False})
    check("POST update account", r.get("ok") is True and r["account"]["enabled"] is False, r)
    r = cgi("POST", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/accounts/delete",
            {"id": r["account"]["id"]})
    check("POST delete account", r.get("ok") is True, r)
    r = cgi("POST", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/settings",
            {"schedule_times": ["08:00", "20:30"], "webhook_url": "", "notify_on": ["fail", "success"]})
    check("POST settings", r.get("ok") is True and r["settings"]["schedule_times"] == ["08:00", "20:30"], r)
    r = cgi("POST", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/settings",
            {"schedule_times": ["bad"]})
    check("POST settings rejects bad time", "error" in r, r)
    r = cgi("GET", "/cgi/ThirdParty/AgentCheckin/index.cgi/api/nope")
    check("unknown route 404", "error" in r, r)

    print("== 5. TRAE jwt region decode ==")
    rc, out, err = run_py(
        "import trae;"
        "tok='eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyUmVnaW9uIjp7InJlZ2lvbiI6IkNOIn19.sig';"
        "p=trae.jwt_payload(tok); print(p.get('userRegion'))")
    check("jwt decode", rc == 0 and "CN" in out, out + err[-200:])

    print("== 6. workbuddy token expiry logic ==")
    rc, out, err = run_py(
        "import workbuddy, time;"
        "print(workbuddy.token_expiring({'expiresAt': int(time.time()) + 3600}));"
        "print(workbuddy.token_expiring({'expiresAt': int(time.time()) + 30*86400}));"
        "print(workbuddy.token_expiring({'expiresAt': 0}));"
        "print(workbuddy.is_global({'domain':'workbuddy.ai'}), workbuddy.is_global({'domain':'codebuddy.cn'}))")
    check("token expiry", rc == 0 and out.split() == ["True", "False", "False", "True", "False"], out + err[-200:])

    print("== 7. live network: workbuddy oauth state endpoint ==")
    rc, out, err = run_py(
        "import workbuddy;"
        "state, url, err = workbuddy.begin_oauth();"
        "print(bool(state), bool(url), err)")
    if rc == 0 and out.startswith("True True None"):
        check("oauth begin (live)", True)
    else:
        check("oauth begin (live)", False, out + err[-300:])

    print("== 8. live network: trae status with bogus token -> no_session ==")
    rc, out, err = run_py(
        "import trae;"
        "acc={'id':'t1','auth':{'accessToken':'bogus.token.value'}};"
        "r=trae.checkin_once(acc); print(r['status'], r['message'][:40])")
    check("trae bogus token handled", rc == 0 and ("no_session" in out or "fail" in out or "retry_later" in out),
          out + err[-300:])

    print("== 9. live network: workbuddy status with bogus token ==")
    rc, out, err = run_py(
        "import engine;"
        "acc=engine.add_workbuddy_manual('bogus','bogus-token');"
        "r=engine._run_workbuddy(acc); print(r['status'], r['message'][:40])")
    check("wb bogus token handled", rc == 0 and r_status_ok(out), out + err[-300:])

    print("== 10. daemon tick (engine mocked, empty results) ==")
    mock = (
        "import daemon, engine, common\n"
        "calls = []\n"
        "def fake_run(account_id=None, reason='x'):\n"
        "    calls.append(reason)\n"
        "    st = common.load_state()\n"
        "    st.setdefault('today', {})[common.today_str()] = {}\n"  # empty results (no accounts)
        "    common.save_state(st)\n"
        "    return {}\n"
        "engine.run_all = fake_run\n"
        "daemon.tick()\n"
        "daemon.tick()\n"
        "daemon.tick()\n"
        "print(calls)\n"
    )
    rc, out, err = run_py(mock)
    # first tick = catch-up (marks passed slots); later ticks must be silent
    check("daemon catch-up runs once even with empty results",
          rc == 0 and out.strip() == "['startup-catchup']", out + err[-300:])

    print("== 11. webhook payload builders ==")
    rc, out, err = run_py(
        "import engine, json\n"
        "cases = [\n"
        "  ('auto', 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=x'),\n"
        "  ('auto', 'https://oapi.dingtalk.com/robot/send?access_token=x'),\n"
        "  ('auto', 'https://open.feishu.cn/open-apis/bot/v2/hook/x'),\n"
        "  ('auto', 'https://api.day.app/ABCKEY/'),\n"
        "  ('auto', 'https://example.com/hook'),\n"
        "  ('wecom', 'https://example.com/hook'),\n"
        "  ('bark', 'https://api.day.app/ABCKEY'),\n"
        "]\n"
        "for wt, u in cases:\n"
        "    url, body = engine.notify_payload(wt, u, 'TITLE', 'CONTENT')\n"
        "    print(url + '||' + (body.decode() if body else 'GET'))\n"
    )
    lines = out.strip().splitlines() if rc == 0 else []
    check("webhook: wecom payload", len(lines) > 0 and json.loads(lines[0].split("||")[1]) == {"msgtype": "text", "text": {"content": "TITLE\nCONTENT"}}, out + err[-200:])
    check("webhook: dingtalk payload", len(lines) > 1 and json.loads(lines[1].split("||")[1]) == {"msgtype": "text", "text": {"content": "TITLE\nCONTENT"}}, out)
    check("webhook: feishu payload", len(lines) > 2 and json.loads(lines[2].split("||")[1]) == {"msg_type": "text", "content": {"text": "TITLE\nCONTENT"}}, out)
    check("webhook: bark url-path GET", len(lines) > 3 and lines[3] == "https://api.day.app/ABCKEY/TITLE/CONTENT||GET", out)
    check("webhook: generic payload", len(lines) > 4 and json.loads(lines[4].split("||")[1]).get("title") == "TITLE", out)
    check("webhook: manual type override", len(lines) > 5 and json.loads(lines[5].split("||")[1]).get("msgtype") == "text", out)
    check("webhook: bark without trailing slash", len(lines) > 6 and lines[6] == "https://api.day.app/ABCKEY/TITLE/CONTENT||GET", out)

    print("== 12. qoder module (offline logic) ==")
    rc, out, err = run_py(
        "import qoder, json, base64, time\n"
        "# device headers with defaults + overrides\n"
        "acc = {'auth': {'accessToken': 'x'}, 'device': {'machineId': 'MID', 'version': '1.2.3'}}\n"
        "h = qoder.device_headers(acc)\n"
        "print('clientType=%s machineId=%s version=%s emptyToken=%r' % (h['Cosy-ClientType'], h['Cosy-MachineId'], h['Cosy-Version'], h['Cosy-MachineToken']))\n"
        "# jwt helpers\n"
        "payload = base64.urlsafe_b64encode(json.dumps({'exp': 9999999999, 'sub': 'u123'}).encode()).decode().rstrip('=')\n"
        "tok = 'h.' + payload + '.s'\n"
        "print('exp:', qoder.jwt_exp(tok), 'uid:', qoder.jwt_uid(tok))\n"
        "# global detection\n"
        "print('global:', qoder.is_global({'domain': 'openapi.qoder.sh'}), qoder.is_global({'domain': 'openapi.qoder.com.cn'}))\n"
        "# 国际版同一账号还挂着 VIEW_DETAILS 的订阅推广（claimStatus=CLAIMED，无 benefit）\n"
        "# —— 它不是签到活动，必须被排除\n"
        "intl = [{'actionType': 'VIEW_DETAILS', 'claimStatus': 'CLAIMED', 'campaignId': 'i1'}]\n"
        "print('intl_benefits:', len(qoder._benefits(intl)))\n"
        "# 每日轮次：actionType=CLAIM_BENEFIT、窗口开放\n"
        "now0 = int(time.time())\n"
        "daily = [{'actionType': 'CLAIM_BENEFIT', 'claimStatus': 'CLAIMABLE', 'campaignId': 'd1',\n"
        "          'startAt': now0 - 60, 'endAt': now0 + 600}]\n"
        "print('daily_benefits:', len(qoder._benefits(daily)))\n"
        "# 过期轮次要排除\n"
        "expired = [{'actionType': 'CLAIM_BENEFIT', 'claimStatus': 'CLAIMABLE', 'campaignId': 'd2',\n"
        "            'startAt': now0 - 600, 'endAt': now0 - 60}]\n"
        "print('expired_benefits:', len(qoder._benefits(expired)))\n"
        "# 国内版 schema\n"
        "campaigns = [{'actionType': 'CLAIM_BENEFIT', 'claimStatus': 'CLAIMABLE', 'campaignId': 'c1', 'benefit': {'amount': 100}}]\n"
        "b = qoder._benefits(campaigns)\n"
        "print('benefits:', len(b), 'claimable:', len([c for c in b if c.get('claimStatus') == 'CLAIMABLE']))\n"
        "# already-claimed detection\n"
        "now = int(time.time())\n"
        "c2 = [{'actionType': 'CLAIM_BENEFIT', 'claimStatus': 'CLAIMED', 'startAt': now - 100, 'endAt': now + 100}]\n"
        "bb = qoder._benefits(c2)\n"
        "print('claimed_today:', len([c for c in bb if c.get('claimStatus') == 'CLAIMED' and (c.get('startAt') or 0) <= now < (c.get('endAt') or 0)]))\n"
    )
    lines = out.splitlines() if rc == 0 else []
    check("qoder device headers", lines and lines[0] == "clientType=10 machineId=MID version=1.2.3 emptyToken=''", out + err[-200:])
    check("qoder jwt helpers", len(lines) > 1 and lines[1] == "exp: 9999999999 uid: u123", out)
    check("qoder global detection", len(lines) > 2 and lines[2] == "global: True False", out)
    check("qoder campaign parsing", "benefits: 1 claimable: 1" in out, out)
    intl_line = next((l for l in lines if l.startswith("intl_benefits:")), "")
    check("qoder intl: VIEW_DETAILS excluded", intl_line == "intl_benefits: 0", out)
    check("qoder daily CLAIM_BENEFIT kept", "daily_benefits: 1" in out, out)
    check("qoder expired round excluded", "expired_benefits: 0" in out, out)
    check("qoder already-claimed detection", "claimed_today: 1" in out, out)

    print("== 13. engine: add qoder account + token masking ==")
    rc, out, err = run_py(
        "import engine, json\n"
        "a = engine.add_qoder('Qoder 测试', 'eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ1MSJ9.sig', 'rt123', {'machineId': 'MID', 'machineToken': 'MTOK'})\n"
        "print(a['id'], a['platform'], a['device']['machineId'])\n"
        "d = engine.dashboard()\n"
        "s = json.dumps(d, ensure_ascii=False)\n"
        "print('masked:', 'eyJhbGciOiJSUzI1NiJ9' not in s and 'rt123' not in s)\n"
    )
    check("engine add qoder", rc == 0 and "qoder_" in out and "MID" in out, out + err[-200:])
    check("qoder token masked in dashboard", "masked: True" in out, out)

    print("== 14. umid module (offline) ==")
    rc, out, err = run_py(
        "import umid\n"
        "st = umid.status()\n"
        "print('supported:', st['supported'], '| arch:', st['cpu_arch'])\n"
        "print('installed:', st['installed'])\n"
        "import io, tarfile\n"
        "buf = io.BytesIO()\n"
        "with tarfile.open(fileobj=buf, mode='w') as tf:\n"
        "    info = tarfile.TarInfo('package/bundle/qodercli.js'); info.size = 5\n"
        "    tf.addfile(info, io.BytesIO(b'hello'))\n"
        "print('tar_extract_ok:', umid._extract_tar_entry(buf.getvalue(), 'package/bundle/qodercli.js') == b'hello')\n"
    )
    lines = out.splitlines() if rc == 0 else []
    check("umid: arch supported", len(lines) > 0 and "supported: True" in lines[0], out + err[-200:])
    check("umid: tar extraction", len(lines) > 2 and "tar_extract_ok: True" in lines[2], out)

    print()
    print("PASSED: %d  FAILED: %d" % (len(PASS), len(FAIL)))
    if FAIL:
        print("failures:", FAIL)
    shutil.rmtree(DEVROOT, ignore_errors=True)
    return 1 if FAIL else 0


def r_status_ok(out):
    return any(s in out for s in ("no_session", "fail", "ok", "already"))


if __name__ == "__main__":
    sys.exit(main())