Spaces:

longxing
/

gmn2a

Running

gmn2a / tests /test_admin_panel.py

Fix streaming replay and isolate temporary chats

fae37b1 about 1 month ago

5.78 kB

"""校验管理面板前端与后端状态接口是否对得上。
检查三件事：
1. admin.html 里 getElementById 引用的 id 都在 HTML 里存在；
2. admin.html 里读的 data.X 字段都在 /api/status 的返回里；
3. admin.html 里读的 session.X 字段都在 session_manager.status() 里。
"""

import asyncio
import os
import re
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

os.environ.setdefault("GEMINI_COOKIE_PATH", tempfile.mkdtemp(prefix="gs-admin-"))
os.environ.setdefault("SECURE_1PSID", "PSID-ADMIN")
os.environ.setdefault("SECURE_1PSIDTS", "TS-ADMIN")

PASS, FAIL = [], []


def check(name, cond, extra=""):
	(PASS if cond else FAIL).append(name)
	print(("[OK]   " if cond else "[FAIL] ") + name + (f"  {extra}" if extra and not cond else ""))


def main():
	html = (ROOT / "admin.html").read_text(encoding="utf-8")

	# 1. id 引用完整性
	declared = set(re.findall(r'\bid="([^"]+)"', html))
	referenced = set(re.findall(r"getElementById\(['\"]([^'\"]+)['\"]\)", html))
	missing_ids = sorted(referenced - declared)
	check("1a getElementById 引用的 id 全部存在", not missing_ids, f"缺失: {missing_ids}")
	print(f"       HTML 声明 {len(declared)} 个 id，JS 引用 {len(referenced)} 个")

	# 新增字段必须真的被引用（否则等于没接上）
	for element_id in (
		"sessionStatus",
		"sessionSource",
		"rotationTask",
		"rotationCount",
		"rotationAge",
		"live1psidts",
		"cached1psidts",
		"sessionErrorRow",
		"sessionError",
		"sessionRecovery",
		"lastResponse",
	):
		check(f"1b 面板已接线 {element_id}", element_id in declared and element_id in referenced)

	# 2/3. 字段名对得上
	import admin
	import main

	status = asyncio.run(admin.get_status(token=""))
	api_keys = set(status.keys())
	session_keys = set(main.session_manager.status().keys())

	check("2a /api/status 含 session 块", "session" in api_keys)
	check("2b /api/status 含扁平化的 rotation_count", "rotation_count" in api_keys)

	# 只检查 fetchStatus 内部对 /api/status 返回值的引用，避免把其它接口的字段算进来
	status_start = html.index("async function fetchStatus()")
	status_end = html.index("// 获取模型列表", status_start)
	status_body = html[status_start:status_end]
	data_refs = set(re.findall(r"\bdata\.([a-z_][a-z0-9_]*)", status_body))
	bad_data = sorted(r for r in data_refs if r not in api_keys)
	check("2c fetchStatus 引用的 data.X 都存在", not bad_data, f"多余: {bad_data}")
	print(f"       fetchStatus 引用 {len(data_refs)} 个字段")

	session_refs = set(re.findall(r"\bsession\.([a-z_][a-z0-9_]*)", status_body))
	bad_session = sorted(r for r in session_refs if r not in session_keys)
	check("3a 面板引用的 session.X 都存在", not bad_session, f"多余: {bad_session}")
	print(f"       面板引用 {len(session_refs)} 个 session 字段")

	# 面板依赖的关键字段确实有值
	for key in ("rotation_count", "last_1psidts_change_age_sec", "rotation_task_alive", "supervisor_alive", "live_1psidts_masked"):
		check(f"3b session 状态含 {key}", key in session_keys)

	# 防回归：面板不应再直接用 env 里的静态 1PSIDTS 当作"生效值"
	check("4a 面板展示了生效中的 1PSIDTS", "live1psidts" in referenced)
	check("4b 面板展示了缓存里的 1PSIDTS", "cached1psidts" in referenced)

	# ---------------------------------------------------------------- 功能开关：改了就立即生效
	# 面板的开关直接改运行时变量，不需要重启；同时要把派生的 AUTO_DELETE_CHAT 一起重算。
	saved_write_env = admin.write_env
	admin.write_env = lambda updates: None  # 别在测试里真写 .env
	try:
		saved_flags = (
			main.ENABLE_THINKING,
			main.TEMPORARY_CHAT,
			main.AUTO_DELETE_CHAT_SETTING,
		)

		# 从一个确定的状态开始，避免依赖环境默认值（TEMPORARY_CHAT 默认是开的）
		main.TEMPORARY_CHAT = False
		main.AUTO_DELETE_CHAT_SETTING = True
		main.refresh_derived_flags()

		res = asyncio.run(admin.update_config(admin.ConfigUpdate(feature="thinking", enabled=True), token=""))
		check("5a 开启思考后运行时变量立即为真（无需重启）", main.ENABLE_THINKING is True, str(main.ENABLE_THINKING))
		check("5b 返回信息说明已立即生效", "立即生效" in res.get("message", ""), str(res.get("message")))

		asyncio.run(admin.update_config(admin.ConfigUpdate(feature="autoDelete", enabled=True), token=""))
		check("5c 自动删除正常生效", main.AUTO_DELETE_CHAT is True, str(main.AUTO_DELETE_CHAT))

		asyncio.run(admin.update_config(admin.ConfigUpdate(feature="temporary", enabled=True), token=""))
		check("5d 开临时对话后，自动删除被派生为关", main.AUTO_DELETE_CHAT is False, str(main.AUTO_DELETE_CHAT))

		res2 = asyncio.run(admin.update_config(admin.ConfigUpdate(feature="autoDelete", enabled=True), token=""))
		check("5e 临时对话开着时，自动删除仍然不生效", main.AUTO_DELETE_CHAT is False, str(main.AUTO_DELETE_CHAT))
		check("5f 且返回信息给出了原因", "临时对话" in res2.get("message", ""), str(res2.get("message")))

		asyncio.run(admin.update_config(admin.ConfigUpdate(feature="temporary", enabled=False), token=""))
		check("5g 关掉临时对话后，自动删除恢复", main.AUTO_DELETE_CHAT is True, str(main.AUTO_DELETE_CHAT))

		# 复原
		main.ENABLE_THINKING, main.TEMPORARY_CHAT, main.AUTO_DELETE_CHAT_SETTING = saved_flags
		main.refresh_derived_flags()
	finally:
		admin.write_env = saved_write_env

	print(f"\n通过 {len(PASS)} / {len(PASS) + len(FAIL)}")
	if FAIL:
		print("失败项：")
		for name in FAIL:
			print("  -", name)
	return 1 if FAIL else 0


