409 lines
17 KiB
Python
409 lines
17 KiB
Python
"""web /v1 路由测试(无 DB 面)—— 拆分后首批路由级回归。
|
|
|
|
覆盖三类不碰 DB 的行为(handler 里带 DB 查询的端点只测 401 门 —— 依赖先于
|
|
handler 执行,永远到不了 DB):
|
|
- 公开端点:/healthz、/v1/changelog、/ 302
|
|
- 鉴权门:11 个 router 的代表端点无 token 一律 401;坏 platform_key 403;
|
|
platform_key 对但 user_id 非法 400(在 ensure_user_row 落库之前就拦下)
|
|
- FS 类业务:kb 建/详/删、skills 列/详/删门、memory 只读、files 子目录
|
|
列/下载/改名/删/拷(顶层目录与 upload 走 DB,留给 ZCBOT_TEST_DB_URL 套件)
|
|
|
|
隔离纪律:token 用测试自设的 PLATFORM_KEY/JWT_SECRET 直签(先于 import web.app
|
|
写进 env,litellm 的 dotenv 副作用不会覆盖已存在值);user_id 随机,产生的
|
|
workspace/users/<uid>/ 子树 teardown 整树删除;TestClient 不进 with(不跑
|
|
lifespan → 不碰 DB/后台协程)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
# 必须先于 import web.app 设好(AuthConfig.from_env fail-fast)。注意全量跑时更早
|
|
# 导入的模块可能已触发 litellm 的 dotenv 副作用把 .env 真值灌进 env —— setdefault
|
|
# 不覆盖,所以下面一律用 AuthConfig.from_env()(与 app 同源),不假设具体 key 值。
|
|
os.environ.setdefault("PLATFORM_KEY", "test-platform-key-nodb")
|
|
os.environ.setdefault("JWT_SECRET", "test-jwt-secret-nodb")
|
|
|
|
from starlette.testclient import TestClient # noqa: E402
|
|
|
|
from web.app import create_app # noqa: E402
|
|
from web.auth import AuthConfig, mint_token # noqa: E402
|
|
from web.schemas import TaskCreateRequest # noqa: E402
|
|
|
|
_app = create_app()
|
|
_client = TestClient(_app) # 刻意不进 with:不跑 lifespan(reaper/调度都不启动)
|
|
|
|
_CFG = AuthConfig.from_env()
|
|
_UID = uuid.uuid4()
|
|
_TOKEN, _ = mint_token(_CFG, _UID)
|
|
_AUTH = {"Authorization": f"Bearer {_TOKEN}"}
|
|
|
|
|
|
def _user_root() -> Path:
|
|
from core.agent_builder import resolve_workspace, user_root
|
|
return user_root(resolve_workspace(None), _UID)
|
|
|
|
|
|
def tearDownModule() -> None:
|
|
d = _user_root()
|
|
if d.is_dir():
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
|
|
class PublicEndpointTests(unittest.TestCase):
|
|
def test_task_create_schema_accepts_working_dir_without_name(self):
|
|
body = TaskCreateRequest(working_dir="低碳水泥项目")
|
|
self.assertEqual(body.name, "")
|
|
self.assertEqual(body.working_dir, "低碳水泥项目")
|
|
self.assertFalse(body.auto_title) # 自动命名由路由据空 name 推断
|
|
|
|
def test_healthz(self):
|
|
r = _client.get("/healthz")
|
|
self.assertEqual(r.status_code, 200)
|
|
body = r.json()
|
|
self.assertEqual(body["status"], "ok")
|
|
self.assertIn("version", body)
|
|
self.assertIn("brand", body)
|
|
|
|
def test_changelog_public(self):
|
|
r = _client.get("/v1/changelog?limit=2")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertIn("entries", r.json())
|
|
# limit 越界 clamp,不 500
|
|
self.assertEqual(_client.get("/v1/changelog?limit=9999").status_code, 200)
|
|
|
|
def test_root_redirects_to_dev_spa(self):
|
|
r = _client.get("/", follow_redirects=False)
|
|
self.assertEqual(r.status_code, 302)
|
|
self.assertEqual(r.headers["location"], "/static/dev.html")
|
|
|
|
|
|
class AuthGateTests(unittest.TestCase):
|
|
# 每个 router 至少一个代表端点:无 token 必须 401(鉴权门接线回归 ——
|
|
# 拆分 router 后若哪个模块忘了 Depends(require_user),这里当场红)
|
|
_PROTECTED = [
|
|
("GET", "/v1/me"),
|
|
("GET", "/v1/tasks"),
|
|
("GET", "/v1/channel_tasks"),
|
|
("GET", "/v1/folders"),
|
|
("GET", "/v1/models"),
|
|
("GET", "/v1/image_models"),
|
|
("GET", "/v1/skills"),
|
|
("GET", "/v1/memory"),
|
|
("GET", "/v1/kb"),
|
|
("GET", "/v1/schedules"),
|
|
("GET", "/v1/files"),
|
|
("GET", "/v1/user/storage"),
|
|
("GET", "/v1/procs"),
|
|
("GET", "/v1/wechat/bind"),
|
|
("GET", "/v1/wecom/bind"),
|
|
("GET", "/v1/external-system-providers"),
|
|
("GET", "/v1/external-systems"),
|
|
("GET", "/v1/tasks/00000000-0000-0000-0000-000000000000/messages"),
|
|
("POST", "/v1/tasks"),
|
|
("POST", "/v1/asr/transcribe"),
|
|
("GET", "/v1/admin/overview"),
|
|
("GET", "/v1/admin/tool-wire-health"),
|
|
("GET", "/v1/admin/external-system-definitions"),
|
|
("GET", "/v1/admin/external-system-users"),
|
|
]
|
|
|
|
def test_protected_endpoints_401_without_token(self):
|
|
for method, path in self._PROTECTED:
|
|
with self.subTest(path=path):
|
|
r = _client.request(method, path)
|
|
self.assertEqual(r.status_code, 401, f"{method} {path} -> {r.status_code}")
|
|
|
|
def test_garbage_token_401(self):
|
|
r = _client.get("/v1/me", headers={"Authorization": "Bearer not-a-jwt"})
|
|
self.assertEqual(r.status_code, 401)
|
|
|
|
def test_login_bad_platform_key_403(self):
|
|
r = _client.post("/v1/auth/login", json={"user_id": str(uuid.uuid4()), "platform_key": "wrong"})
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_login_bad_user_id_400(self):
|
|
# platform_key 对(与 app 同源读 env)、user_id 非 UUID:在 ensure_user_row
|
|
# 落库之前就 400(仅进程内请求,不外发)
|
|
r = _client.post("/v1/auth/login", json={"user_id": "not-a-uuid", "platform_key": _CFG.platform_key})
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
|
|
class ExternalSystemRoutesTests(unittest.TestCase):
|
|
def test_provider_catalog_and_create_are_user_scoped(self):
|
|
provider = {
|
|
"provider": "factory_mes",
|
|
"title": "Factory MES",
|
|
"configured": True,
|
|
}
|
|
with patch("web.routers.external_systems.provider_catalog", return_value=[provider]) as catalog:
|
|
r = _client.get("/v1/external-system-providers", headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json()["providers"], [provider])
|
|
catalog.assert_called_once_with(_UID)
|
|
|
|
created = {"external_system_id": str(uuid.uuid4()), "username_masked": "me***r"}
|
|
definition_id = uuid.uuid4()
|
|
with patch("web.routers.external_systems.create_external_system", return_value=created) as create:
|
|
r = _client.post(
|
|
"/v1/external-systems",
|
|
headers=_AUTH,
|
|
json={
|
|
"definition_id": str(definition_id),
|
|
"name": "Factory MES",
|
|
"username": "mes-user",
|
|
"password": "secret",
|
|
},
|
|
)
|
|
self.assertEqual(r.status_code, 201)
|
|
self.assertEqual(r.json(), created)
|
|
self.assertEqual(create.call_args.args[0], _UID)
|
|
self.assertEqual(create.call_args.kwargs["definition_id"], definition_id)
|
|
|
|
def test_create_accepts_dynamic_credentials(self):
|
|
created = {"external_system_id": str(uuid.uuid4()), "username_masked": "***"}
|
|
definition_id = uuid.uuid4()
|
|
with patch("web.routers.external_systems.create_external_system", return_value=created) as create:
|
|
response = _client.post(
|
|
"/v1/external-systems",
|
|
headers=_AUTH,
|
|
json={
|
|
"definition_id": str(definition_id),
|
|
"name": "LIMS",
|
|
"credentials": {"api_key": "secret-key"},
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(create.call_args.kwargs["credentials"], {"api_key": "secret-key"})
|
|
|
|
def test_invalid_connection_id_is_not_forwarded_to_service(self):
|
|
with patch("web.routers.external_systems.test_external_system") as test_connection:
|
|
r = _client.post("/v1/external-systems/not-a-uuid/test", headers=_AUTH)
|
|
self.assertEqual(r.status_code, 404)
|
|
test_connection.assert_not_called()
|
|
|
|
|
|
class KbRoutesTests(unittest.TestCase):
|
|
def test_kb_crud_fs_only(self):
|
|
self.assertEqual(_client.get("/v1/kb", headers=_AUTH).json(), {"results": []})
|
|
# 建库
|
|
r = _client.post("/v1/kb", json={"name": "测试库"}, headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json()["name"], "测试库")
|
|
# 非法名(路径字符)→ 400
|
|
self.assertEqual(_client.post("/v1/kb", json={"name": "../逃逸"}, headers=_AUTH).status_code, 400)
|
|
# 详情(空库,ingest 进度为内存态)
|
|
r = _client.get("/v1/kb/测试库", headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertIn("ingest", r.json())
|
|
# 不存在的库 / 单篇 → 404
|
|
self.assertEqual(_client.get("/v1/kb/不存在", headers=_AUTH).status_code, 404)
|
|
self.assertEqual(_client.get("/v1/kb/测试库/docs/nope.md", headers=_AUTH).status_code, 404)
|
|
# 删库
|
|
self.assertEqual(_client.delete("/v1/kb/测试库", headers=_AUTH).json(), {"deleted": "测试库"})
|
|
self.assertEqual(_client.delete("/v1/kb/测试库", headers=_AUTH).status_code, 404)
|
|
|
|
|
|
class SkillMemoryRoutesTests(unittest.TestCase):
|
|
def test_skills_list_and_detail(self):
|
|
r = _client.get("/v1/skills", headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
names = {s["name"] for s in r.json()["skills"]}
|
|
self.assertIn("ppt", names) # 内置 skill 该在
|
|
detail = _client.get("/v1/skills/ppt", headers=_AUTH)
|
|
self.assertEqual(detail.status_code, 200)
|
|
self.assertTrue(detail.json()["content"].strip())
|
|
self.assertEqual(_client.get("/v1/skills/不存在的", headers=_AUTH).status_code, 404)
|
|
|
|
def test_builtin_skill_not_deletable(self):
|
|
# 只能删 user 源;对内置返回 404(等同"用户那里没有这个可删的")
|
|
self.assertEqual(_client.delete("/v1/skills/ppt", headers=_AUTH).status_code, 404)
|
|
|
|
def test_memory_readonly_view(self):
|
|
r = _client.get("/v1/memory", headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(_client.get("/v1/memory/extended/nope.md", headers=_AUTH).status_code, 404)
|
|
# 路径穿越文件名 → 404(校验收口 core/memory.py)
|
|
self.assertEqual(_client.get("/v1/memory/extended/..%2Fcore.md", headers=_AUTH).status_code, 404)
|
|
|
|
|
|
class FilesRoutesTests(unittest.TestCase):
|
|
"""files 的纯 FS 面:子目录列表 / 下载 / 越界 / 非顶层改名 / 删 / 拷。
|
|
顶层目录 rename/delete(DB-aware)与 upload(磁盘配额)留给 DB 套件。"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.wd = _user_root() / "route-test-wd"
|
|
(cls.wd / "sub").mkdir(parents=True, exist_ok=True)
|
|
(cls.wd / "figures" / ".meta").mkdir(parents=True, exist_ok=True)
|
|
(cls.wd / "figures" / ".private").mkdir(parents=True, exist_ok=True)
|
|
(cls.wd / "a.txt").write_text("hello", encoding="utf-8")
|
|
(cls.wd / "sub" / "b.txt").write_text("world", encoding="utf-8")
|
|
(cls.wd / "figures" / ".meta" / "image.png.json").write_text(
|
|
"{}", encoding="utf-8"
|
|
)
|
|
(cls.wd / "figures" / ".env").write_text("SECRET=x", encoding="utf-8")
|
|
(_user_root() / ".root-private").mkdir(exist_ok=True)
|
|
|
|
def test_list_subdir(self):
|
|
r = _client.get("/v1/files", params={"path": "route-test-wd"}, headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
body = r.json()
|
|
self.assertTrue(body["exists"])
|
|
names = {e["name"] for e in body["entries"]}
|
|
self.assertEqual(names, {"sub", "figures", "a.txt"})
|
|
self.assertEqual(body["current"], "route-test-wd")
|
|
|
|
def test_dot_directories_are_hidden_by_default(self):
|
|
r = _client.get(
|
|
"/v1/files",
|
|
params={"path": "route-test-wd/figures"},
|
|
headers=_AUTH,
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json()["entries"], [])
|
|
|
|
def test_enumerator_can_include_dot_directories_but_never_dotfiles(self):
|
|
from web.userfiles import enumerate_files
|
|
|
|
entries, _crumbs, _exists = enumerate_files(
|
|
_user_root(),
|
|
self.wd / "figures",
|
|
include_hidden_dirs=True,
|
|
)
|
|
self.assertEqual(
|
|
{entry["name"] for entry in entries},
|
|
{".meta", ".private"},
|
|
)
|
|
|
|
def test_api_hidden_toggle_is_limited_to_selected_task_tree(self):
|
|
task_id = uuid.uuid4()
|
|
with patch(
|
|
"web.routers.files._task_working_dir",
|
|
return_value=(task_id, self.wd),
|
|
):
|
|
shown = _client.get(
|
|
"/v1/files",
|
|
params={
|
|
"path": "route-test-wd/figures",
|
|
"include_hidden": "true",
|
|
"task_id": str(task_id),
|
|
},
|
|
headers=_AUTH,
|
|
).json()
|
|
self.assertEqual(
|
|
{entry["name"] for entry in shown["entries"]},
|
|
{".meta", ".private"},
|
|
)
|
|
self.assertTrue(shown["hidden_dirs_included"])
|
|
|
|
with patch("web.routers.files.system_wd_names", return_value=set()):
|
|
root = _client.get(
|
|
"/v1/files",
|
|
params={"include_hidden": "true", "task_id": str(task_id)},
|
|
headers=_AUTH,
|
|
).json()
|
|
self.assertNotIn(
|
|
".root-private",
|
|
{entry["name"] for entry in root["entries"]},
|
|
)
|
|
self.assertFalse(root["hidden_dirs_included"])
|
|
|
|
def test_create_directory_at_root_and_in_current_directory(self):
|
|
root_name = "mkdir-api-root"
|
|
root_dir = _user_root() / root_name
|
|
try:
|
|
created = _client.post(
|
|
"/v1/files/mkdir",
|
|
json={"path": "", "name": f" {root_name} "},
|
|
headers=_AUTH,
|
|
)
|
|
self.assertEqual(created.status_code, 200)
|
|
self.assertEqual(created.json()["path"], root_name)
|
|
self.assertTrue(root_dir.is_dir())
|
|
|
|
child = _client.post(
|
|
"/v1/files/mkdir",
|
|
json={"path": root_name, "name": "资料"},
|
|
headers=_AUTH,
|
|
)
|
|
self.assertEqual(child.status_code, 200)
|
|
self.assertEqual(child.json()["path"], f"{root_name}/资料")
|
|
self.assertTrue((root_dir / "资料").is_dir())
|
|
|
|
duplicate = _client.post(
|
|
"/v1/files/mkdir",
|
|
json={"path": root_name, "name": "资料"},
|
|
headers=_AUTH,
|
|
)
|
|
self.assertEqual(duplicate.status_code, 409)
|
|
|
|
for name in (".hidden", "../escape", "a/b"):
|
|
with self.subTest(name=name):
|
|
invalid = _client.post(
|
|
"/v1/files/mkdir",
|
|
json={"path": root_name, "name": name},
|
|
headers=_AUTH,
|
|
)
|
|
self.assertEqual(invalid.status_code, 400)
|
|
finally:
|
|
shutil.rmtree(root_dir, ignore_errors=True)
|
|
|
|
def test_download_and_errors(self):
|
|
r = _client.get("/v1/files/download", params={"path": "route-test-wd/a.txt"}, headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.content, b"hello")
|
|
self.assertEqual(
|
|
_client.get("/v1/files/download", params={"path": "route-test-wd/nope.txt"}, headers=_AUTH).status_code,
|
|
404,
|
|
)
|
|
self.assertEqual(
|
|
_client.get("/v1/files/download", params={"path": "route-test-wd"}, headers=_AUTH).status_code,
|
|
400, # 目录不是文件
|
|
)
|
|
|
|
def test_path_escape_rejected(self):
|
|
for bad in ("../x", "/etc/passwd", "..\\x"):
|
|
with self.subTest(path=bad):
|
|
r = _client.get("/v1/files/download", params={"path": bad}, headers=_AUTH)
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
def test_rename_delete_copy_nontop(self):
|
|
(self.wd / "sub" / "c.txt").write_text("c", encoding="utf-8")
|
|
# 非顶层改名:纯 FS,tasks_updated=0
|
|
r = _client.post("/v1/files/rename",
|
|
json={"path": "route-test-wd/sub/c.txt", "new_name": "c2.txt"}, headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json()["tasks_updated"], 0)
|
|
self.assertTrue((self.wd / "sub" / "c2.txt").is_file())
|
|
# 拷贝到子目录内(非顶层,无 DB 闸)
|
|
(self.wd / "dest").mkdir(exist_ok=True)
|
|
r = _client.post("/v1/files/copy",
|
|
json={"paths": ["route-test-wd/sub/c2.txt"], "dest_dir": "route-test-wd/dest"},
|
|
headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertTrue((self.wd / "dest" / "c2.txt").is_file())
|
|
# 目标已存在 → 409(预检整批 abort)
|
|
r = _client.post("/v1/files/copy",
|
|
json={"paths": ["route-test-wd/sub/c2.txt"], "dest_dir": "route-test-wd/dest"},
|
|
headers=_AUTH)
|
|
self.assertEqual(r.status_code, 409)
|
|
# 删文件
|
|
r = _client.post("/v1/files/delete", json={"path": "route-test-wd/dest/c2.txt"}, headers=_AUTH)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertFalse((self.wd / "dest" / "c2.txt").exists())
|
|
# 删非空目录不带 recursive → 400
|
|
r = _client.post("/v1/files/delete", json={"path": "route-test-wd/sub"}, headers=_AUTH)
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|