"""GET /tasks/{id}/tree 端点测试：结构化目录树，不读文件内容。 harness 同 test_server_task_files：任务行直接插库，worktree 用 server.host_worktrees_dir / tenants_root 注入到临时目录。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import unittest from pathlib import Path from urllib.parse import quote try: from cloud.auth import AuthStore from cloud.server import create_server except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.server import create_server class TaskTreeTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.viewer_id, self.viewer_key = self.auth.create_user("viewer") self.outsider_id, self.outsider_key = self.auth.create_user("outsider") self.server = create_server(self.db_path, "127.0.0.1", 0) self.worktrees = os.path.join(self.tmp.name, "host-worktrees") os.makedirs(self.worktrees) self.server.host_worktrees_dir = self.worktrees self.tenants = os.path.join(self.tmp.name, "tenants") self.server.tenants_root = self.tenants self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, self.project = self._req( "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201) self.auth.set_member( self.project["id"], self.owner_id, self.viewer_id, "viewer") self._task("task-tree-1", executor=None) wt = os.path.join(self.worktrees, "task-tree-1") os.makedirs(os.path.join(wt, "sub", "deep")) Path(wt, "report.md").write_text("# Report\n", encoding="utf-8") Path(wt, "sub", "page.html").write_text("

hi

\n", encoding="utf-8") Path(wt, "sub", "deep", "leaf.txt").write_text("leaf\n", encoding="utf-8") os.makedirs(os.path.join(wt, ".git")) Path(wt, ".git", "config").write_text("x", encoding="utf-8") os.symlink(os.path.join(self.tmp.name, "outside.txt"), os.path.join(wt, "escape")) Path(self.tmp.name, "outside.txt").write_text("secret\n", encoding="utf-8") self._task("task-cloud-1", executor="cloud") self._task("task-nowt-1", executor=None) def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() def _task(self, task_id, executor): conn = sqlite3.connect(self.db_path) try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status, executor)" " VALUES(?, ?, ?, ?, ?)", (task_id, self.project["id"], "do it", "completed", executor)) conn.commit() finally: conn.close() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) hdrs = {} payload = None if body is not None: payload = json.dumps(body).encode("utf-8") hdrs["Content-Type"] = "application/json" if key is not None: hdrs["X-API-Key"] = key try: conn.request(method, path, body=payload, headers=hdrs) resp = conn.getresponse() data = resp.read() finally: conn.close() try: parsed = json.loads(data.decode("utf-8")) if data else None except ValueError: parsed = None return resp.status, parsed def _tree(self, task_id, query=None, key=None): url = f"/tasks/{task_id}/tree" if query: url += "?" + "&".join( "%s=%s" % (k, quote(str(v))) for k, v in query.items()) return self._req("GET", url, key=self.owner_key if key is None else key) def test_root_tree_nested_no_content_no_git(self): status, body = self._tree("task-tree-1") self.assertEqual(status, 200, body) self.assertEqual(body["task_id"], "task-tree-1") self.assertFalse(body["truncated"]) names = [(e["name"], e["type"]) for e in body["entries"]] self.assertIn(("sub", "dir"), names) self.assertIn(("report.md", "file"), names) self.assertIn(("escape", "symlink"), names) self.assertNotIn(".git", [e["name"] for e in body["entries"]]) # 不读文件内容：整棵树里没有文件正文。 self.assertNotIn("# Report", json.dumps(body)) sub = [e for e in body["entries"] if e["name"] == "sub"][0] deep = [e for e in sub["children"] if e["name"] == "deep"][0] self.assertEqual(deep["children"][0]["path"], "sub/deep/leaf.txt") def test_subdir_and_depth_and_entries_cap(self): status, body = self._tree("task-tree-1", {"path": "sub"}) self.assertEqual(status, 200, body) self.assertEqual(body["path"], "sub") status, body = self._tree("task-tree-1", {"max_depth": "1"}) self.assertEqual(status, 200, body) sub = [e for e in body["entries"] if e["name"] == "sub"][0] self.assertEqual(sub["children"], []) self.assertTrue(sub["children_truncated"]) status, body = self._tree("task-tree-1", {"max_entries": "1"}) self.assertEqual(status, 200, body) self.assertTrue(body["truncated"]) self.assertEqual(len(body["entries"]), 1) def test_bad_params_escape_missing_file_cloud(self): for query in ({"max_depth": "x"}, {"max_depth": "0"}, {"max_depth": "6"}, {"max_entries": "0"}, {"max_entries": "501"}): status, _ = self._tree("task-tree-1", query) self.assertEqual(status, 400, query) for path in ("../outside.txt", "/etc/passwd", "escape"): status, _ = self._tree("task-tree-1", {"path": path}) self.assertEqual(status, 400, path) status, _ = self._tree("task-tree-1", {"path": "nope"}) self.assertEqual(status, 404) status, _ = self._tree("task-tree-1", {"path": "report.md"}) self.assertEqual(status, 404) status, _ = self._tree("task-nowt-1") self.assertEqual(status, 404) status, _ = self._tree("task-cloud-1") self.assertEqual(status, 409) def test_auth(self): status, _ = self._req("GET", "/tasks/task-tree-1/tree") self.assertEqual(status, 401) status, _ = self._tree("task-tree-1", key=self.outsider_key) self.assertEqual(status, 403) status, _ = self._tree("task-tree-1", key=self.viewer_key) self.assertEqual(status, 200) if __name__ == "__main__": unittest.main()