"""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// 子树 teardown 整树删除;TestClient 不进 with(不跑 lifespan → 不碰 DB/后台协程)。 """ from __future__ import annotations import os import shutil import sys import unittest import uuid from pathlib import Path 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 _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_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/tasks/00000000-0000-0000-0000-000000000000/messages"), ("POST", "/v1/tasks"), ("POST", "/v1/asr/transcribe"), ("GET", "/v1/admin/overview"), ] 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 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 / "a.txt").write_text("hello", encoding="utf-8") (cls.wd / "sub" / "b.txt").write_text("world", encoding="utf-8") 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", "a.txt"}) self.assertEqual(body["current"], "route-test-wd") 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()