if __name__ == "__main__":
	sys.exit(main())

	"""校验管理面板前端与后端状态接口是否对得上。

	检查三件事：
	1. admin.html 里 getElementById 引用的 id 都在 HTML 里存在；
	2. admin.html 里读的 data.X 字段都在 /api/status 的返回里；
	3. admin.html 里读的 session.X 字段都在 session_manager.status() 里。
	"""

	import asyncio
	import os
	import re
	import sys
	import tempfile
	from pathlib import Path

	ROOT = Path(__file__).resolve().parent.parent
	sys.path.insert(0, str(ROOT))

	os.environ.setdefault("GEMINI_COOKIE_PATH", tempfile.mkdtemp(prefix="gs-admin-"))
	os.environ.setdefault("SECURE_1PSID", "PSID-ADMIN")
	os.environ.setdefault("SECURE_1PSIDTS", "TS-ADMIN")

	PASS, FAIL = [], []


	def check(name, cond, extra=""):
	(PASS if cond else FAIL).append(name)
	print(("[OK] " if cond else "[FAIL] ") + name + (f" {extra}" if extra and not cond else ""))


	def main():
	html = (ROOT / "admin.html").read_text(encoding="utf-8")

	# 1. id 引用完整性
	declared = set(re.findall(r'\bid="([^"]+)"', html))
	referenced = set(re.findall(r"getElementById\(['\"]([^'\"]+)['\"]\)", html))
	missing_ids = sorted(referenced - declared)
	check("1a getElementById 引用的 id 全部存在", not missing_ids, f"缺失: {missing_ids}")
	print(f" HTML 声明 {len(declared)} 个 id，JS 引用 {len(referenced)} 个")

	# 新增字段必须真的被引用（否则等于没接上）
	for element_id in (
	"sessionStatus",
	"sessionSource",
	"rotationTask",
	"rotationCount",
	"rotationAge",
	"live1psidts",
	"cached1psidts",
	"sessionErrorRow",
	"sessionError",
	"sessionRecovery",
	"lastResponse",
	):
	check(f"1b 面板已接线 {element_id}", element_id in declared and element_id in referenced)

	# 2/3. 字段名对得上
	import admin
	import main

	status = asyncio.run(admin.get_status(token=""))
	api_keys = set(status.keys())
	session_keys = set(main.session_manager.status().keys())

	check("2a /api/status 含 session 块", "session" in api_keys)
	check("2b /api/status 含扁平化的 rotation_count", "rotation_count" in api_keys)

	# 只检查 fetchStatus 内部对 /api/status 返回值的引用，避免把其它接口的字段算进来
	status_start = html.index("async function fetchStatus()")
	status_end = html.index("// 获取模型列表", status_start)
	status_body = html[status_start:status_end]
	data_refs = set(re.findall(r"\bdata\.([a-z_][a-z0-9_]*)", status_body))
	bad_data = sorted(r for r in data_refs if r not in api_keys)
	check("2c fetchStatus 引用的 data.X 都存在", not bad_data, f"多余: {bad_data}")
	print(f" fetchStatus 引用 {len(data_refs)} 个字段")

	session_refs = set(re.findall(r"\bsession\.([a-z_][a-z0-9_]*)", status_body))
	bad_session = sorted(r for r in session_refs if r not in session_keys)
	check("3a 面板引用的 session.X 都存在", not bad_session, f"多余: {bad_session}")
	print(f" 面板引用 {len(session_refs)} 个 session 字段")

	# 面板依赖的关键字段确实有值
	for key in ("rotation_count", "last_1psidts_change_age_sec", "rotation_task_alive", "supervisor_alive", "live_1psidts_masked"):
	check(f"3b session 状态含 {key}", key in session_keys)

	# 防回归：面板不应再直接用 env 里的静态 1PSIDTS 当作"生效值"
	check("4a 面板展示了生效中的 1PSIDTS", "live1psidts" in referenced)
	check("4b 面板展示了缓存里的 1PSIDTS", "cached1psidts" in referenced)

	# ---------------------------------------------------------------- 功能开关：改了就立即生效
	# 面板的开关直接改运行时变量，不需要重启；同时要把派生的 AUTO_DELETE_CHAT 一起重算。
	saved_write_env = admin.write_env
	admin.write_env = lambda updates: None # 别在测试里真写 .env
	try:
	saved_flags = (
	main.ENABLE_THINKING,
	main.TEMPORARY_CHAT,
	main.AUTO_DELETE_CHAT_SETTING,
	)

	# 从一个确定的状态开始，避免依赖环境默认值（TEMPORARY_CHAT 默认是开的）
	main.TEMPORARY_CHAT = False
	main.AUTO_DELETE_CHAT_SETTING = True
	main.refresh_derived_flags()

	res = asyncio.run(admin.update_config(admin.ConfigUpdate(feature="thinking", enabled=True), token=""))
	check("5a 开启思考后运行时变量立即为真（无需重启）", main.ENABLE_THINKING is True, str(main.ENABLE_THINKING))
	check("5b 返回信息说明已立即生效", "立即生效" in res.get("message", ""), str(res.get("message")))

	asyncio.run(admin.update_config(admin.ConfigUpdate(feature="autoDelete", enabled=True), token=""))
	check("5c 自动删除正常生效", main.AUTO_DELETE_CHAT is True, str(main.AUTO_DELETE_CHAT))

	asyncio.run(admin.update_config(admin.ConfigUpdate(feature="temporary", enabled=True), token=""))
	check("5d 开临时对话后，自动删除被派生为关", main.AUTO_DELETE_CHAT is False, str(main.AUTO_DELETE_CHAT))

	res2 = asyncio.run(admin.update_config(admin.ConfigUpdate(feature="autoDelete", enabled=True), token=""))
	check("5e 临时对话开着时，自动删除仍然不生效", main.AUTO_DELETE_CHAT is False, str(main.AUTO_DELETE_CHAT))
	check("5f 且返回信息给出了原因", "临时对话" in res2.get("message", ""), str(res2.get("message")))

	asyncio.run(admin.update_config(admin.ConfigUpdate(feature="temporary", enabled=False), token=""))
	check("5g 关掉临时对话后，自动删除恢复", main.AUTO_DELETE_CHAT is True, str(main.AUTO_DELETE_CHAT))

	# 复原
	main.ENABLE_THINKING, main.TEMPORARY_CHAT, main.AUTO_DELETE_CHAT_SETTING = saved_flags
	main.refresh_derived_flags()
	finally:
	admin.write_env = saved_write_env

	print(f"\n通过 {len(PASS)} / {len(PASS) + len(FAIL)}")
	if FAIL:
	print("失败项：")
	for name in FAIL:
	print(" -", name)
	return 1 if FAIL else 0


	if __name__ == "__main__":
	sys.exit(main())