"""对话内 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()