"""P4-07 real browser operations verification using Playwright (headless Chromium). Runs against https://192.168.199.22 with the admin account from env (P4_OPS_EMAIL, P4_OPS_PASSWORD) which must be supplied by the invoking shell. Verifies desktop and mobile viewports, admin operations, denial for a normal user, and that an entitlement change (VIP grant) affects a controlled user account and TTS limits. Never prints the admin password. """ from __future__ import annotations import json import os import sys from uuid import uuid4 import httpx from playwright.sync_api import sync_playwright BASE = "https://192.168.199.22" API = f"{BASE}/api/v1" def need(name: str) -> str: value = os.getenv(name) if not value: raise SystemExit(f"missing env {name}") return value def api_login(email: str, password: str) -> httpx.Client: import time session = httpx.Client(verify=False) for _ in range(30): csrf = session.get(f"{API}/auth/csrf").json()["csrf_token"] resp = session.post(f"{API}/auth/login", json={"email": email, "password": password}, headers={"X-CSRF-Token": csrf}) if resp.status_code == 200: session.headers.update({"X-CSRF-Token": csrf}) return session if resp.status_code == 429: time.sleep(15) continue raise SystemExit(f"login failed {resp.status_code}: {resp.text}") raise SystemExit("login rate-limited and retries exhausted") def main() -> None: admin_email = need("P4_OPS_EMAIL") admin_password = need("P4_OPS_PASSWORD") controlled_email = need("P4_CTRL_EMAIL") controlled_password = need("P4_CTRL_PASSWORD") results: dict = {} with sync_playwright() as p: browser = p.chromium.launch() desktop = browser.new_context(viewport={"width": 1440, "height": 900}, ignore_https_errors=True) mobile = browser.new_context(viewport={"width": 390, "height": 844}, ignore_https_errors=True) # Obtain the admin session cookie once via the API (which handles rate-limits), # then inject it into the browser contexts. This avoids consuming the login # rate-limit through repeated browser form logins. admin_http = api_login(admin_email, admin_password) admin_cookie = admin_http.cookies.get("kaotings_session") admin_http.close() cookie = {"name": "kaotings_session", "value": admin_cookie, "domain": "192.168.199.22", "path": "/", "secure": True} desktop.add_cookies([cookie]) mobile.add_cookies([cookie]) # --- Desktop: admin lands on account + admin dashboard --- page = desktop.new_page() page.goto(f"{BASE}/account") page.wait_for_load_state("networkidle") results["admin_login"] = ("管理后台" in page.content()) or ("基本资料" in page.content()) # --- Admin dashboard --- page.goto(f"{BASE}/admin") page.wait_for_load_state("networkidle") results["admin_dashboard"] = "管理概览" in page.content() # --- Users: search + detail + operations --- page.goto(f"{BASE}/admin/users") page.wait_for_load_state("networkidle") page.get_by_role("button", name="查询").click() page.wait_for_load_state("networkidle") results["users_table_renders"] = "用户与会员" in page.content() # open a user detail row (first 详情 button) detail_btn = page.get_by_role("button", name="详情").first if detail_btn.is_visible(): detail_btn.click() page.wait_for_load_state("networkidle") results["user_detail_renders"] = "额度周期" in page.content() or "操作原因" in page.content() # empty-state: filter by a non-existent email shows the empty message page.fill('input[aria-label="按邮箱筛选"]', "p4-no-such-user-zz@example.com") page.get_by_role("button", name="查询").click() try: page.wait_for_selector("text=无匹配用户", timeout=5000) results["users_empty_state"] = True except Exception: results["users_empty_state"] = False page.fill('input[aria-label="按邮箱筛选"]', "") page.get_by_role("button", name="查询").click() page.wait_for_load_state("networkidle") # error feedback: open detail, attempt an operation with empty reason -> inline error detail2 = page.get_by_role("button", name="详情").first if detail2.is_visible(): detail2.click() page.wait_for_load_state("networkidle") if page.get_by_role("button", name="调整额度").count() > 0: page.get_by_role("button", name="调整额度").click() page.wait_for_load_state("networkidle") results["operation_reason_required_feedback"] = "请填写原因" in page.content() or "操作原因" in page.content() # --- TTS + audit --- page.goto(f"{BASE}/admin/tts") page.wait_for_load_state("networkidle") results["tts_tasks_renders"] = "任务查询" in page.content() page.get_by_role("button", name="管理员审计").click() page.wait_for_load_state("networkidle") results["audit_renders"] = "审计" in page.content() or "操作者" in page.content() # --- Mobile: admin nav + users page responsive (admin cookie already injected) --- mpage = mobile.new_page() mpage.goto(f"{BASE}/admin/users") mpage.wait_for_load_state("networkidle") results["mobile_users_paginates"] = "第" in mpage.content() or "用户" in mpage.content() # --- Normal user denied admin (login via API, inject cookie in a dedicated context) --- ncontext = browser.new_context(viewport={"width": 1440, "height": 900}, ignore_https_errors=True) nuser_page = ncontext.new_page() # log the controlled user in via API to obtain its session cookie ctrl_http = api_login(controlled_email, controlled_password) ucookie = ctrl_http.cookies.get("kaotings_session") ctrl_http.close() token = {"name": "kaotings_session", "value": ucookie, "domain": "192.168.199.22", "path": "/", "secure": True} ncontext.add_cookies([token]) nuser_page.goto(f"{BASE}/admin") nuser_page.wait_for_load_state("networkidle") nuser_content = nuser_page.content() # Authoritative: the denial notice is rendered and no admin-only operational content is shown. # (Admin page titles are embedded in the RSC payload even when hidden, so use the denial notice # plus the absence of a real operations panel, e.g. the quota ledger card.) results["normal_user_denied"] = ("无权访问" in nuser_content) and ("累计使用" not in nuser_content) # --- Entitlement change affects controlled user account --- # Grant VIP via the admin API to the controlled user, then check /account reflects it admin_session = api_login(admin_email, admin_password) # find the controlled user id ulist = admin_session.get(f"{API}/admin/users", params={"email": controlled_email}).json() target = next(i for i in ulist["items"] if i["email"] == controlled_email) grant = admin_session.put(f"{API}/admin/users/{target['id']}/membership", json={"plan": "vip", "expires_at": None, "reason": "P4-07 controlled VIP grant"}, headers={"X-CSRF-Token": admin_session.headers["X-CSRF-Token"]}) assert grant.status_code == 200, grant.text admin_session.close() # refresh controlled user's account page (cookie already present in nuser_page) nuser_page.goto(f"{BASE}/account") nuser_page.wait_for_load_state("networkidle") acc_content = nuser_page.content() results["entitlement_vip_reflected"] = "VIP" in acc_content # entitlement also governs TTS submission: submit a small task as the controlled (VIP) user tts_client = api_login(controlled_email, controlled_password) tts_resp = tts_client.post(f"{API}/tts/tasks", json={"text": "P4 权益验证", "voice_id": "default", "parameters": {"format": "wav", "speed": 1}}, headers={"Idempotency-Key": f"p4-ent-{uuid4().hex}"}) results["entitlement_tts_202"] = tts_resp.status_code in {202, 200, 409} tts_client.close() ncontext.close() desktop.close() mobile.close() browser.close() # cleanup: disable the controlled user so it can't affect real operations cs = api_login(admin_email, admin_password) ulist = cs.get(f"{API}/admin/users", params={"email": controlled_email}).json() tid = next(i for i in ulist["items"] if i["email"] == controlled_email)["id"] cs.patch(f"{API}/admin/users/{tid}/status", json={"status": "disabled", "reason": "P4-07 cleanup"}, headers={"X-CSRF-Token": cs.headers["X-CSRF-Token"]}) cs.close() print(json.dumps(results, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()