fnos-agent-checkin/test_live_nas.py

#!/usr/bin/env python3
"""Full user-journey test against the REAL NAS installation."""
import json
import subprocess
import sys
import time
import urllib.request
import urllib.error

BASE = "https://192.168.31.10:5667/cgi/ThirdParty/AgentCheckin/index.cgi"
PASS, FAIL = [], []


def check(name, cond, extra=""):
    (PASS if cond else FAIL).append(name)
    print(("  PASS " if cond else "  FAIL ") + name + ("" if cond else "  " + str(extra)[:300]))


def cookie():
    r = subprocess.run(["bash", "/tmp/get_cookie.sh"], capture_output=True, text=True, timeout=60)
    return r.stdout.strip()


def api(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(BASE + path, data=data, method=method)
    if data:
        req.add_header("Content-Type", "application/json")
    import ssl
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    import http.cookiejar
    opener = urllib.request.build_opener(
        urllib.request.HTTPSHandler(context=ctx),
        urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
    # inject ost cookie manually
    req.add_header("Cookie", "ost=" + COOKIE)
    try:
        with opener.open(req, timeout=60) as resp:
            return resp.status, json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read().decode("utf-8"))


COOKIE = cookie()
print("cookie acquired:", COOKIE[:16], "...")

print("== 1. dashboard state ==")
st, d = api("GET", "/api/state")
check("GET /api/state", st == 200 and d.get("today") is not None, (st, d))
check("daemon alive", d.get("daemon_alive") is True, d.get("daemon_alive"))
check("next run scheduled", bool(d.get("next_run_at")), d.get("next_run_at"))
print("   daemon:", d.get("daemon"), "| next:", d.get("next_run_at"))

print("== 2. add TRAE account (fake token) ==")
st, d = api("POST", "/api/accounts/add_trae", {
    "name": "NAS 测试账号",
    "access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyUmVnaW9uIjp7InJlZ2lvbiI6IkNOIn19.sig",
})
check("add trae account", st == 200 and d.get("ok") is True, (st, d))
trae_id = d.get("account", {}).get("id")
check("token masked in response", "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9" not in json.dumps(d), d)

print("== 3. add WorkBuddy account (manual, fake token) ==")
st, d = api("POST", "/api/accounts/add_wb_manual", {
    "name": "NAS 测试 WB", "access_token": "fake-wb-token-xyz", "domain": "codebuddy.cn",
})
check("add wb account", st == 200 and d.get("ok") is True, (st, d))
wb_id = d.get("account", {}).get("id")

print("== 4. account list ==")
st, d = api("GET", "/api/accounts")
check("2 accounts listed", len(d.get("accounts", [])) == 2, d)
check("no raw tokens anywhere", "fake-wb-token-xyz" not in json.dumps(d), d)

print("== 5. trigger check-in (立即签到) ==")
st, d = api("POST", "/api/checkin", {})
check("checkin triggered", st == 200 and d.get("ok") is True, (st, d))
print("   mode:", d.get("mode"))

print("== 6. wait for daemon to execute ==")
deadline = time.time() + 60
results = {}
while time.time() < deadline:
    st, d = api("GET", "/api/state")
    tr = d.get("today_results") or {}
    if trae_id in tr and wb_id in tr:
        results = tr
        break
    time.sleep(2)
check("both accounts ran", bool(results), d.get("today_results"))
if results:
    for aid, r in results.items():
        print("   %s (%s): %s — %s" % (r.get("name"), r.get("platform"), r.get("status"), r.get("message")))
    check("trae handled gracefully (no_session expected)",
          results.get(trae_id, {}).get("status") in ("no_session", "fail", "ok", "already", "retry_later"),
          results.get(trae_id))
    check("wb handled gracefully",
          results.get(wb_id, {}).get("status") in ("no_session", "fail", "ok", "already", "retry_later"),
          results.get(wb_id))

print("== 7. logs ==")
st, d = api("GET", "/api/logs?lines=30")
check("logs returned", st == 200 and len(d.get("lines", [])) > 0, (st, d))
for line in d.get("lines", [])[-4:]:
    print("   ", line)

print("== 8. settings update ==")
st, d = api("POST", "/api/settings", {
    "schedule_times": ["00:05", "09:00", "12:30", "19:00", "22:00"],
    "webhook_url": "", "notify_on": ["fail", "success"],
})
check("settings saved", st == 200 and "22:00" in d["settings"]["schedule_times"], (st, d))
st, d = api("GET", "/api/state")
check("settings reflected in state", "22:00" in (d.get("settings") or {}).get("schedule_times", []), d.get("settings"))

print("== 9. account disable/enable/delete ==")
st, d = api("POST", "/api/accounts/update", {"id": wb_id, "enabled": False})
check("disable account", st == 200 and d["account"]["enabled"] is False, (st, d))
st, d = api("POST", "/api/accounts/update", {"id": wb_id, "enabled": True})
check("enable account", st == 200 and d["account"]["enabled"] is True, (st, d))
st, d = api("POST", "/api/accounts/delete", {"id": wb_id})
check("delete wb account", st == 200 and d.get("ok") is True, (st, d))
st, d = api("POST", "/api/accounts/delete", {"id": trae_id})
check("delete trae account", st == 200 and d.get("ok") is True, (st, d))
st, d = api("GET", "/api/accounts")
check("accounts empty again", len(d.get("accounts", [])) == 0, d)

print("== 10. static assets ==")
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
for path, marker in [("/", "Agent 签到助手"), ("/assets/app.js", "AgentCheckin"),
                     ("/assets/style.css", "--primary"), ("/assets/logo.svg", "<svg")]:
    req = urllib.request.Request(BASE + path)
    req.add_header("Cookie", "ost=" + COOKIE)
    with urllib.request.urlopen(req, timeout=20, context=ctx) as r:
        body = r.read().decode("utf-8", "replace")
    check("GET %s" % path, r.status == 200 and marker in body, body[:100])

print()
print("PASSED: %d  FAILED: %d" % (len(PASS), len(FAIL)))
if FAIL:
    print("failures:", FAIL)
sys.exit(1 if FAIL else 0)