host
ce7fb3afd6
- cloud/tree.py: structured worktree tree (dirs first, ignore
.git/node_modules, symlinks not followed, entry/depth
truncation flags, text render for injection)
- GET /tasks/{id}/tree with /files auth + worktree resolution,
max_depth 1..5, max_entries 1..500, escape 400, missing 404,
cloud/cross-host 409
- MCP get_task_tree translating to the new endpoint
- worker first-round injection of a repo-checkout top-level
snapshot (2 layers, 120 entries) after repo resolution,
labelled as checkout snapshot not worktree; skipped for
coordinator/resume/fork, silent when no host source
- docs/project-tree.md; tests for tree, endpoint, injection,
MCP translation
111 lines
4.5 KiB
Python
111 lines
4.5 KiB
Python
"""worker 首轮目录树注入单测：只改内存副本，跳过规则与来源。
|
||
|
||
不跑真实模型、不派发：直接调 _apply_tree_injection，Worker
|
||
用 test_worker 同款最小构造（db 在临时目录、假 host.py）。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.server import init_db
|
||
from cloud.worker import Worker
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.server import init_db
|
||
from cloud.worker import Worker
|
||
|
||
|
||
def _completed_runner(cmd):
|
||
from types import SimpleNamespace
|
||
return SimpleNamespace(returncode=0, stdout="ok", stderr="")
|
||
|
||
|
||
class TreeInjectionTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
init_db(self.db_path)
|
||
self.repo = os.path.join(self.tmp.name, "repo")
|
||
os.makedirs(os.path.join(self.repo, "pkg"))
|
||
Path(self.repo, "pkg", "mod.py").write_text("x", encoding="utf-8")
|
||
Path(self.repo, "README.md").write_text("hi", encoding="utf-8")
|
||
os.makedirs(os.path.join(self.repo, ".git"))
|
||
self.worker = Worker(
|
||
self.db_path, "/fake/host.py",
|
||
runner=_completed_runner,
|
||
env={"XCM_API_KEY": "test-placeholder"})
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _task(self, **over):
|
||
task = {"id": "t1", "project_id": "p1", "prompt": "做一件事",
|
||
"repo": self.repo}
|
||
task.update(over)
|
||
return task
|
||
|
||
def test_injects_snapshot_and_keeps_prompt(self):
|
||
out = self.worker._apply_tree_injection(self._task())
|
||
self.assertTrue(out["_tree_injected"])
|
||
self.assertTrue(out["prompt"].endswith("做一件事"))
|
||
self.assertIn("[项目目录结构：仓库检出顶层快照", out["prompt"])
|
||
self.assertIn("pkg/", out["prompt"])
|
||
self.assertIn("mod.py", out["prompt"])
|
||
self.assertNotIn(".git", out["prompt"])
|
||
# 原任务不被改动：注入只改内存副本。
|
||
original = self._task()
|
||
self.worker._apply_tree_injection(original)
|
||
self.assertEqual(original["prompt"], "做一件事")
|
||
self.assertNotIn("_tree_injected", original)
|
||
|
||
def test_goal_task_updates_goal_too(self):
|
||
out = self.worker._apply_tree_injection(
|
||
self._task(goal_mode=1, goal="目标"))
|
||
self.assertIn("[项目目录结构", out["goal"])
|
||
self.assertTrue(out["goal"].endswith("目标"))
|
||
|
||
def test_skip_coordinator_resume_fork_and_no_source(self):
|
||
for over in ({"coordinator_parts": [{"name": "a"}]},
|
||
{"_resume": True},
|
||
{"_goal_resume_breakpoint": True},
|
||
{"forked_from": "parent"}):
|
||
out = self.worker._apply_tree_injection(self._task(**over))
|
||
self.assertEqual(out["prompt"], "做一件事", over)
|
||
self.assertNotIn("_tree_injected", out)
|
||
out = self.worker._apply_tree_injection(
|
||
self._task(repo="/nonexistent/repo"))
|
||
self.assertEqual(out["prompt"], "做一件事")
|
||
self.assertNotIn("_tree_injected", out)
|
||
|
||
def test_tenant_missing_checkout_with_repo_url_never_falls_back(self):
|
||
# 租户 + 有 repo_url + 检出缺失时，绝不回落到 test_repo
|
||
# 模板快照（那会误导）；repo_url 为空时才回落。
|
||
from cloud.sandbox import DockerExecutor
|
||
test_repo = os.path.join(self.tmp.name, "test-repo")
|
||
os.makedirs(test_repo)
|
||
Path(test_repo, "sample.py").write_text("x", encoding="utf-8")
|
||
self.worker.tenant_isolation = True
|
||
self.worker.executor = DockerExecutor.__new__(DockerExecutor)
|
||
self.worker.tenants_root = os.path.join(self.tmp.name, "tenants")
|
||
self.worker._tenant_dirs = lambda pid: {"test_repo": test_repo}
|
||
task = self._task(repo="/var/lib/codex-app-server/repos/p1")
|
||
self.worker._project_repo_url = lambda pid: "https://example.invalid/r.git"
|
||
self.assertIsNone(self.worker._tree_snapshot_source(task))
|
||
self.worker._project_repo_url = lambda pid: None
|
||
self.assertEqual(self.worker._tree_snapshot_source(task), test_repo)
|
||
|
||
def test_empty_repo_no_section(self):
|
||
empty = os.path.join(self.tmp.name, "empty")
|
||
os.makedirs(empty)
|
||
out = self.worker._apply_tree_injection(self._task(repo=empty))
|
||
self.assertEqual(out["prompt"], "做一件事")
|
||
self.assertNotIn("_tree_injected", out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|