331 lines
13 KiB
Python
331 lines
13 KiB
Python
"""项目待办标记工具单测：todo-list / todo-create / todo-update
|
|
解析、sandbox 提取、worker 投递与首轮注入。
|
|
|
|
不跑真实模型、不联网：解析是纯函数；worker 投递与注入用
|
|
init_db 建的临时 SQLite 库，直接调 _deliver_marker_tools 与
|
|
_apply_todo_injection，不走 run_one 全量流程。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
try:
|
|
from cloud.db import connect as db_connect
|
|
from cloud.marker_tools import (
|
|
extract_todo_creates,
|
|
extract_todo_lists,
|
|
extract_todo_updates,
|
|
format_todo,
|
|
)
|
|
from cloud.sandbox import LocalExecutor, _parse_host_result
|
|
from cloud.server import init_db
|
|
from cloud.todos import create_todo, list_todos
|
|
from cloud.worker import Worker
|
|
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.db import connect as db_connect
|
|
from cloud.marker_tools import (
|
|
extract_todo_creates,
|
|
extract_todo_lists,
|
|
extract_todo_updates,
|
|
format_todo,
|
|
)
|
|
from cloud.sandbox import LocalExecutor, _parse_host_result
|
|
from cloud.server import init_db
|
|
from cloud.todos import create_todo, list_todos
|
|
from cloud.worker import Worker
|
|
|
|
FAKE_ENV = {"XCM_API_KEY": "test-placeholder"}
|
|
|
|
|
|
class ExtractTestCase(unittest.TestCase):
|
|
def test_list_default_and_alias(self):
|
|
out = extract_todo_lists("```todo-list\n{}\n```")
|
|
self.assertEqual(out, [{"status": None, "q": None}])
|
|
out = extract_todo_lists("```todo_list\n{\"status\": \"blocked\"}\n```")
|
|
self.assertEqual(out, [{"status": "blocked", "q": None}])
|
|
|
|
def test_list_invalid_status_skipped(self):
|
|
out = extract_todo_lists(
|
|
"```todo-list\n{\"status\": \"doing\"}\n```\n"
|
|
"```todo-list\n{\"status\": \"pending\"}\n```")
|
|
self.assertEqual(len(out), 1)
|
|
self.assertEqual(out[0]["status"], "pending")
|
|
|
|
def test_list_cap_five(self):
|
|
text = "\n".join("```todo-list\n{}\n```" for _ in range(6))
|
|
self.assertEqual(len(extract_todo_lists(text)), 5)
|
|
|
|
def test_create_valid_and_defaults(self):
|
|
out = extract_todo_creates(
|
|
"```todo-create\n{\"title\": \"做 X\"}\n```")
|
|
self.assertEqual(len(out), 1)
|
|
self.assertEqual(out[0]["title"], "做 X")
|
|
self.assertIsNone(out[0]["status"])
|
|
self.assertIsNone(out[0]["priority"])
|
|
|
|
def test_create_invalid_skipped(self):
|
|
cases = [
|
|
"{\"title\": \"\"}",
|
|
"{\"title\": \"x\", \"status\": \"doing\"}",
|
|
"{\"title\": \"x\", \"priority\": \"urgent\"}",
|
|
"{\"title\": \"x\", \"due_at\": \"明天\"}",
|
|
"{\"title\": \"x\", \"sort_order\": 1.5}",
|
|
"{\"description\": \"no title\"}",
|
|
]
|
|
for body in cases:
|
|
with self.subTest(body=body):
|
|
self.assertEqual(
|
|
extract_todo_creates("```todo-create\n%s\n```" % body),
|
|
[])
|
|
|
|
def test_create_due_ms_tolerance(self):
|
|
out = extract_todo_creates(
|
|
"```todo-create\n{\"title\": \"x\", \"due_at\": 1700000000000}\n```")
|
|
self.assertEqual(out[0]["due_at"], 1700000000.0)
|
|
|
|
def test_update_valid_and_empty_skipped(self):
|
|
out = extract_todo_updates(
|
|
"```todo-update\n{\"id\": \"abc\", \"status\": \"completed\"}\n```")
|
|
self.assertEqual(out, [{"id": "abc",
|
|
"fields": {"status": "completed"}}])
|
|
self.assertEqual(extract_todo_updates(
|
|
"```todo-update\n{\"id\": \"abc\"}\n```"), [])
|
|
self.assertEqual(extract_todo_updates(
|
|
"```todo-update\n{\"id\": \"abc\", \"status\": \"doing\"}\n```"), [])
|
|
self.assertEqual(extract_todo_updates(
|
|
"```todo-update\n{\"status\": \"completed\"}\n```"), [])
|
|
|
|
def test_update_unknown_field_ignored(self):
|
|
out = extract_todo_updates(
|
|
"```todo-update\n{\"id\": \"abc\", \"status\": \"blocked\","
|
|
" \"note\": \"x\"}\n```")
|
|
self.assertEqual(out[0]["fields"], {"status": "blocked"})
|
|
|
|
def test_format_todo_truncates_and_due_utc(self):
|
|
line = format_todo({"title": "x" * 120, "status": "pending",
|
|
"id": "t1", "priority": "high",
|
|
"due_at": 1700000000, "task_id": None})
|
|
self.assertIn("…", line)
|
|
self.assertIn("due=2023-11-14T22:13:20Z", line)
|
|
|
|
def test_sandbox_extracts_both_executors(self):
|
|
text = ("```todo-list\n{}\n```\n"
|
|
"```todo-create\n{\"title\": \"做 X\"}\n```\n"
|
|
"```todo-update\n{\"id\": \"t1\", \"status\": \"completed\"}"
|
|
"\n```")
|
|
fake = SimpleNamespace(
|
|
returncode=0,
|
|
stdout=json.dumps({"status": "completed", "text": text}),
|
|
stderr="")
|
|
local = LocalExecutor("/fake/host.py")._result_from_runner("t1", fake)
|
|
self.assertEqual(len(local["todo_lists"]), 1)
|
|
self.assertEqual(len(local["todo_creates"]), 1)
|
|
self.assertEqual(len(local["todo_updates"]), 1)
|
|
docker = _parse_host_result("t1", fake)
|
|
self.assertEqual(len(docker["todo_lists"]), 1)
|
|
self.assertEqual(len(docker["todo_creates"]), 1)
|
|
self.assertEqual(len(docker["todo_updates"]), 1)
|
|
|
|
|
|
class WorkerTodoTestCase(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._insert("PM", status="running", created_by="tester")
|
|
self._member("tester", "maintainer")
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def _insert(self, task_id, status="queued", created_by="tester",
|
|
project_id="p1", **cols):
|
|
now = time.time()
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO tasks(id, project_id, prompt, status,"
|
|
" idempotency_key, depends_on, priority, created_by,"
|
|
" created_at, updated_at, last_heartbeat)"
|
|
" VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)",
|
|
(task_id, project_id, "do work", status, created_by,
|
|
now, now))
|
|
for key, value in cols.items():
|
|
conn.execute(
|
|
"UPDATE tasks SET %s = ? WHERE id = ?" % key,
|
|
(value, task_id))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _member(self, user_id, role):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"CREATE TABLE IF NOT EXISTS users(id TEXT PRIMARY KEY,"
|
|
" username TEXT UNIQUE, api_key_hash TEXT,"
|
|
" created_at REAL)")
|
|
conn.execute(
|
|
"CREATE TABLE IF NOT EXISTS project_members("
|
|
"project_id TEXT, user_id TEXT, role TEXT,"
|
|
" PRIMARY KEY(project_id, user_id))")
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO project_members"
|
|
"(project_id, user_id, role) VALUES('p1', ?, ?)",
|
|
(user_id, role))
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO users(id, username, api_key_hash,"
|
|
" created_at) VALUES(?, ?, '', 0)", (user_id, user_id))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _worker(self):
|
|
return Worker(self.db_path, "/fake/host.py", env=FAKE_ENV)
|
|
|
|
def _seed_todo(self, title, **kw):
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
return create_todo(conn, "p1", "tester", title, **kw)
|
|
finally:
|
|
conn.close()
|
|
|
|
def _followup_bodies(self):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT body FROM task_followups"
|
|
" WHERE task_id = 'PM'").fetchall()
|
|
return [row[0] for row in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
def _todos(self):
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
return list_todos(conn, "p1")
|
|
finally:
|
|
conn.close()
|
|
|
|
def test_deliver_create_update_list(self):
|
|
worker = self._worker()
|
|
sender = worker.get_task("PM")
|
|
todo = self._seed_todo("旧项")
|
|
worker._deliver_marker_tools(sender, {
|
|
"status": "completed",
|
|
"todo_creates": [{"title": "新项", "description": None,
|
|
"status": None, "priority": "high",
|
|
"task_id": None, "due_at": None,
|
|
"sort_order": None}],
|
|
"todo_updates": [{"id": todo["id"],
|
|
"fields": {"status": "completed"}}],
|
|
"todo_lists": [{"status": None, "q": None}]})
|
|
titles = [t["title"] for t in self._todos()]
|
|
self.assertIn("新项", titles)
|
|
updated = [t for t in self._todos() if t["id"] == todo["id"]][0]
|
|
self.assertEqual(updated["status"], "completed")
|
|
self.assertIsNotNone(updated["completed_at"])
|
|
bodies = self._followup_bodies()
|
|
self.assertEqual(len(bodies), 1)
|
|
# 列表默认只返回未完成项：已完成的旧项不在列表段里，
|
|
# 新项在（更新报告行里提到旧项是正常的，不算列表）。
|
|
self.assertIn("新项", bodies[0])
|
|
list_section = bodies[0].split("待办列表", 1)[1]
|
|
self.assertNotIn("旧项", list_section)
|
|
|
|
def test_deliver_cross_project_update_rejected(self):
|
|
worker = self._worker()
|
|
sender = worker.get_task("PM")
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
other = create_todo(conn, "p2", "tester", "别项目")
|
|
finally:
|
|
conn.close()
|
|
worker._deliver_marker_tools(sender, {
|
|
"status": "completed",
|
|
"todo_creates": [], "todo_lists": [],
|
|
"todo_updates": [{"id": other["id"],
|
|
"fields": {"status": "completed"}}]})
|
|
bodies = self._followup_bodies()
|
|
self.assertEqual(len(bodies), 1)
|
|
self.assertIn("不存在", bodies[0])
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
kept = [t for t in list_todos(conn, "p2")
|
|
if t["id"] == other["id"]][0]
|
|
finally:
|
|
conn.close()
|
|
self.assertEqual(kept["status"], "pending")
|
|
|
|
def test_deliver_denied_without_write_permission(self):
|
|
self._member("viewer1", "viewer")
|
|
self._insert("PM2", status="running", created_by="viewer1")
|
|
worker = self._worker()
|
|
sender = worker.get_task("PM2")
|
|
worker._deliver_marker_tools(sender, {
|
|
"status": "completed",
|
|
"todo_creates": [{"title": "不应建", "description": None,
|
|
"status": None, "priority": None,
|
|
"task_id": None, "due_at": None,
|
|
"sort_order": None}],
|
|
"todo_lists": [], "todo_updates": []})
|
|
self.assertEqual(
|
|
[t["title"] for t in self._todos()], [])
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
bodies = [row[0] for row in conn.execute(
|
|
"SELECT body FROM task_followups"
|
|
" WHERE task_id = 'PM2'").fetchall()]
|
|
finally:
|
|
conn.close()
|
|
self.assertIn("无 create_task 权限", bodies[0])
|
|
|
|
def test_deliver_goal_sender_ignored(self):
|
|
self._insert("G1", status="running", goal_mode=1)
|
|
worker = self._worker()
|
|
sender = worker.get_task("G1")
|
|
worker._deliver_marker_tools(sender, {
|
|
"status": "completed",
|
|
"todo_creates": [{"title": "不应建", "description": None,
|
|
"status": None, "priority": None,
|
|
"task_id": None, "due_at": None,
|
|
"sort_order": None}],
|
|
"todo_lists": [], "todo_updates": []})
|
|
self.assertEqual([t["title"] for t in self._todos()], [])
|
|
|
|
def test_injection_lists_open_only(self):
|
|
self._seed_todo("待做", status="pending")
|
|
self._seed_todo("进行中", status="in_progress")
|
|
self._seed_todo("已做完", status="completed")
|
|
worker = self._worker()
|
|
task = {"id": "PM", "project_id": "p1", "prompt": "做事",
|
|
"created_by": "tester"}
|
|
out = worker._apply_todo_injection(task)
|
|
self.assertIn("待做", out["prompt"])
|
|
self.assertIn("进行中", out["prompt"])
|
|
self.assertNotIn("已做完", out["prompt"])
|
|
self.assertIn("todo-update", out["prompt"])
|
|
# 原 prompt 不落库：内存副本改了，调用方传的 dict 不动。
|
|
self.assertEqual(task["prompt"], "做事")
|
|
|
|
def test_injection_skips_goal_and_empty(self):
|
|
worker = self._worker()
|
|
goal = {"id": "G1", "project_id": "p1", "prompt": "目标",
|
|
"created_by": "tester", "goal_mode": 1}
|
|
self.assertIs(worker._apply_todo_injection(goal), goal)
|
|
empty = {"id": "PM", "project_id": "p1", "prompt": "做事",
|
|
"created_by": "tester"}
|
|
self.assertIs(worker._apply_todo_injection(empty), empty)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|