diff --git a/tests/test_web_routes_nodb.py b/tests/test_web_routes_nodb.py new file mode 100644 index 0000000..fb55527 --- /dev/null +++ b/tests/test_web_routes_nodb.py @@ -0,0 +1,239 @@ +"""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()