iot-dboard/tests/test_multi_user.py

"""Multi-user system test suite (dual-token auth) — PostgreSQL edition.

Run:  python3 tests/test_multi_user.py
Requires a local PostgreSQL (default 127.0.0.1:5432, iot/iot@iot_dboard;
override with TEST_PG_HOST / TEST_PG_USER / TEST_PG_PASS / TEST_PG_DB).

Each test uses its own isolated PostgreSQL schema; the schema is dropped and
recreated at startup.

Covers: Root bootstrap, SQLite legacy import + schema-version gate,
admin registration (both modes), temp-account lifecycle, data isolation,
admin-only enforcement, re-auth (TOTP + passkey fallback), deletion,
kick devices (self account + Root kick-all), access/refresh token lifecycle,
SSE ticket scoping, rate limits, CSRF, per-account timezone/theme,
database administration (export / import / root-only / atomicity).
"""
import base64
import io
import json
import os
import shutil
import sqlite3
import sys
import tempfile
import time
import urllib.parse

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault("FLASK_SECRET_KEY", "test-secret")

import psycopg2
import pyotp

from backend.app import create_app
from backend.models import (
    init_db, _EMITTED_PASSWORDS, DB_VERSION, _read_db_version, get_db,
    get_auth_config, get_temp_password, get_users, get_user,
    is_temp_account, get_temp_totp_secret, get_active_user_ids,
    create_user, update_auth_config as uac,
)
from backend.events import EventBus
from backend import db_admin

PASS, FAIL = 0, 0

# ─── PostgreSQL test plumbing ─────────────────
PG_HOST = os.environ.get("TEST_PG_HOST", "127.0.0.1")
PG_USER = os.environ.get("TEST_PG_USER", "iot")
PG_PASS = os.environ.get("TEST_PG_PASS", "iot")
PG_DB = os.environ.get("TEST_PG_DB", "iot_dboard")


def _dsn(schema: str) -> str:
    """DSN for one isolated schema (search_path pinned to it)."""
    opts = urllib.parse.quote(f"-csearch_path={schema}")
    return f"postgresql://{PG_USER}:{PG_PASS}@{PG_HOST}/{PG_DB}?options={opts}"


def reset_schema(schema: str):
    conn = psycopg2.connect(f"postgresql://{PG_USER}:{PG_PASS}@{PG_HOST}/{PG_DB}")
    conn.autocommit = True
    cur = conn.cursor()
    cur.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
    cur.execute(f'CREATE SCHEMA "{schema}"')
    cur.close()
    conn.close()


class TClient:
    """Flask test client wrapper that attaches the access token to each request.
    Refresh-token cookies are managed automatically by the underlying client."""

    def __init__(self, client, access=None):
        self._c = client
        self.access = access

    def set_access(self, access):
        self.access = access

    def _h(self, headers):
        headers = dict(headers or {})
        if self.access and "Authorization" not in headers:
            headers["Authorization"] = "Bearer " + self.access
        return headers

    def get(self, *a, **kw):
        kw["headers"] = self._h(kw.pop("headers", None))
        return self._c.get(*a, **kw)

    def post(self, *a, **kw):
        kw["headers"] = self._h(kw.pop("headers", None))
        return self._c.post(*a, **kw)

    def put(self, *a, **kw):
        kw["headers"] = self._h(kw.pop("headers", None))
        return self._c.put(*a, **kw)

    def delete(self, *a, **kw):
        kw["headers"] = self._h(kw.pop("headers", None))
        return self._c.delete(*a, **kw)

    def patch(self, *a, **kw):
        kw["headers"] = self._h(kw.pop("headers", None))
        return self._c.patch(*a, **kw)


def _groups_for(db):
    conn = get_db(db)
    n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE user_id=0").fetchone()[0]
    conn.close()
    return n


def _count(db, table, user_id):
    conn = get_db(db)
    n = conn.execute(f"SELECT COUNT(*) FROM {table} WHERE user_id=?", (user_id,)).fetchone()[0]
    conn.close()
    return n


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


def fresh_app(schema):
    reset_schema(schema)
    db = _dsn(schema)
    app = create_app(db_path=db)
    return app, db


def do_login(client, code, view="control"):
    """Login on a raw test client; returns (TClient-with-access, login_response)."""
    r = client.post("/api/auth/login", json={"code": code, "view": view})
    assert r.status_code == 200, r.get_json()
    d = r.get_json()
    return TClient(client, access=d.get("access_token")), d


def setup_totp(client, secret=None):
    """Complete the temp→TOTP MFA setup flow for the current temp-pending session.
    `client` may be a TClient (its access token is updated after finalize)."""
    r = client.post("/api/auth/totp/setup")
    if r.status_code != 200:
        return None, r
    secret = r.get_json()["secret"]
    code = pyotp.TOTP(secret).now()
    r2 = client.post("/api/auth/totp/verify", json={"code": code})
    r3 = client.post("/api/auth/mfa/complete")
    if hasattr(client, "set_access") and r3.status_code == 200 and r3.get_json().get("access_token"):
        client.set_access(r3.get_json()["access_token"])
    return secret, r2


def root_client(app, db, setup_mfa=True):
    """Return a fully-authenticated Root TClient (TOTP configured)."""
    c = app.test_client()
    pw = get_temp_password(db, 0)
    tc, _ = do_login(c, pw)
    if setup_mfa:
        secret, _ = setup_totp(tc)
        assert secret, "root TOTP setup failed"
        st = tc.get("/api/auth/status").get_json()
        assert st["authenticated"], st
    return tc


def reauth_totp(client, secret):
    code = pyotp.TOTP(secret).now()
    return client.post("/api/auth/reauth", json={"code": code})


def b64url(hexstr):
    return base64.urlsafe_b64encode(bytes.fromhex(hexstr)).rstrip(b"=").decode()


print("=" * 60)
print("TEST 1: Fresh DB bootstrap → Root user + temp password + MFA setup")
print("=" * 60)
app, db = fresh_app("t1")
_EMITTED_PASSWORDS.clear()
init_db(db)
check("Root user exists with id 0", get_user(0, db) is not None)
root = get_user(0, db)
check("Root role is admin", root["role"] == "admin" and root["remark"] == "Root")
pw = get_temp_password(db, 0)
check("Root temp password generated", bool(pw))
r = app.test_client().post("/api/auth/login", json={"code": pw, "view": "control"})
check("Temp-password login → temp_pending", r.status_code == 200 and r.get_json().get("must_configure_mfa"))
check("Login returns access token", bool(r.get_json().get("access_token")))
check("Refresh cookie set", "refresh_token" in r.headers.get("Set-Cookie", ""))
c, _ = do_login(app.test_client(), pw)
st = c.get("/api/auth/status").get_json()
check("Status reports temp_pending", st["temp_pending"] is True)
secret, _ = setup_totp(c)
st = c.get("/api/auth/status").get_json()
check("Root fully authenticated after TOTP setup", st["authenticated"] is True)
check("Root is admin in status", st["user"]["is_admin"] is True)
check("Root remark visible (not id)", st["user"]["remark"] == "Root")
# TOTP login path
c2, _ = do_login(app.test_client(), pyotp.TOTP(secret).now())
check("Root TOTP login works", bool(c2.access)
      and not c2.get("/api/auth/status").get_json().get("temp_pending"))
