186 lines
8.8 KiB
Python
186 lines
8.8 KiB
Python
"""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()
|