施工员andmultica-agent
3eaa36a8d5
主站 - 新增 m0006 invite_codes 迁移；注册强制邀请码（首个管理员除外），消码与建号 同一事务，并发提交只有一个能成功 - 新增 /api/hub-admin/* 服务端点（共享 HUB_ADMIN_TOKEN，先于鉴权校验），供数据 中枢桥接读写会话/密码/模型池/会员/邀请码，并提供供应商模型列表拉取 - 前端：注册表单加邀请码（桌面 login、index.html、移动端）；「系统管理」改为 「数据中枢」入口指向 8766，原模型池与会员管理分区移除，仅留「行情管理」； 随之清理陈旧 CSS 数据中枢 - 取消独立账号：删除 hub_admin/hub_sessions 与登录、改密、锁定逻辑，改为校验 主站 xiaobai_session，仅管理员可进，CSRF 由会话派生，危险操作二次确认走主站 - 控制台新增数据源凭证可编辑区（原有内容一项不删）、供应商制模型池（自动拉取 /models，失败退回卡内手动录入）、会员管理与邀请码页 - 日夜双主题：颜色收敛为同名 token 换值，SVG 改用 inline style 以吃到变量 自测 - 主站 verify_baseline 通过（498 项）；数据中枢 235 项通过 - tools/verify_datahub_console.py 端到端跑通两服务真实对话； tools/verify_datahub_console_ui.py 浏览器跑通门禁/凭证/模型池/会员/主题/1030 窄屏 Co-authored-by: multica-agent <github@multica.ai>
94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from http import HTTPStatus
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from backend.application import HUB_SERVICE_HANDLERS, RequestHandler
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SERVICE_TOKEN = "hub-admin-token-for-tests"
|
|
|
|
|
|
class FakeHeaders(dict):
|
|
def get(self, name, default=""): # type: ignore[override]
|
|
return super().get(name, default)
|
|
|
|
|
|
class HubAdminBridgeTests(unittest.TestCase):
|
|
def handler(self, path: str, token: str, calls: list[str]) -> RequestHandler:
|
|
handler = RequestHandler.__new__(RequestHandler)
|
|
handler.path = path
|
|
handler.headers = FakeHeaders({"X-Hub-Admin-Token": token} if token else {})
|
|
handler.responses = []
|
|
handler.send_json = lambda payload, status=HTTPStatus.OK, headers=None: (
|
|
handler.responses.append((status, payload))
|
|
)
|
|
handler.require_auth = lambda: calls.append("auth") or True
|
|
handler.require_csrf = lambda: calls.append("csrf") or True
|
|
handler.require_access = lambda method, route: calls.append("access") or True
|
|
return handler
|
|
|
|
def test_every_bridge_path_has_a_real_handler(self) -> None:
|
|
for path, handler_name in HUB_SERVICE_HANDLERS.items():
|
|
with self.subTest(path=path):
|
|
self.assertTrue(path.startswith("/api/hub-admin/"))
|
|
self.assertTrue(callable(getattr(RequestHandler, handler_name)))
|
|
|
|
def test_bridge_paths_stay_out_of_the_browser_route_registry(self) -> None:
|
|
registry = json.loads(
|
|
(ROOT / "config" / "api.config.json").read_text(encoding="utf-8")
|
|
)
|
|
registered = {route["path"] for route in registry["routes"]}
|
|
for path in HUB_SERVICE_HANDLERS:
|
|
with self.subTest(path=path):
|
|
self.assertNotIn(path, registered)
|
|
self.assertIsNone(RequestHandler.route_registry.resolve("POST", path))
|
|
|
|
def test_missing_or_wrong_service_token_is_rejected(self) -> None:
|
|
for token in ("", "wrong-token"):
|
|
with self.subTest(token=token), mock.patch.dict(
|
|
"os.environ", {"HUB_ADMIN_TOKEN": SERVICE_TOKEN}
|
|
):
|
|
calls: list[str] = []
|
|
handler = self.handler("/api/hub-admin/status", token, calls)
|
|
handler.hub_system_status = lambda: calls.append("dispatched")
|
|
|
|
RequestHandler.do_POST(handler)
|
|
|
|
self.assertEqual(calls, [])
|
|
status, payload = handler.responses[-1]
|
|
self.assertEqual(status, HTTPStatus.UNAUTHORIZED)
|
|
self.assertIn("服务令牌", payload["error"])
|
|
|
|
def test_unset_server_token_refuses_every_bridge_call(self) -> None:
|
|
with mock.patch.dict("os.environ", {"HUB_ADMIN_TOKEN": ""}):
|
|
calls: list[str] = []
|
|
handler = self.handler("/api/hub-admin/status", SERVICE_TOKEN, calls)
|
|
handler.hub_system_status = lambda: calls.append("dispatched")
|
|
|
|
RequestHandler.do_POST(handler)
|
|
|
|
self.assertEqual(calls, [])
|
|
self.assertEqual(handler.responses[-1][0], HTTPStatus.UNAUTHORIZED)
|
|
|
|
def test_valid_service_token_dispatches_without_session_guards(self) -> None:
|
|
for path, handler_name in HUB_SERVICE_HANDLERS.items():
|
|
with self.subTest(path=path), mock.patch.dict(
|
|
"os.environ", {"HUB_ADMIN_TOKEN": SERVICE_TOKEN}
|
|
):
|
|
calls: list[str] = []
|
|
handler = self.handler(path, SERVICE_TOKEN, calls)
|
|
setattr(handler, handler_name, lambda: calls.append(handler_name))
|
|
|
|
RequestHandler.do_POST(handler)
|
|
|
|
self.assertEqual(calls, [handler_name])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|