diff --git a/RUN.md b/RUN.md index c49fa37..f708b7b 100644 --- a/RUN.md +++ b/RUN.md @@ -144,7 +144,16 @@ - **未绑定成员发消息 → 回绑定指引**(不再静默):聊天优先布局下新员工第一动作就是打字,回调对未绑定成员的 text/图片/文件消息每条回一句"先去控制台绑定"(事件不回)。未绑定成员点菜单「工作台」则落在绑定提示页(不自动建号)。 - **channel 长会话上下文(微信/企业微信通用,0019)**:常驻会话不再无限膨胀。① **自动分段**——入站时距上次消息超过 `config.json` 的 `channel.session_gap_hours`(默 **6** 小时,设 `<=0` 关闭)→ 软重置:只把「最后一条 user 消息起」喂模型(保留上一轮做续聊锚点),之前的历史仍全留 DB,网页端照旧翻完整记录;② **手动新话题**——用户在微信/企业微信里直接发「新话题 / 新会话 / `/new` / 清空上下文」→ 硬重置,彻底从零(回执提示已归档)。两者都**不删任何消息**,只移动「喂给模型的窗口起点」`tasks.context_base_idx`。网页端「清空对话」(`POST /v1/tasks/{id}/clear`)仍整清并把 base 归 0。需 `main.py db upgrade head` 带上 `0019`。 - **PG**:`ZCBOT_DB_URL` 必填。本地 docker compose / 远端 dev / 生产任选;未设置时启动清晰报错,不引导 docker(§7.4)。 -- **测试库(可选,`ZCBOT_TEST_DB_URL`)**:DB 级单测(`tests/test_usage_report.py` / `tests/test_scheduler.py`)**只认这个显式变量、绝不回退 `.env` 的 `ZCBOT_DB_URL`**——后者可能经隧道指向生产库,测试插入的到点 job 会被生产实例调度守护真跑一次(2026-07-23 实锤)。未设则这两组自动 skip;要跑就建个专用库(`createdb zcbot_test` + `env ZCBOT_DB_URL=<测试库> main.py db upgrade head`)后 `set ZCBOT_TEST_DB_URL=...`。 +- **测试库(可选,`ZCBOT_TEST_DB_URL`)**:DB 级单测(`tests/test_usage_report.py` / `tests/test_scheduler.py` / `tests/test_web_routes_db.py`)**只认这个显式变量、绝不回退 `.env` 的 `ZCBOT_DB_URL`**——后者可能经隧道指向生产库,测试插入的到点 job 会被生产实例调度守护真跑一次(2026-07-23 实锤)。未设则这几组自动 skip。一键起库(docker,端口 5433 避开本地 5432): + ```bash + docker run -d --name zcbot-test-pg -e POSTGRES_PASSWORD=zcbot_test \ + -e POSTGRES_DB=zcbot_test -p 127.0.0.1:5433:5432 postgres:16-alpine + ZCBOT_DB_URL="postgresql+psycopg://postgres:zcbot_test@127.0.0.1:5433/zcbot_test" \ + .venv/Scripts/python.exe main.py db upgrade head + ZCBOT_TEST_DB_URL="postgresql+psycopg://postgres:zcbot_test@127.0.0.1:5433/zcbot_test" \ + .venv/Scripts/python.exe -m unittest discover -s tests -p "test_*.py" + ``` + 容器留着复用即可(alpine 版很轻);schema 升级后重跑一次 `db upgrade head`。 - **Auth env**:`PLATFORM_KEY` + `JWT_SECRET` 任一缺失 web 启动 fail-fast。生成随机串:`python -c "import secrets; print(secrets.token_urlsafe(48))"`。 - **用户管理**(`users.email/password_hash/role`,0005 UNIQUE(email)、0009 role):dev SPA 登录后端。发用户两条路径任选:CLI `main.py user add`(下方),或在登录页右下角"+ 管理员添加用户"链接(需先设 `ZCBOT_ADMIN_TOKEN` env,弹窗输入 email/密码/管理员口令/角色)。撤用户 `DELETE FROM users WHERE email=...`(先 DELETE 该 user 的 tasks)。**用户自助改密**:登录后顶栏「改密码」按钮(走 `POST /v1/auth/change_password`,需知道旧密码);改邮箱 / 用户忘了旧密码无法自助 → 手动 SQL(见故障兜底)。 - **角色与管理后台**(`users.role` ∈ `user`/`admin`):admin 才显顶栏"管理"入口 → `/static/admin.html`(非 admin 403)。页面:左侧目录(点击滚到对应区)+ 运行态/任务/用户用量/按模型/各用户用量/存储;「按模型」「各用户用量」支持时间筛选(全部/近7天/近30天)+ 排序(按成本/按用量),「各用户用量」「存储」分页;顶栏「导出 PDF」走浏览器打印(在打印对话框选"另存为 PDF",列表取前 10)。提管理员 `main.py user role --email X --role admin`(改完即时生效,role 走 DB 查不进 JWT)。`ZCBOT_ADMIN_TOKEN` 是另一回事(发用户共享口令),与 role 互不相干。 diff --git a/tests/test_web_routes_db.py b/tests/test_web_routes_db.py new file mode 100644 index 0000000..0867cba --- /dev/null +++ b/tests/test_web_routes_db.py @@ -0,0 +1,244 @@ +"""web /v1 路由测试(DB 面)—— 补 test_web_routes_nodb 留待的 DB-aware 分支。 + +仅当显式设 `ZCBOT_TEST_DB_URL` 才跑(同 test_usage_report 门控纪律,绝不回退 +.env —— 它可能指生产);一次性测试库:docker run postgres + main.py db upgrade, +见 RUN.md「测试库」段。 + +覆盖: +- tasks CRUD 全链(建/列/详/改/软删/恢复/同 wd 共享)+ folders +- files 顶层目录 DB-aware:rename 级联改 tasks.working_dir、running→409、 + move 被引用→409、递归删被引用→409、软删后放行 +- upload(磁盘配额 gate 路径)+ 根目录列表(system_wd_names DB 查询) +- clear / cancel 的状态闸;schedules 404 路径;/v1/models 档位过滤 + +隔离:测试专属 User 行 + 随机 uid 子树,teardown DB 行与 FS 整树删除; +POST messages / optimize_prompt(会起真 LLM)不在此测。 +""" +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])) + +os.environ.setdefault("PLATFORM_KEY", "test-platform-key-db") +os.environ.setdefault("JWT_SECRET", "test-jwt-secret-db") + + +def _test_db_ready() -> bool: + """只认显式 ZCBOT_TEST_DB_URL(教训见 test_usage_report 同名函数)。""" + url = os.environ.get("ZCBOT_TEST_DB_URL", "").strip() + if not url: + return False + os.environ["ZCBOT_DB_URL"] = url + return True + + +try: + if not _test_db_ready(): + raise RuntimeError("ZCBOT_TEST_DB_URL 未设") + from core.storage import session_scope + from core.storage.models import Message, Task, UsageEvent, User + + with session_scope() as _s: + _s.execute(__import__("sqlalchemy").select(1)) + _DB_OK = True +except Exception: + _DB_OK = False + +if _DB_OK: + from starlette.testclient import TestClient + from web.app import create_app + from web.auth import AuthConfig, mint_token + + _app = create_app() + _client = TestClient(_app) # 不进 with:不跑 lifespan + _UID = uuid.uuid4() + _TOKEN, _ = mint_token(AuthConfig.from_env(), _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 setUpModule() -> None: + if not _DB_OK: + raise unittest.SkipTest("ZCBOT_TEST_DB_URL 未设或测试库不可达(绝不用 .env 的库)") + with session_scope() as s: + s.add(User(user_id=_UID, email=f"test-routes-db-{_UID.hex[:8]}@invalid.local")) + + +def tearDownModule() -> None: + if not _DB_OK: + return + from sqlalchemy import delete, select + with session_scope() as s: + tids = s.execute(select(Task.task_id).where(Task.user_id == _UID)).scalars().all() + if tids: + s.execute(delete(Message).where(Message.task_id.in_(tids))) + s.execute(delete(UsageEvent).where(UsageEvent.user_id == _UID)) + s.execute(delete(Task).where(Task.user_id == _UID)) + s.execute(delete(User).where(User.user_id == _UID)) + d = _user_root() + if d.is_dir(): + shutil.rmtree(d, ignore_errors=True) + + +def _set_run_status(task_id: str, status: str) -> None: + from sqlalchemy import update + with session_scope() as s: + s.execute(update(Task).where(Task.task_id == uuid.UUID(task_id)).values(run_status=status)) + + +class TasksCrudTests(unittest.TestCase): + def test_create_list_get_patch_softdelete_restore(self): + r = _client.post("/v1/tasks", json={"name": "路由测试任务"}, headers=_AUTH) + self.assertEqual(r.status_code, 201, r.text) + t = r.json() + tid = t["task_id"] + self.assertEqual(t["name"], "路由测试任务") + self.assertTrue(t["working_dir"].endswith("/路由测试任务")) + self.assertTrue(t["model_profile"]) # 默认模型已填 + self.assertEqual(t["run_status"], "idle") + + # 列表含它,分页壳完整 + r = _client.get("/v1/tasks", headers=_AUTH) + body = r.json() + self.assertGreaterEqual(body["count"], 1) + self.assertIn(tid, {x["task_id"] for x in body["results"]}) + + # 详情带上下文压力字段 + d = _client.get(f"/v1/tasks/{tid}", headers=_AUTH).json() + self.assertIn("context_window_chars", d) + self.assertIn("context_folds", d) + + # PATCH:非法 status 400;description 生效 + self.assertEqual( + _client.patch(f"/v1/tasks/{tid}", json={"status": "active"}, headers=_AUTH).status_code, 400) + d = _client.patch(f"/v1/tasks/{tid}", json={"description": "备注"}, headers=_AUTH).json() + self.assertEqual(d["description"], "备注") + + # 软删 → 列表消失;恢复 → 回来 + self.assertEqual(_client.delete(f"/v1/tasks/{tid}", headers=_AUTH).status_code, 204) + self.assertNotIn(tid, {x["task_id"] for x in _client.get("/v1/tasks", headers=_AUTH).json()["results"]}) + self.assertEqual(_client.post(f"/v1/tasks/{tid}/restore", headers=_AUTH).status_code, 200) + self.assertIn(tid, {x["task_id"] for x in _client.get("/v1/tasks", headers=_AUTH).json()["results"]}) + + # 非法 id → 404 + self.assertEqual(_client.get("/v1/tasks/not-a-uuid", headers=_AUTH).status_code, 404) + + def test_shared_working_dir_and_folders(self): + for name in ("共享目录甲", "共享目录乙"): + r = _client.post("/v1/tasks", json={"name": name, "working_dir": "共享项目"}, headers=_AUTH) + self.assertEqual(r.status_code, 201, r.text) + folders = _client.get("/v1/folders", headers=_AUTH).json()["folders"] + hit = [f for f in folders if f["name"] == "共享项目"] + self.assertEqual(len(hit), 1) + self.assertGreaterEqual(hit[0]["n_tasks"], 2) + + def test_clear_and_cancel_gates(self): + tid = _client.post("/v1/tasks", json={"name": "状态闸任务"}, headers=_AUTH).json()["task_id"] + # idle 时 cancel → 409 + self.assertEqual(_client.post(f"/v1/tasks/{tid}/cancel", headers=_AUTH).status_code, 409) + # running 时 clear → 409;回 idle 后 clear → 200 且归零 + _set_run_status(tid, "running") + self.assertEqual(_client.post(f"/v1/tasks/{tid}/clear", headers=_AUTH).status_code, 409) + _set_run_status(tid, "idle") + d = _client.post(f"/v1/tasks/{tid}/clear", headers=_AUTH).json() + self.assertEqual((d["n_messages"], d["tokens"]), (0, 0)) + # 空任务 messages / outline + m = _client.get(f"/v1/tasks/{tid}/messages", headers=_AUTH).json() + self.assertEqual(m["messages"], []) + self.assertFalse(m["has_more"]) + self.assertEqual(_client.get(f"/v1/tasks/{tid}/outline", headers=_AUTH).json()["items"], []) + + +class FilesDbAwareTests(unittest.TestCase): + """顶层目录 = task.working_dir 的 DB-aware 分支(§7.4 唯一 mutation 入口)。""" + + def _mk_task(self, name: str, wd: str) -> str: + r = _client.post("/v1/tasks", json={"name": name, "working_dir": wd}, headers=_AUTH) + self.assertEqual(r.status_code, 201, r.text) + return r.json()["task_id"] + + def test_toplevel_rename_cascades_db(self): + tid = self._mk_task("改名任务", "改名前目录") + r = _client.post("/v1/files/rename", + json={"path": "改名前目录", "new_name": "改名后目录"}, headers=_AUTH) + self.assertEqual(r.status_code, 200, r.text) + self.assertEqual(r.json()["tasks_updated"], 1) + # DB 级联生效 + d = _client.get(f"/v1/tasks/{tid}", headers=_AUTH).json() + self.assertTrue(d["working_dir"].endswith("/改名后目录")) + self.assertTrue((_user_root() / "改名后目录").is_dir()) + + def test_toplevel_rename_blocked_while_running(self): + tid = self._mk_task("跑动中任务", "跑动中目录") + _set_run_status(tid, "running") + try: + r = _client.post("/v1/files/rename", + json={"path": "跑动中目录", "new_name": "别名"}, headers=_AUTH) + self.assertEqual(r.status_code, 409) + finally: + _set_run_status(tid, "idle") + + def test_toplevel_move_and_recursive_delete_blocked(self): + tid = self._mk_task("被引用任务", "被引用目录") + (_user_root() / "搬运目标").mkdir(exist_ok=True) + # 被 task 引用的顶层目录:move → 409;递归删 → 409 + r = _client.post("/v1/files/move", + json={"paths": ["被引用目录"], "dest_dir": "搬运目标"}, headers=_AUTH) + self.assertEqual(r.status_code, 409) + r = _client.post("/v1/files/delete", + json={"path": "被引用目录", "recursive": True}, headers=_AUTH) + self.assertEqual(r.status_code, 409) + # 软删 task 后不再算引用 → 递归删放行 + self.assertEqual(_client.delete(f"/v1/tasks/{tid}", headers=_AUTH).status_code, 204) + r = _client.post("/v1/files/delete", + json={"path": "被引用目录", "recursive": True}, headers=_AUTH) + self.assertEqual(r.status_code, 200, r.text) + + def test_upload_and_root_listing(self): + # upload 走磁盘配额 gate(无扫描记录 → 放行);根目录列表走 system_wd_names DB 查询 + r = _client.post("/v1/files/upload", data={"path": "上传目录"}, + files={"files": ("hello.txt", b"hi", "text/plain")}, headers=_AUTH) + self.assertEqual(r.status_code, 200, r.text) + self.assertEqual(r.json()["count"], 1) + names = {e["name"] for e in _client.get("/v1/files", headers=_AUTH).json()["entries"]} + self.assertIn("上传目录", names) + # 非法文件名 → 400 + r = _client.post("/v1/files/upload", data={"path": "上传目录"}, + files={"files": ("../escape.txt", b"x", "text/plain")}, headers=_AUTH) + self.assertEqual(r.status_code, 400) + + +class MiscDbRoutesTests(unittest.TestCase): + def test_models_list_nonempty(self): + r = _client.get("/v1/models", headers=_AUTH) + self.assertEqual(r.status_code, 200) + models = r.json()["models"] + self.assertTrue(models) + self.assertTrue(any(m["is_default"] for m in models)) + + def test_me_profile(self): + d = _client.get("/v1/me", headers=_AUTH).json() + self.assertEqual(d["user_id"], str(_UID)) + self.assertEqual(d["role"], "user") + + def test_schedules_empty_and_404(self): + self.assertEqual(_client.get("/v1/schedules", headers=_AUTH).json(), {"results": []}) + self.assertEqual( + _client.patch(f"/v1/schedules/{uuid.uuid4()}", json={"enabled": False}, headers=_AUTH).status_code, + 404, + ) + self.assertEqual(_client.delete(f"/v1/schedules/{uuid.uuid4()}", headers=_AUTH).status_code, 404) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/routers/tasks.py b/web/routers/tasks.py index 22a3620..58ed8d0 100644 --- a/web/routers/tasks.py +++ b/web/routers/tasks.py @@ -391,7 +391,13 @@ def register_task_routes(app, *, require_user) -> None: select(func.count()).select_from(Message).where(Message.task_id == tid) ).scalar_one() usage = usage_aggregates(s, [tid]) - return task_dict(row, n_messages=n, usage=usage.get(tid)) + # 序列化必须在 session 内:updated_at 是 server-side onupdate,flush 后被 + # 标记 expired,出了 session 再读会 DetachedInstanceError(真软删过的恢复 + # 路径 500;幂等路径无脏标记所以此前没暴露 —— test_web_routes_db 抓出)。 + s.flush() + s.refresh(row) + d = task_dict(row, n_messages=n, usage=usage.get(tid)) + return d @app.patch("/v1/tasks/{task_id}", tags=["tasks"]) def patch_task(