215 lines
7.6 KiB
Python
215 lines
7.6 KiB
Python
"""对话内 working_dir 延迟改名：工具登记 + worker 收尾时序。"""
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
from uuid import uuid4
|
|
|
|
from core.task_actions import DeferredTaskActions
|
|
from core.working_dirs import (
|
|
WorkingDirBusyError,
|
|
WorkingDirRenameResult,
|
|
rename_working_dir,
|
|
)
|
|
from tools.rename_working_dir import RenameWorkingDirTool
|
|
|
|
|
|
class RenameWorkingDirToolTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tmp.name)
|
|
self.wd = self.root / "旧目录"
|
|
self.wd.mkdir()
|
|
self.actions = DeferredTaskActions()
|
|
self.tool = RenameWorkingDirTool(
|
|
self.actions,
|
|
working_dir=self.wd,
|
|
base_dir=self.wd,
|
|
user_root=self.root,
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.tmp.cleanup()
|
|
|
|
def test_registers_and_replaces_deferred_name(self) -> None:
|
|
out = self.tool.execute(" 新目录 ")
|
|
self.assertIn("[OK]", out)
|
|
self.assertEqual(self.actions.rename_working_dir_to, "新目录")
|
|
|
|
out = self.tool.execute("最终目录")
|
|
self.assertIn("更新为", out)
|
|
self.assertEqual(self.actions.rename_working_dir_to, "最终目录")
|
|
self.assertTrue(self.wd.is_dir(), "工具调用阶段不应立即改动文件系统")
|
|
|
|
def test_rejects_invalid_same_or_existing_name(self) -> None:
|
|
for bad in ("", ".memory", "a/b", "a\\b", "旧目录"):
|
|
with self.subTest(name=bad):
|
|
self.actions.rename_working_dir_to = ""
|
|
self.assertIn("[Error]", self.tool.execute(bad))
|
|
self.assertEqual(self.actions.rename_working_dir_to, "")
|
|
|
|
(self.root / "已存在").mkdir()
|
|
self.assertIn("[Error]", self.tool.execute("已存在"))
|
|
self.assertEqual(self.actions.rename_working_dir_to, "")
|
|
|
|
|
|
class DeferredRenameWorkerTests(unittest.TestCase):
|
|
def test_normal_run_renames_after_status_commit_before_done(self) -> None:
|
|
from web import runs
|
|
|
|
tid = uuid4()
|
|
uid = uuid4()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
old = Path(tmp) / "旧目录"
|
|
old.mkdir()
|
|
new = old.parent / "新目录"
|
|
actions = DeferredTaskActions(rename_working_dir_to="新目录")
|
|
agent = SimpleNamespace(
|
|
run=MagicMock(return_value="ok"),
|
|
deferred_actions=actions,
|
|
sink=None,
|
|
)
|
|
order: list[str] = []
|
|
|
|
@contextmanager
|
|
def fake_scope():
|
|
yield SimpleNamespace(execute=MagicMock())
|
|
order.append("status_committed")
|
|
|
|
broker = MagicMock()
|
|
|
|
def fake_rename(**kwargs):
|
|
order.append("renamed")
|
|
self.assertEqual(kwargs["old_path"], old)
|
|
self.assertEqual(kwargs["new_path"], new)
|
|
return WorkingDirRenameResult(old, new, 1)
|
|
|
|
with (
|
|
patch("core.agent_builder.build_agent", return_value=(
|
|
agent, MagicMock(), str(tid), MagicMock(), old,
|
|
)),
|
|
patch("core.agent_builder.sync_task_tokens"),
|
|
patch.object(runs, "session_scope", fake_scope),
|
|
patch.object(runs, "broker", broker),
|
|
patch("core.working_dirs.rename_working_dir", side_effect=fake_rename),
|
|
):
|
|
runs.run_agent_bg(tid, uid, "把目录改名")
|
|
|
|
self.assertEqual(order, ["status_committed", "renamed"])
|
|
broker.clear_cancel.assert_called_once_with(tid)
|
|
broker.close.assert_called_once_with(tid)
|
|
broker.emit.assert_any_call(
|
|
tid,
|
|
{
|
|
"type": "warn",
|
|
"level": "info",
|
|
"msg": "工作目录已重命名为 新目录",
|
|
},
|
|
)
|
|
|
|
def test_cancelled_run_does_not_apply_deferred_rename(self) -> None:
|
|
from web import runs
|
|
|
|
tid = uuid4()
|
|
uid = uuid4()
|
|
actions = DeferredTaskActions(rename_working_dir_to="新目录")
|
|
agent = SimpleNamespace(
|
|
run=MagicMock(return_value="[cancelled]"),
|
|
deferred_actions=actions,
|
|
sink=None,
|
|
)
|
|
|
|
@contextmanager
|
|
def fake_scope():
|
|
yield SimpleNamespace(execute=MagicMock())
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
old = Path(tmp) / "旧目录"
|
|
old.mkdir()
|
|
with (
|
|
patch("core.agent_builder.build_agent", return_value=(
|
|
agent, MagicMock(), str(tid), MagicMock(), old,
|
|
)),
|
|
patch("core.agent_builder.sync_task_tokens"),
|
|
patch.object(runs, "session_scope", fake_scope),
|
|
patch.object(runs, "broker", MagicMock()),
|
|
patch("core.working_dirs.rename_working_dir") as rename,
|
|
):
|
|
runs.run_agent_bg(tid, uid, "停止")
|
|
rename.assert_not_called()
|
|
|
|
|
|
class WorkingDirServiceTests(unittest.TestCase):
|
|
def test_updates_all_associated_tasks_and_renames_fs(self) -> None:
|
|
uid = uuid4()
|
|
tids = [uuid4(), uuid4()]
|
|
rows = [
|
|
SimpleNamespace(task_id=tid, run_status="idle")
|
|
for tid in tids
|
|
]
|
|
session = MagicMock()
|
|
session.execute.side_effect = [
|
|
SimpleNamespace(all=lambda: rows),
|
|
MagicMock(),
|
|
]
|
|
|
|
@contextmanager
|
|
def fake_scope():
|
|
yield session
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
old = Path(tmp) / "旧目录"
|
|
new = Path(tmp) / "新目录"
|
|
old.mkdir()
|
|
(old / "产物.txt").write_text("ok", encoding="utf-8")
|
|
with (
|
|
patch("core.working_dirs.session_scope", fake_scope),
|
|
patch("core.working_dirs.check_no_subtask"),
|
|
patch("core.working_dirs.to_db_path", side_effect=["old-db", "new-db"]),
|
|
):
|
|
result = rename_working_dir(
|
|
user_id=uid,
|
|
old_path=old,
|
|
new_path=new,
|
|
)
|
|
|
|
self.assertEqual(result.tasks_updated, 2)
|
|
self.assertFalse(old.exists())
|
|
self.assertEqual((new / "产物.txt").read_text(encoding="utf-8"), "ok")
|
|
self.assertEqual(session.execute.call_count, 2)
|
|
|
|
def test_active_associated_task_blocks_before_fs_change(self) -> None:
|
|
uid = uuid4()
|
|
rows = [SimpleNamespace(task_id=uuid4(), run_status="running")]
|
|
session = MagicMock()
|
|
session.execute.return_value = SimpleNamespace(all=lambda: rows)
|
|
|
|
@contextmanager
|
|
def fake_scope():
|
|
yield session
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
old = Path(tmp) / "旧目录"
|
|
new = Path(tmp) / "新目录"
|
|
old.mkdir()
|
|
with (
|
|
patch("core.working_dirs.session_scope", fake_scope),
|
|
patch("core.working_dirs.to_db_path", side_effect=["old-db", "new-db"]),
|
|
):
|
|
with self.assertRaises(WorkingDirBusyError):
|
|
rename_working_dir(
|
|
user_id=uid,
|
|
old_path=old,
|
|
new_path=new,
|
|
)
|
|
self.assertTrue(old.is_dir())
|
|
self.assertFalse(new.exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|