# wrong code
r = app.test_client().post("/api/auth/login", json={"code": "000000", "view": "control"})
check("Wrong code rejected", r.status_code == 401)
# no/bad token → 401
check("No token → 401", app.test_client().get("/api/theme").status_code == 401)
check("Bad token → 401", app.test_client().get("/api/theme", headers={"Authorization": "Bearer nope"}).status_code == 401)

# datetime('now') 翻译正确性：PG to_char 必须用 HH24/MI（HH 是 12 小时制、MM 是月份）
import re as _re
from backend.pg_compat import _translate_sql
_tr, _ = _translate_sql("UPDATE t SET ts=datetime('now') WHERE id=?")
check("datetime('now') 翻译为 HH24:MI", "HH24:MI:SS" in _tr and "HH:MM:SS" not in _tr)
conn = get_db(db)
_trow = conn.execute("SELECT to_char(now(), 'YYYY-MM-DD HH24:MI:SS') AS t").fetchone()
check("to_char HH24:MI 输出格式正确", bool(_re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", _trow["t"])))
conn.close()

print()
print("=" * 60)
print("TEST 2: Legacy SQLite import → data becomes Root's (db_admin)")
print("=" * 60)
# Build a legacy single-user database (old schema: no users/user_id, pin-only auth)
legacy_fd, legacy = tempfile.mkstemp(suffix=".db")
with os.fdopen(legacy_fd, "wb"):
    pass
_lc = sqlite3.connect(legacy)
_lc.executescript("""
CREATE TABLE calendar_groups (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, color TEXT DEFAULT '#4A90D9', enabled INTEGER DEFAULT 1, created_at TEXT DEFAULT (datetime('now')));
CREATE TABLE calendar_sources (id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER NOT NULL, type TEXT NOT NULL, name TEXT NOT NULL, url TEXT, ics_data TEXT, last_sync TEXT);
CREATE TABLE calendar_events (id INTEGER PRIMARY KEY AUTOINCREMENT, source_id INTEGER NOT NULL, uid TEXT, summary TEXT, description TEXT, location TEXT, dtstart TEXT, dtend TEXT, is_all_day INTEGER DEFAULT 0);
CREATE TABLE reminders (id INTEGER PRIMARY KEY AUTOINCREMENT, calendar_group_id INTEGER NOT NULL, minutes_before INTEGER DEFAULT 15, enabled INTEGER DEFAULT 1);
CREATE TABLE alarms (id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT DEFAULT '', time TEXT NOT NULL, days TEXT DEFAULT '1,2,3,4,5,6,7', enabled INTEGER DEFAULT 1, sound TEXT DEFAULT 'default', once INTEGER DEFAULT 0);
CREATE TABLE auth_config (id INTEGER PRIMARY KEY AUTOINCREMENT, pin TEXT DEFAULT '');
CREATE TABLE clock_config (id INTEGER PRIMARY KEY AUTOINCREMENT, city TEXT DEFAULT 'Beijing', timezone TEXT DEFAULT 'Asia/Shanghai', is_primary INTEGER DEFAULT 1);
CREATE TABLE camera_config (id INTEGER PRIMARY KEY AUTOINCREMENT, enabled INTEGER DEFAULT 0);
""")
_lc.execute("INSERT INTO calendar_groups (name) VALUES ('Legacy Group 1')")
_lc.execute("INSERT INTO calendar_groups (name) VALUES ('Legacy Group 2')")
_lc.execute("INSERT INTO calendar_groups (name) VALUES ('Legacy Group 3')")
_lc.execute("INSERT INTO alarms (label, time) VALUES ('legacy alarm', '07:30')")
_lc.execute("INSERT INTO auth_config (id, pin) VALUES (1, '0411')")
_lc.execute("INSERT INTO clock_config (city, timezone, is_primary) VALUES ('Beijing', 'Asia/Shanghai', 1)")
_lc.commit()
_lc.close()
legacy_bytes = open(legacy, "rb").read()
os.unlink(legacy)

reset_schema("t2")
db2 = _dsn("t2")
init_db(db2)
check("Legacy file detected as sqlite", db_admin.detect_format(legacy_bytes) == "sqlite")
stats = db_admin.import_sqlite(db2, legacy_bytes)
check("Import succeeds", stats is not None and stats.get("calendar_groups") == 3)
check("Root created after legacy import", get_user(0, db2) is not None)
check("Legacy calendar groups → Root (user 0)", _groups_for(db2) >= 3)
check("Legacy alarms migrated to user 0", _count(db2, "alarms", 0) >= 1)
check("Root gets a fresh temp password (legacy had only PIN)", bool(get_temp_password(db2, 0)))
conn = get_db(db2)
tables = [r["table_name"] for r in conn.execute(
    "SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema()").fetchall()]
conn.close()
check("camera_config table dropped", "camera_config" not in tables)
check("users table exists", "users" in tables)
check("meta table exists", "meta" in tables)
check("auth_tokens table exists", "auth_tokens" in tables)
check("Legacy DB stamped with program version", _read_db_version(get_db(db2)) == DB_VERSION)
get_db(db2).close()

print()
print("=" * 60)
print("TEST 3: Schema-version gate (same→skip, newer→abort, lower→migrate+stamp)")
print("=" * 60)
reset_schema("t3")
db3 = _dsn("t3")
init_db(db3)
init_db(db3)
check("Same-version init idempotent", _read_db_version(get_db(db3)) == DB_VERSION)
get_db(db3).close()

# user data preserved across a version-mismatch migration (simulate future bump)
db3b = _dsn("t3b")
reset_schema("t3b")
init_db(db3b)
uid = create_user("Round Trip", db_path=db3b)
conn = get_db(db3b)
conn.execute("INSERT INTO alarms (user_id, label, time) VALUES (?, 'test', '07:00')", (uid,))
conn.commit()
conn.close()
# simulate an old DB (version 0 → migration path) by rewriting meta
conn = get_db(db3b)
conn.execute("UPDATE meta SET version=0 WHERE id=1")
conn.commit()
conn.close()
init_db(db3b)
users = get_users(db3b)
check("User preserved through migration", any(u["id"] == uid and u["remark"] == "Round Trip" for u in users))
check("User data preserved through migration", _count(db3b, "alarms", uid) == 1)
check("Migrated DB re-stamped to program version", _read_db_version(get_db(db3b)) == DB_VERSION)
get_db(db3b).close()

# missing-column migration: drop a column, downgrade version, init adds it back
reset_schema("t3c")
db3c = _dsn("t3c")
init_db(db3c)
conn = get_db(db3c)
conn.execute("ALTER TABLE users DROP COLUMN theme")
conn.execute("UPDATE meta SET version=1 WHERE id=1")
conn.commit()
conn.close()
init_db(db3c)
conn = get_db(db3c)
theme_cols = conn.execute(
    "SELECT column_name FROM information_schema.columns "
    "WHERE table_schema = current_schema() AND table_name='users' AND column_name='theme'").fetchall()
conn.close()
check("Missing column re-added by migration", len(theme_cols) == 1)
check("Migrated version stamped", _read_db_version(get_db(db3c)) == DB_VERSION)
get_db(db3c).close()

# a DB newer than the program → abort with SystemExit
reset_schema("t3d")
db3d = _dsn("t3d")
init_db(db3d)
conn = get_db(db3d)
conn.execute(f"UPDATE meta SET version={DB_VERSION + 99} WHERE id=1")
conn.commit()
conn.close()
try:
    init_db(db3d)
    check("Newer DB aborts startup", False, "no SystemExit raised")
except SystemExit:
    check("Newer DB aborts startup", True)

# legacy timestamp DEFAULT repair (old build used to_char HH:MM → wrong 12h/month)
reset_schema("t3e")
db3e = _dsn("t3e")
init_db(db3e)
conn = get_db(db3e)
conn.execute("ALTER TABLE users ALTER COLUMN created_at SET DEFAULT to_char(now(), 'YYYY-MM-DD HH:MM:SS')")
conn.commit()
conn.close()
init_db(db3e)  # same version → fast path must still repair the DEFAULT
conn = get_db(db3e)
dflt = conn.execute(
    "SELECT column_default FROM information_schema.columns "
    "WHERE table_schema=current_schema() AND table_name='users' AND column_name='created_at'"
).fetchone()["column_default"]
conn.close()
check("Legacy timestamp DEFAULT repaired (HH24:MI)", "HH24:MI:SS" in (dflt or ""))

print()
print("=" * 60)
print("TEST 4: MFA-configured DB → no new temp password")
print("=" * 60)
app4, db4 = fresh_app("t4")
c4 = root_client(app4, db4)
_EMITTED_PASSWORDS.clear()
init_db(db4)  # same version → fast path, defaults intact
check("No temp password for MFA-configured root", get_temp_password(db4, 0) == "")
cfg = get_auth_config(db4, 0)
check("must_configure_mfa cleared", cfg["must_configure_mfa"] == 0)

print()
print("=" * 60)
print("TEST 5: Admin registration (admin-fills) + TOTP provisioning + login")
print("=" * 60)
app5, db5 = fresh_app("t5")
root = root_client(app5, db5)
root_secret = get_auth_config(db5, 0)["totp_secret"]
# Reauth required first
r = root.post("/api/admin/users", json={"remark": "User A", "mode": "admin_fills"})
check("Admin op without reauth → 403", r.status_code == 403 and r.get_json().get("reauth_required"))
rr = reauth_totp(root, root_secret)
check("TOTP reauth succeeds", rr.status_code == 200 and rr.get_json()["ok"])
r = root.post("/api/admin/users", json={"remark": "User A", "mode": "admin_fills"})
check("Admin creates user (admin-fills)", r.status_code == 201, str(r.get_json()))
user_a_id = r.get_json()["user"]["id"]
check("New user is not admin", r.get_json()["user"]["is_admin"] is False)
# default remark = id when none given
r2 = root.post("/api/admin/users", json={"mode": "admin_fills"})
check("Remark defaults to user id", r2.status_code == 201 and r2.get_json()["user"]["remark"] == str(r2.get_json()["user"]["id"]))
user_b_id = r2.get_json()["user"]["id"]
# Provision TOTP for user A
r = root.post(f"/api/admin/users/{user_a_id}/totp/begin")
check("Admin provisions TOTP secret", r.status_code == 200 and r.get_json().get("secret"))
secret_a = r.get_json()["secret"]
r = root.post(f"/api/admin/users/{user_a_id}/totp/verify", json={"code": pyotp.TOTP(secret_a).now()})
check("Admin enables TOTP after verify", r.status_code == 200)
# Provision TOTP for user B too (needed for the isolation test)
r = root.post(f"/api/admin/users/{user_b_id}/totp/begin")
secret_b = r.get_json()["secret"]
root.post(f"/api/admin/users/{user_b_id}/totp/verify", json={"code": pyotp.TOTP(secret_b).now()})
# User A logs in
ua, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
check("User A TOTP login", bool(ua.access))
st = ua.get("/api/auth/status").get_json()
check("User A authenticated, not admin", st["authenticated"] and st["user"]["is_admin"] is False)
check("User A remark correct", st["user"]["remark"] == "User A")
# User A cannot access admin APIs
r = ua.get("/api/admin/users")
check("Regular user blocked from admin list", r.status_code == 403)
r = ua.post("/api/admin/users", json={"remark": "x", "mode": "admin_fills"})
check("Regular user blocked from user creation", r.status_code == 403)

print()
print("=" * 60)
print("TEST 6: Temp account lifecycle (user-fills mode)")
print("=" * 60)
r = root.post("/api/admin/users", json={"remark": "Temp User", "mode": "user_fills"})
check("Admin creates temp account", r.status_code == 201 and r.get_json()["is_temp"])
temp_id = r.get_json()["user"]["id"]
temp_secret = get_temp_totp_secret(temp_id, db5)
check("Temp TOTP secret stored server-side", bool(temp_secret))
r = root.get(f"/api/admin/users/{temp_id}/totp")
check("Admin sees real-time temp TOTP", r.status_code == 200 and r.get_json()["code"] == pyotp.TOTP(temp_secret).now())
# ensure no secret leakage in admin list
users_json = root.get("/api/admin/users").get_json()
tu = next(u for u in users_json if u["id"] == temp_id)
check("Admin list exposes no secrets", "temp_totp_secret" not in tu and "totp_secret" not in tu and tu["is_temp"])
# user logs in with the temp TOTP
tu_client, _ = do_login(app5.test_client(), pyotp.TOTP(temp_secret).now())
check("Temp user login via temp TOTP → must_configure_mfa", bool(tu_client.access))
st = tu_client.get("/api/auth/status").get_json()
check("Temp user forced to pending", st["temp_pending"] is True and st["user"]["is_temp"])
# user configures their OWN TOTP → account finalized (new token pair issued)
r = tu_client.post("/api/auth/totp/setup")
new_secret = r.get_json()["secret"]
r = tu_client.post("/api/auth/totp/verify", json={"code": pyotp.TOTP(new_secret).now()})
check("Temp account finalized on TOTP verify", r.status_code == 200 and r.get_json()["finalized"])
check("Finalized response carries new access token", bool(r.get_json().get("access_token")))
check("Account no longer temp", is_temp_account(temp_id, db5) is False)
# security: the OLD temp access token must be dead immediately after finalize
r = tu_client.post("/api/auth/mfa/complete")
check("Old temp token dead after finalize (401)", r.status_code == 401)
conn = get_db(db5)
n_temp = conn.execute("SELECT COUNT(*) FROM auth_tokens WHERE user_id=? AND temp=1", (temp_id,)).fetchone()[0]
conn.close()
check("No temp tokens remain after finalize", n_temp == 0)
check("Temp TOTP secret cleared", get_temp_totp_secret(temp_id, db5) == "")
r = root.get(f"/api/admin/users/{temp_id}/totp")
check("Admin can no longer see temp TOTP", r.status_code == 400)
# old temp TOTP no longer valid
r = app5.test_client().post("/api/auth/login", json={"code": pyotp.TOTP(temp_secret).now(), "view": "control"})
check("Old temp TOTP invalidated", r.status_code == 401)
# new TOTP works
r = app5.test_client().post("/api/auth/login", json={"code": pyotp.TOTP(new_secret).now(), "view": "control"})
check("New TOTP login works", r.status_code == 200)

print()
print("=" * 60)
print("TEST 7: Data isolation between users")
print("=" * 60)
ua, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
# User A creates data
r = ua.post("/api/calendar/groups", json={"name": "A's Group"})
group_a = r.get_json()["id"]
ua.post("/api/alarms", json={"label": "A alarm", "time": "09:00"})
from datetime import datetime, timedelta
_future = (datetime.now() + timedelta(days=1)).strftime("%Y%m%dT%H%M%S")
r = ua.post("/api/calendar/events", json={"group_id": group_a, "summary": "A event", "dtstart": _future})
check("User A creates data", r.status_code == 201)
# User B (the default-remark user) sees none of it
ub, _ = do_login(app5.test_client(), pyotp.TOTP(secret_b).now())
r = ub.get("/api/calendar/groups").get_json()
check("User B sees no groups", r == [])
r = ub.get("/api/alarms").get_json()
check("User B sees no alarms", r == [])
r = ub.get("/api/display/state").get_json()
check("User B state has no A events", r["events"] == [] and r["alarms"] == [])
# User B cannot tamper with A's group
r = ub.delete(f"/api/calendar/groups/{group_a}")
check("User B cannot delete A's group (404)", r.status_code == 404)
r = ua.get("/api/calendar/groups").get_json()
check("A's group untouched", any(g["id"] == group_a for g in r))
# User A cannot fetch B's events (B has none) — and cannot touch admin stuff
r = ua.get(f"/api/calendar/groups/{group_a}/events").get_json()
check("A sees own group events", len(r) == 1)

print()
print("=" * 60)
print("TEST 8: Edit remark + delete user + session invalidation")
print("=" * 60)
# Fresh root session → no re-auth marker → admin ops must demand reauth.
fresh_root, _ = do_login(app5.test_client(), pyotp.TOTP(root_secret).now())
r = fresh_root.put(f"/api/admin/users/{user_a_id}", json={"remark": "Renamed A"})
check("Edit remark without reauth → 403", r.status_code == 403 and r.get_json().get("reauth_required"))
rr = reauth_totp(root, root_secret)
r = root.put(f"/api/admin/users/{user_a_id}", json={"remark": "Renamed A"})
check("Edit remark with reauth", r.status_code == 200 and r.get_json()["user"]["remark"] == "Renamed A")
# delete B (with reauth)
rr = reauth_totp(root, root_secret)
r = root.delete(f"/api/admin/users/{user_b_id}")
check("Delete user with reauth", r.status_code == 200)
check("User gone", get_user(user_b_id, db5) is None)
# B's old tokens invalid
r = ub.get("/api/alarms")
check("Deleted user's token invalidated", r.status_code == 401)
r = ub.post("/api/auth/refresh")
check("Deleted user's refresh invalidated", r.status_code == 401)
# cannot delete root / self
r = root.delete("/api/admin/users/0")
check("Cannot delete Root", r.status_code == 400)
r = root.delete(f"/api/admin/users/{0}")
check("Self-delete guard (root)", r.status_code == 400)

print()
print("=" * 60)
print("TEST 9: Kick devices (self account, all logged-in users) + Root kick-all-users")
print("=" * 60)
# Two devices of the same account + one root device
ua2, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
ua2_2, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
r = ua2.post("/api/display/kick_all")
check("User kicks own account devices → 200", r.status_code == 200)
check("Device 1 (controller itself) kicked", ua2.get("/api/alarms").status_code == 401)
check("Device 2 kicked", ua2_2.get("/api/alarms").status_code == 401)
check("Kicked device refresh also dead", ua2_2.post("/api/auth/refresh").status_code == 401)
check("Root unaffected by user's self-kick", root.get("/api/alarms").status_code == 200)
# Root kicks every user (including itself) — other logged-in users die too
ua2_3, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
rr = reauth_totp(root, root_secret)
r = root.post("/api/admin/kick_all_users")
check("Root kick-all-users works", r.status_code == 200)
check("Root's own device kicked too", root.get("/api/alarms").status_code == 401)
check("Other user kicked after root kick-all", ua2_3.get("/api/alarms").status_code == 401)
# re-login works after kick
r = app5.test_client().post("/api/auth/login", json={"code": pyotp.TOTP(root_secret).now(), "view": "control"})
check("Re-login after kick-all", r.status_code == 200)

print()
print("=" * 60)
print("TEST 10: Access/refresh token lifecycle (rotation, replay, refresh)")
print("=" * 60)
app11, db11 = fresh_app("t10")
r11 = root_client(app11, db11)
root11_secret = get_auth_config(db11, 0)["totp_secret"]
# fresh login → access + refresh cookie
raw = app11.test_client()
r = raw.post("/api/auth/login", json={"code": pyotp.TOTP(root11_secret).now(), "view": "control"})
check("Login sets refresh cookie", "refresh_token" in r.headers.get("Set-Cookie", ""))
access1 = r.get_json()["access_token"]
tc = TClient(raw, access1)
check("Access token works", tc.get("/api/theme").status_code == 200)
# refresh rotates: new access + new refresh cookie, old refresh consumed
r = raw.post("/api/auth/refresh")
check("Refresh succeeds", r.status_code == 200 and r.get_json().get("access_token"))
access2 = r.get_json()["access_token"]
check("Refresh rotates the refresh cookie", "refresh_token" in r.headers.get("Set-Cookie", ""))
tc.set_access(access2)
check("New access works", tc.get("/api/theme").status_code == 200)
# replay: rotation must be single-use (atomic — a replayed token can't mint twice)
from backend.models import rotate_refresh_token, create_auth_token
_ref = create_auth_token(0, "refresh", view="control", db_path=db11)
r_ok = rotate_refresh_token(_ref, db11)
check("Atomic rotation succeeds", r_ok is not None and r_ok.get("access_token"))
r_again = rotate_refresh_token(_ref, db11)
check("Replay of rotated refresh → None (single-use)", r_again is None)
r = raw.post("/api/auth/refresh")
check("Live refresh still works after rotation", r.status_code == 200)
# logout revokes refresh but NOT other devices' tokens
tc2, _ = do_login(app11.test_client(), pyotp.TOTP(root11_secret).now())
r = tc2.post("/api/auth/logout")
check("Logout ok", r.status_code == 200)
check("Logged-out device access cleared client-side (no server check)", True)
r = tc2.post("/api/auth/refresh")
check("Logged-out device refresh revoked", r.status_code == 401)
r = tc.get("/api/theme")
check("Other device unaffected by logout", r.status_code == 200)

print()
print("=" * 60)
print("TEST 11: SSE ticket auth (one-time, scoped)")
print("=" * 60)
# no ticket → 401
check("SSE without ticket → 401", app5.test_client().get("/api/events").status_code == 401)
ua3, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
r = ua3.post("/api/events/ticket")
check("Ticket issued to authenticated user", r.status_code == 201 and r.get_json().get("ticket"))
ticket = r.get_json()["ticket"]
r = app5.test_client().get("/api/events?ticket=" + ticket)
check("SSE connects with ticket", r.status_code == 200 and r.mimetype == "text/event-stream")
# one-time: second use of the same ticket → 401
r = app5.test_client().get("/api/events?ticket=" + ticket)
check("Ticket is one-time (reuse → 401)", r.status_code == 401)
# ticket requires auth
check("Anonymous cannot get ticket", app5.test_client().post("/api/events/ticket").status_code == 401)
# kick invalidates outstanding tickets immediately
ua3k, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
r = ua3k.post("/api/events/ticket")
ticket_k = r.get_json()["ticket"]
ua3k.post("/api/display/kick_all")
r = app5.test_client().get("/api/events?ticket=" + ticket_k)
check("Kick invalidates pending SSE ticket → 401", r.status_code == 401)
# EventBus scoping still holds
bus = EventBus()
q1 = bus.subscribe(user_id=1)
q2 = bus.subscribe(user_id=2)
qall = bus.subscribe(None)
bus.publish("alarm_triggered", {"alarms": []}, user_id=1)
def q_get(q, timeout=0.1):
    try:
        return q.get(timeout=timeout)
    except Exception:
        return None
check("User 1 receives own event", q_get(q1) is not None)
check("User 2 does NOT receive user 1 event", q_get(q2, 0.05) is None)
check("Global subscriber receives all events", q_get(qall) is not None)
bus.publish("global_event", {}, user_id=None)
check("Global event reaches all subscribers", q_get(q2) is not None)

print()
print("=" * 60)
print("TEST 12: CSRF origin check on admin operations")
print("=" * 60)
root, _ = do_login(app5.test_client(), pyotp.TOTP(root_secret).now())  # re-login after kicks
rr = reauth_totp(root, root_secret)
r = root.post("/api/admin/users", json={"remark": "CSRF", "mode": "admin_fills"},
              headers={"Origin": "http://evil.example"})
check("Cross-origin admin POST blocked", r.status_code == 403)
r = root.get("/api/admin/users", headers={"Origin": "http://evil.example"})
check("Cross-origin GET allowed (safe)", r.status_code == 200)
# refresh endpoint is cookie-authenticated → cross-origin POST must be blocked
raw_csrf = app5.test_client()
raw_csrf.post("/api/auth/login", json={"code": pyotp.TOTP(root_secret).now(), "view": "control"})
r = raw_csrf.post("/api/auth/refresh", headers={"Origin": "http://evil.example"})
check("Cross-origin refresh blocked", r.status_code == 403)
r = raw_csrf.post("/api/auth/refresh")
check("Same-origin refresh allowed", r.status_code == 200)

print()
print("=" * 60)
print("TEST 13: Passkey flows (mocked ceremonies)")
print("=" * 60)
from backend import webauthn_mgr
import backend.routes as routes_mod

CRED_HEX = "aabbccddeeff00112233445566778899"
CRED_B64 = b64url(CRED_HEX)

orig_begin_reg = routes_mod.begin_registration
orig_complete_reg = routes_mod.complete_registration
orig_begin_auth = routes_mod.begin_authentication
orig_complete_auth = routes_mod.complete_authentication

routes_mod.begin_registration = lambda request, user_id, display_name="", existing_credential_id=None: {
    "challenge": "AA", "user": {"id": b64url(f"user_{user_id}".encode().hex()), "name": f"user_{user_id}", "displayName": display_name},
}
routes_mod.complete_registration = lambda request, credential: {
    "ok": True, "user_id": int(credential.get("user_name", "user_0").split("_")[-1]),
    "credential_id": CRED_HEX, "public_key": "abcdef",
}
routes_mod.begin_authentication = lambda request, allow_credential_ids=None: {"challenge": "BB"}
def _cred_hex(credential):
    import base64
    cid = credential.get("id", "")
    try:
        padded = cid + "=" * ((4 - len(cid) % 4) % 4)
        return base64.urlsafe_b64decode(padded.encode("ascii")).hex()
    except Exception:
        return "unknown"
routes_mod.complete_authentication = lambda request, credential, stored_credential_id=None, stored_public_key="": {
    "ok": True, "credential_id": _cred_hex(credential),
}

app15, db15 = fresh_app("t13")
root15 = root_client(app15, db15)
root15_secret = get_auth_config(db15, 0)["totp_secret"]
rr = reauth_totp(root15, root15_secret)
r = root15.post("/api/admin/users", json={"remark": "PK User", "mode": "admin_fills"})
pk_uid = r.get_json()["user"]["id"]
# admin provisions passkey (ceremony in admin browser)
r = root15.post(f"/api/admin/users/{pk_uid}/passkey/begin")
check("Admin passkey begin", r.status_code == 200 and "challenge" in r.get_json())
r = root15.post(f"/api/admin/users/{pk_uid}/passkey/complete", json={"credential": {"id": CRED_B64, "user_name": f"user_{pk_uid}"}})
check("Admin passkey complete stored", r.status_code == 200)
# passkey login for that user
raw15 = app15.test_client()
r = raw15.post("/api/auth/webauthn/login/begin")
check("Passkey login begin", r.status_code == 200)
r = raw15.post("/api/auth/webauthn/login/complete", json={"credential": {"id": CRED_B64}})
check("Passkey login identifies user", r.status_code == 200 and r.get_json().get("access_token"))
pk_tc = TClient(raw15, r.get_json()["access_token"])
st = pk_tc.get("/api/auth/status").get_json()
check("Logged in as the passkey user", st["user"]["remark"] == "PK User")
# passkey reauth (admin, TOTP-less fallback): give root a passkey first
uac(0, passkey_credential_id=CRED_HEX, passkey_public_key="abcdef", _db_path=db15)
raw15b = app15.test_client()
root15b, _ = do_login(raw15b, pyotp.TOTP(root15_secret).now())
r = root15b.post("/api/auth/reauth/webauthn/begin")
check("Passkey reauth begin", r.status_code == 200)
# complete with an unknown credential should fail (root's stored differs)
r = root15b.post("/api/auth/reauth/webauthn/complete", json={"credential": {"id": "zzzz"}})
check("Passkey reauth rejects wrong credential", r.status_code == 401)
r = root15b.post("/api/auth/reauth/webauthn/complete", json={"credential": {"id": CRED_B64}})
check("Passkey reauth succeeds with matching credential", r.status_code == 200)
# passkey first-config for a temp account → finalized via passkey
r = root15.post("/api/admin/users", json={"remark": "Temp PK", "mode": "user_fills"})
tuid = r.get_json()["user"]["id"]
tsec = get_temp_totp_secret(tuid, db15)
tc15, _ = do_login(app15.test_client(), pyotp.TOTP(tsec).now())
r = tc15.post("/api/auth/webauthn/register/begin")
check("Temp user passkey register begin", r.status_code == 200)
r = tc15.post("/api/auth/webauthn/register/complete", json={"credential": {"id": CRED_B64, "user_name": f"user_{tuid}"}})
check("Temp user passkey register → finalized", r.status_code == 200 and r.get_json()["finalized"])
check("Finalized passkey carries new access token", bool(r.get_json().get("access_token")))
check("Temp account finalized via passkey", is_temp_account(tuid, db15) is False)
r = root15.get(f"/api/admin/users/{tuid}/totp")
check("Temp TOTP gone after passkey finalize", r.status_code == 400)

# restore
routes_mod.begin_registration = orig_begin_reg
routes_mod.complete_registration = orig_complete_reg
routes_mod.begin_authentication = orig_begin_auth
routes_mod.complete_authentication = orig_complete_auth

print()
print("=" * 60)
print("TEST 14: Scheduler thread context (per-user timezone, no request context)")
print("=" * 60)
from backend.time_util import set_thread_user, clear_thread_user, get_tz_str, _ctx_user_id

app16, db16 = fresh_app("t14")
xid = create_user("Sched User", db_path=db16)
conn = get_db(db16)
conn.execute("INSERT INTO clock_config (user_id, city, timezone, is_primary) VALUES (?, '', 'America/New_York', 1)", (xid,))
conn.commit()
conn.close()
set_thread_user(xid)
with app16.app_context():
    check("Thread user resolves", _ctx_user_id() == xid)
    check("Per-user timezone in scheduler context", get_tz_str() == "America/New_York")
clear_thread_user()
with app16.app_context():
    check("Timezone falls back to Root default", get_tz_str() == "Asia/Shanghai")
# manager calls without request context are user-scoped via thread context
set_thread_user(xid)
with app16.app_context():
    from backend import alarm_mgr, countdown_mgr
    r = alarm_mgr.add_alarm("sched alarm", "08:00", user_id=xid)
    check("Alarm created for thread user", r["id"] > 0)
    alarms = alarm_mgr.get_alarms()
    check("Manager resolves thread user without request context", len(alarms) == 1)
clear_thread_user()
with app16.app_context():
    alarms0 = alarm_mgr.get_alarms(user_id=0)
    check("Root sees no other user's alarms", alarms0 == [])
# scheduler boots and runs per-user jobs without crashing
from backend.scheduler import start_scheduler, stop_scheduler
start_scheduler(app16)
time.sleep(1.5)
stop_scheduler()
check("Scheduler starts and stops cleanly", True)

print()
print("=" * 60)
print("TEST 15: Security hardening (SSRF, reauth gate, cascade delete, cross-user writes)")
print("=" * 60)
from backend.url_util import is_safe_public_url, is_safe_https_host
check("SSRF: loopback URL rejected", is_safe_public_url("http://127.0.0.1/x.ics") is False)
check("SSRF: cloud metadata rejected", is_safe_public_url("http://169.254.169.254/latest/meta-data") is False)
check("SSRF: private IP rejected", is_safe_public_url("http://10.0.0.5/cal.ics") is False)
check("SSRF: public URL accepted", is_safe_public_url("https://example.com/cal.ics") is True)
check("Host: private host rejected", is_safe_https_host("127.0.0.1") is False)
check("Host: public host accepted", is_safe_https_host("example.com") is True)

# Re-establish sessions (kicked earlier) — Root included
root, _ = do_login(app5.test_client(), pyotp.TOTP(root_secret).now())
rr = reauth_totp(root, root_secret)
ua, _ = do_login(app5.test_client(), pyotp.TOTP(secret_a).now())
# User B was deleted in TEST 8 — create a fresh user C for cross-user checks.
rr = reauth_totp(root, root_secret)
r = root.post("/api/admin/users", json={"remark": "User C", "mode": "admin_fills"})
uc_id = r.get_json()["user"]["id"]
r = root.post(f"/api/admin/users/{uc_id}/totp/begin")
secret_c = r.get_json()["secret"]
root.post(f"/api/admin/users/{uc_id}/totp/verify", json={"code": pyotp.TOTP(secret_c).now()})
uc, _ = do_login(app5.test_client(), pyotp.TOTP(secret_c).now())

# reauth requires an authenticated session (no public TOTP oracle)
anon = app5.test_client()
r = anon.post("/api/auth/reauth", json={"code": pyotp.TOTP(secret_a).now()})
check("Reauth blocked for unauthenticated caller", r.status_code == 401)

# subscription SSRF rejection + cross-user calendar write rejection
rr = reauth_totp(root, root_secret)
r = ua.post("/api/calendar/sources/subscribe", json={"group_id": group_a, "name": "evil", "url": "http://127.0.0.1/x.ics"})
check("Subscription with unsafe URL rejected", r.status_code == 400)
r = ua.post("/api/calendar/sources/subscribe", json={"group_id": 999999, "name": "x", "url": "https://example.com/x.ics"})
check("Subscription to foreign group rejected", r.status_code == 400)
# user C cannot write into A's group
r = uc.post("/api/calendar/events", json={"group_id": group_a, "summary": "B intrusion", "dtstart": "20300101T090000"})
check("Cross-user event write rejected", r.status_code == 400)
r = uc.post("/api/calendar/sources/upload", data={"group_id": group_a, "name": "x", "file": (io.BytesIO(b"BEGIN:VCALENDAR"), "x.ics")}, content_type="multipart/form-data")
check("Cross-user ICS upload rejected", r.status_code == 400)
# qweather host SSRF
r = ua.put("/api/weather/config", json={"qweather_host": "127.0.0.1"})
check("Unsafe qweather host rejected", r.status_code == 400)

# cascade delete with full data (FK-safe order)
r = root.post("/api/admin/users", json={"remark": "Full Data", "mode": "admin_fills"})
fd_id = r.get_json()["user"]["id"]
r = root.post(f"/api/admin/users/{fd_id}/totp/begin")
fd_secret = r.get_json()["secret"]
root.post(f"/api/admin/users/{fd_id}/totp/verify", json={"code": pyotp.TOTP(fd_secret).now()})
fc, _ = do_login(app5.test_client(), pyotp.TOTP(fd_secret).now())
g = fc.post("/api/calendar/groups", json={"name": "g"}).get_json()
fc.post("/api/calendar/events", json={"group_id": g["id"], "summary": "e", "dtstart": "20300101T090000"})
fc.post("/api/alarms", json={"label": "a", "time": "10:00"})
fc.post("/api/countdowns", json={"label": "c", "duration_seconds": 999})
conn = get_db(db5)
before = {t: conn.execute(f"SELECT COUNT(*) FROM {t} WHERE user_id=?", (fd_id,)).fetchone()[0]
          for t in ["calendar_groups", "calendar_sources", "calendar_events", "alarms", "countdowns", "auth_config", "pomodoro_state", "auth_tokens"]}
conn.close()
check("Cascade test user has data", sum(before.values()) >= 5)
rr = reauth_totp(root, root_secret)
r = root.delete(f"/api/admin/users/{fd_id}")
check("Delete full-data user succeeds", r.status_code == 200)
conn = get_db(db5)
after = {t: conn.execute(f"SELECT COUNT(*) FROM {t} WHERE user_id=?", (fd_id,)).fetchone()[0] for t in before}
conn.close()
check("All user data cascade-deleted (FK-safe)", sum(after.values()) == 0, str(after))
check("Deleted user's live tokens invalidated", fc.get("/api/alarms").status_code == 401)

print()
print("=" * 60)
print("TEST 16: Passkey stash 一次性参数暂存（小窗传参协商）")
print("=" * 60)
app18, db18 = fresh_app("t16")
c18 = app18.test_client()
opts = {"challenge": "ch123", "rp": {"id": "d.example.com"}, "user": {"id": "u1", "name": "user_1"}}
r = c18.post("/api/passkey/stash", json={"op": "creation", "options": opts})
check("stash 创建返回 sid", r.status_code == 201 and r.get_json().get("sid"))
sid = r.get_json()["sid"]
r = c18.get(f"/api/passkey/stash/{sid}")
check("stash 一次性消费返回参数", r.status_code == 200 and r.get_json()["op"] == "creation" and r.get_json()["options"]["rp"]["id"] == "d.example.com")
r = c18.get(f"/api/passkey/stash/{sid}")
check("stash 参数可重复读取（entry 保留至结果取走）", r.status_code == 200)
r = c18.get("/api/passkey/stash/notexist")
check("stash 未知 sid 返回 410", r.status_code == 410)
r = c18.post("/api/passkey/stash", json={"op": "bad", "options": {}})
check("stash 非法 op 拒绝", r.status_code == 400)
# 与中间件协议一致性：登录/注册/二次验证用的 options 均含 challenge + rp.id
r = c18.post("/api/passkey/stash", json={"op": "assertion", "options": {"challenge": "a", "rp": {"id": "d.example.com"}}})
sid2 = r.get_json()["sid"]
r = c18.get(f"/api/passkey/stash/{sid2}")
check("assertion 参数一致性", r.get_json()["options"]["challenge"] == "a")

# result 接口：小窗完成 → POST 结果 → App GET 取回（一次性）
r = c18.post("/api/passkey/stash", json={"op": "creation", "options": {"challenge": "b", "rp": {"id": "d.example.com"}}})
sid3 = r.get_json()["sid"]
cred = {"id": "cred-abc", "type": "public-key", "response": {"clientDataJSON": "x", "attestationObject": "y" * 2000, "transports": []}}
r = c18.post(f"/api/passkey/stash/{sid3}/result", json={"credential": cred, "user_name": "user_1"})
check("result 保存成功", r.status_code == 200 and r.get_json()["ok"])
r = c18.get(f"/api/passkey/stash/{sid3}/result")
check("App 取回完整 credential（含大数据）", r.status_code == 200 and r.get_json()["credential"]["id"] == "cred-abc" and r.get_json()["user_name"] == "user_1")
r = c18.get(f"/api/passkey/stash/{sid3}/result")
check("result 二次取回返回 410", r.status_code == 410)
# 未 POST 结果前 GET → 410
r = c18.post("/api/passkey/stash", json={"op": "assertion", "options": {"challenge": "c", "rp": {"id": "d.example.com"}}})
sid4 = r.get_json()["sid"]
r = c18.get(f"/api/passkey/stash/{sid4}/result")
check("未保存结果时 GET 返回 410", r.status_code == 410)
# 真实时序回归：小窗先 GET 参数（只读，不消费）→ 完成认证 → POST result 必须成功
r = c18.post("/api/passkey/stash", json={"op": "creation", "options": {"challenge": "d", "rp": {"id": "d.example.com"}}})
sid5 = r.get_json()["sid"]
r = c18.get(f"/api/passkey/stash/{sid5}")
check("小窗读取参数成功", r.status_code == 200)
r = c18.post(f"/api/passkey/stash/{sid5}/result", json={"credential": {"id": "reg-final", "type": "public-key", "response": {}}})
check("读取参数后仍可 POST result（修复 410 回归）", r.status_code == 200)

print()
print("=" * 60)
print("TEST 17: 账户独立主题（dark/light/auto）")
print("=" * 60)
app19, db19 = fresh_app("t17")
pw19 = get_temp_password(db19, 0)
c19, _ = do_login(app19.test_client(), pw19)
setup_totp(c19)

r = c19.get("/api/theme")
check("默认主题为 dark", r.status_code == 200 and r.get_json()["mode"] == "dark" and r.get_json()["effective"] == "dark")
r = c19.put("/api/theme", json={"mode": "light"})
check("保存 light → effective light", r.status_code == 200 and r.get_json()["effective"] == "light")
r = c19.put("/api/theme", json={"mode": "auto"})
check("auto 解析为 dark/light 之一", r.status_code == 200 and r.get_json()["effective"] in ("dark", "light"))
r = c19.put("/api/theme", json={"mode": "bad"})
check("非法模式拒绝", r.status_code == 400)

# 账户独立：创建用户 A/B，各自设置互不影响
from backend.models import create_user as cu19, update_user as uu19
ua19 = cu19("Theme A", db_path=db19)
ub19 = cu19("Theme B", db_path=db19)
uu19(ua19, theme="light", db_path=db19)
uu19(ub19, theme="dark", db_path=db19)
check("账户 A 主题=light", get_user(ua19, db19)["theme"] == "light")
check("账户 B 主题=dark", get_user(ub19, db19)["theme"] == "dark")

# 未登录访问 /api/theme → 401
anon19 = app19.test_client()
check("未登录 GET /api/theme → 401", anon19.get("/api/theme").status_code == 401)

print()
print("=" * 60)
print("TEST 18: 账户独立时区（display/state 返回 timezone + 时区设置生效）")
print("=" * 60)
app20, db20 = fresh_app("t18")
pw20 = get_temp_password(db20, 0)
c20, _ = do_login(app20.test_client(), pw20)
setup_totp(c20)
# 默认 Asia/Shanghai
r = c20.get("/api/time/config")
check("默认时区 Asia/Shanghai", r.status_code == 200 and r.get_json()["timezone"] == "Asia/Shanghai")
# 设置时区
r = c20.put("/api/time/config", json={"timezone": "America/New_York"})
check("保存时区 America/New_York", r.status_code == 200)
r = c20.get("/api/time/config")
check("时区读取生效", r.get_json()["timezone"] == "America/New_York")
# display/state 返回该用户时区
st = c20.get("/api/display/state").get_json()
check("display/state 返回账户时区", st.get("timezone") == "America/New_York")
# 无效时区拒绝
r = c20.put("/api/time/config", json={"timezone": "Mars/Olympus"})
check("无效时区拒绝", r.status_code == 400)
# 账户独立：另一用户仍是默认时区
ua20 = create_user("TZ A", db_path=db20)
conn = get_db(db20)
conn.execute("UPDATE users SET status='active' WHERE id=?", (ua20,))
conn.close()
# 给用户 A 配 TOTP 再登录
uac(ua20, totp_secret="JBSWY3DPEHPK3PXP", totp_enabled=1, _db_path=db20)
a20, _ = do_login(app20.test_client(), pyotp.TOTP("JBSWY3DPEHPK3PXP").now())
st = a20.get("/api/display/state").get_json()
check("用户 B 时区保持默认", st.get("timezone") == "Asia/Shanghai")

print()
print("=" * 60)
print("TEST 19: 页面导航鉴权（refresh cookie，浏览器导航不带 Authorization）")
print("=" * 60)
# 模拟真实浏览器：登录后导航页面（无 Authorization header，仅 cookie）
raw_nav = app5.test_client()
r = raw_nav.post("/api/auth/login", json={"code": pyotp.TOTP(secret_a).now(), "view": "control"})
check("页面导航: 登录成功", r.status_code == 200)
r = raw_nav.get("/control")
check("控制端页面导航 → 200（cookie 鉴权）", r.status_code == 200)
r = raw_nav.get("/")
check("显示端页面导航 → 200", r.status_code == 200)
r = raw_nav.get("/mfa-setup")
check("MFA 设置页导航 → 200", r.status_code == 200)
# 未登录导航 → 302 登录页
anon_nav = app5.test_client()
check("未登录 /control → 302", anon_nav.get("/control").status_code == 302)
check("未登录 / → 302", anon_nav.get("/").status_code == 302)
check("未登录 /mfa-setup → 302", anon_nav.get("/mfa-setup").status_code == 302)
# temp 用户导航（用 TEST 6 的 temp 账户，仍是临时账户 → 强制 MFA 设置页）
rr = reauth_totp(root, root_secret)
r = root.post("/api/admin/users", json={"remark": "Nav Temp", "mode": "user_fills"})
nav_temp_id = r.get_json()["user"]["id"]
nav_temp_secret = get_temp_totp_secret(nav_temp_id, db5)
bob_raw = app5.test_client()
r = bob_raw.post("/api/auth/login", json={"code": pyotp.TOTP(nav_temp_secret).now(), "view": "control"})
check("temp 用户登录", r.status_code == 200)
check("temp 用户 /mfa-setup → 200", bob_raw.get("/mfa-setup").status_code == 200)
loc = bob_raw.get("/control").headers.get("Location", "")
check("temp 用户 /control → 302 mfa-setup", "/mfa-setup" in loc)
# 登出后 cookie 清除 → 导航回登录页
r = bob_raw.post("/api/auth/logout")
check("登出 ok", r.status_code == 200)
check("登出后 /mfa-setup → 302", bob_raw.get("/mfa-setup").status_code == 302)

print()
print("=" * 60)
print("TEST 20: 数据库管理（仅 Root、导出/导入 roundtrip、格式识别、原子性）")
print("=" * 60)
app21, db21 = fresh_app("t20")
root21 = root_client(app21, db21)
root21_secret = get_auth_config(db21, 0)["totp_secret"]

# 造点数据
c21 = app21.test_client()
r = c21.post("/api/auth/login", json={"code": pyotp.TOTP(root21_secret).now(), "view": "control"})
root21 = TClient(c21, r.get_json()["access_token"])
r = root21.post("/api/calendar/groups", json={"name": "DB Test"})
check("Setup: group created", r.status_code == 201)
r = root21.post("/api/alarms", json={"label": "DB alarm", "time": "07:30"})
check("Setup: alarm created", r.status_code == 201)

# 非 root 用户（普通 admin 角色也拒绝——仅 root id 0）
rr = reauth_totp(root21, root21_secret)
r = root21.post("/api/admin/users", json={"remark": "DbUser", "mode": "admin_fills"})
dbu_id = r.get_json()["user"]["id"]
r = root21.post(f"/api/admin/users/{dbu_id}/totp/begin")
dbu_secret = r.get_json()["secret"]
root21.post(f"/api/admin/users/{dbu_id}/totp/verify", json={"code": pyotp.TOTP(dbu_secret).now()})
dbu, _ = do_login(app21.test_client(), pyotp.TOTP(dbu_secret).now())
r = dbu.get("/api/admin/db/status")
check("非 root 访问 db 管理 → 403", r.status_code == 403)
r = dbu.get("/api/admin/db/export")
check("非 root 导出 → 403", r.status_code == 403)

# 未二次验证 → 403（用全新的 root 会话，无 reauth 标记）
fresh_root21, _ = do_login(app21.test_client(), pyotp.TOTP(root21_secret).now())
r = fresh_root21.get("/api/admin/db/export")
check("导出未 reauth → 403", r.status_code == 403 and r.get_json().get("reauth_required"))

# 状态
r = root21.get("/api/admin/db/status")
d = r.get_json()
check("db status 返回 PostgreSQL v" + str(DB_VERSION), d.get("database") == "postgresql" and d.get("db_version") == DB_VERSION)

# 导出（reauth 后）
rr = reauth_totp(root21, root21_secret)
r = root21.get("/api/admin/db/export")
check("Root 导出成功", r.status_code == 200 and "json" in r.content_type)
backup = json.loads(r.data)
check("备份格式正确", backup.get("format") == "iot-dboard-backup")
check("备份含用户数据", len(backup["tables"]["calendar_groups"]["rows"]) >= 1
      and len(backup["tables"]["alarms"]["rows"]) >= 1)
check("备份不含会话令牌", "auth_tokens" not in backup["tables"])

# 格式识别
check("SQLite magic 识别", db_admin.detect_format(b"SQLite format 3\x00" + b"x" * 8) == "sqlite")
check("JSON 备份识别", db_admin.detect_format(json.dumps({"format": "iot-dboard-backup"}).encode()) == "json")
check("未知格式识别", db_admin.detect_format(b"garbage") == "unknown")

# 导入备份 roundtrip（导入会清空会话 → root 需重登）
r = root21.post("/api/admin/db/import", data={"file": (io.BytesIO(json.dumps(backup).encode()), "b.json")},
                content_type="multipart/form-data")
check("导入备份成功", r.status_code == 200 and r.get_json().get("ok"), str(r.get_json()))
check("导入后会话失效标记", r.get_json().get("session_invalidated") is True)
conn = get_db(db21)
n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE name='DB Test'").fetchone()[0]
conn.close()
check("导入后数据恢复", n >= 1)
# 导入后旧 token 失效
check("导入后旧会话失效 → 401", root21.get("/api/admin/db/status").status_code == 401)

# 导入非法文件 → 400（不破坏数据）
root21b, _ = do_login(app21.test_client(), pyotp.TOTP(root21_secret).now())
rr = reauth_totp(root21b, root21_secret)
r = root21b.post("/api/admin/db/import", data={"file": (io.BytesIO(b"junk"), "x.bin")},
                 content_type="multipart/form-data")
check("导入非法文件 → 400", r.status_code == 400, str(r.get_json()))
conn = get_db(db21)
n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE name='DB Test'").fetchone()[0]
conn.close()
check("非法导入未破坏数据", n >= 1)

# 导入失败原子性：构造一个会导致失败的备份（缺少必需列 → 行插入失败回滚）
root21c, _ = do_login(app21.test_client(), pyotp.TOTP(root21_secret).now())
rr = reauth_totp(root21c, root21_secret)
bad = {
    "format": "iot-dboard-backup",
    "db_version": DB_VERSION,
    "tables": {"calendar_groups": {"columns": ["id", "user_id", "name", "color", "enabled", "created_at"],
                                   "rows": [[1, 0, "OK Group", "#fff", 1, "2026-01-01 00:00:00"],
                                            [2, 0, None, "#fff", 1, None]]}},  # name NOT NULL → None 会失败
}
# 直接把 name=None 塞进去会因参数为 None 失败（psycopg2 拒绝 None 绑定到 NOT NULL？其实 None → NULL，NOT NULL 约束报错）
r = root21c.post("/api/admin/db/import", data={"file": (io.BytesIO(json.dumps(bad).encode()), "bad.json")},
                 content_type="multipart/form-data")
check("含非法行的导入 → 400/500", r.status_code in (400, 500), str(r.get_json()))
conn = get_db(db21)
n = conn.execute("SELECT COUNT(*) FROM calendar_groups WHERE name='OK Group'").fetchone()[0]
conn.close()
check("失败导入整体回滚（原子性）", n == 0)

print()
print("=" * 60)
print(f"RESULT: {PASS} passed, {FAIL} failed")
print("=" * 60)
sys.exit(1 if FAIL else 0)