"""GET /tasks/{id}/files 端点测试：本地 worktree 文件读取。 只用标准库 http.client，不跑真实模型，不联网。 任务行直接插库（不走 POST /tasks，避免 worker 派发）， worktree 用 server.host_worktrees_dir 注入到临时目录。 """ 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 TaskFilesTestCase(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-files-1", executor=None) wt = os.path.join(self.worktrees, "task-files-1") os.makedirs(os.path.join(wt, "sub")) Path(wt, "report.md").write_text("# Report\n", encoding="utf-8") Path(wt, "sub", "page.html").write_text("

hi

\n", encoding="utf-8") # 非法 UTF-8 字节（\x00 是合法 UTF-8，不能用来测二进制拒绝）。 Path(wt, "blob.bin").write_bytes(b"\x89PNG\r\n\x1a\n\xff\xfe") Path(wt, "big.txt").write_text("x" * (256 * 1024 + 1), 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-docker-1", executor="docker") tenant_wt = os.path.join( self.tenants, self.project["id"], "work", "host-worktrees", "task-docker-1") os.makedirs(tenant_wt) Path(tenant_wt, "report.md").write_text("Total: 85.15\n", encoding="utf-8") self._task("task-docker-nodir", executor="docker") 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 _files(self, task_id, path=None, key=None): url = f"/tasks/{task_id}/files" if path is not None: url += "?path=" + quote(path) return self._req("GET", url, key=self.owner_key if key is None else key) def test_list_top_level_recursive(self): status, body = self._files("task-files-1") self.assertEqual(status, 200, body) paths = sorted(e["path"] for e in body["entries"]) self.assertIn("report.md", paths) self.assertIn("sub/page.html", paths) self.assertNotIn("escape", paths) # 符号链接不跟进列表 def test_read_file_and_subdir_listing(self): status, body = self._files("task-files-1", "report.md") self.assertEqual(status, 200, body) self.assertEqual(body["content"], "# Report\n") status, body = self._files("task-files-1", "sub") self.assertEqual(status, 200, body) self.assertEqual([e["path"] for e in body["entries"]], ["sub/page.html"]) def test_path_escape_blocked(self): for bad in ("../outside.txt", "/etc/passwd", "escape", "sub/../../x"): status, body = self._files("task-files-1", bad) self.assertIn(status, (400, 404), (bad, body)) # escape 是指到 worktree 外的符号链接：必须 400，不是读到 secret。 status, body = self._files("task-files-1", "escape") self.assertEqual(status, 400, body) def test_missing_file_and_missing_worktree(self): status, _ = self._files("task-files-1", "nope.md") self.assertEqual(status, 404) status, _ = self._files("task-nowt-1") self.assertEqual(status, 404) status, _ = self._files("task-unknown") self.assertEqual(status, 404) def test_binary_and_oversized(self): status, body = self._files("task-files-1", "blob.bin") self.assertEqual(status, 409, body) status, body = self._files("task-files-1", "big.txt") self.assertEqual(status, 413, body) def test_tenant_docker_task_files(self): # 租户 Docker 任务走 tenants 下的任务级子目录，不再 409。 status, body = self._files("task-docker-1") self.assertEqual(status, 200, body) self.assertEqual([e["path"] for e in body["entries"]], ["report.md"]) status, body = self._files("task-docker-1", "report.md") self.assertEqual(status, 200, body) self.assertEqual(body["content"], "Total: 85.15\n") def test_docker_task_without_worktree_404_and_cloud_409(self): status, _ = self._files("task-docker-nodir") self.assertEqual(status, 404) status, body = self._files("task-cloud-1") self.assertEqual(status, 409, body) def test_auth(self): status, _ = self._req("GET", "/tasks/task-files-1/files") self.assertEqual(status, 401) status, _ = self._files("task-files-1", key=self.outsider_key) self.assertEqual(status, 403) status, _ = self._files("task-files-1", key=self.viewer_key) self.assertEqual(status, 200) if __name__ == "__main__": unittest.main()