#!/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" "# campaign parsing: claimable -> claim flow mocked\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", len(lines) > 3 and lines[3] == "benefits: 1 claimable: 1", out) check("qoder already-claimed detection", len(lines) > 4 and lines[4] == "claimed_today: 1", 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() 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())