Python · 2560 bytes Raw Blame History
1 """Tests for runtime-owned rollback planning."""
2
3 from __future__ import annotations
4
5 import pytest
6
7 from loader.runtime.rollback import (
8 RollbackPlan,
9 RollbackType,
10 create_rollback_plan_for_action,
11 execute_rollback,
12 get_undo_command,
13 is_destructive_tool,
14 )
15
16
17 def test_rollback_plan_formats_reverse_steps() -> None:
18 plan = RollbackPlan()
19 plan.add_file_creation("new.txt")
20 plan.add_file_modification("config.json", '{"a":1}')
21
22 assert plan.get_rollback_steps() == [
23 "Restore: config.json",
24 "Delete: new.txt",
25 ]
26 assert plan.to_prompt() == (
27 "Rollback plan:\n"
28 " 1. Restore: config.json\n"
29 " 2. Delete: new.txt"
30 )
31
32
33 def test_get_undo_command_handles_common_install_commands() -> None:
34 assert get_undo_command("mkdir docs") == "rmdir docs"
35 assert get_undo_command("npm install react") == "npm uninstall react"
36 assert get_undo_command("pip install pytest") == "pip uninstall -y pytest"
37
38
39 def test_is_destructive_tool_covers_patch_and_bash_patterns() -> None:
40 assert is_destructive_tool("patch", {"file_path": "notes.txt"}) is True
41 assert is_destructive_tool("bash", {"command": "git checkout -- README.md"}) is True
42 assert is_destructive_tool("bash", {"command": "ls -la"}) is False
43
44
45 @pytest.mark.asyncio
46 async def test_create_rollback_plan_for_new_write_returns_delete_action(tmp_path) -> None:
47 target = tmp_path / "notes.txt"
48
49 async def read_file(_path: str) -> str:
50 raise AssertionError("new files should not be read for rollback")
51
52 action = await create_rollback_plan_for_action(
53 "write",
54 {"file_path": str(target), "content": "alpha\n"},
55 read_file,
56 )
57
58 assert action is not None
59 assert action.type == RollbackType.FILE_DELETE
60 assert action.file_path == str(target)
61
62
63 @pytest.mark.asyncio
64 async def test_execute_rollback_restores_file_and_marks_action(tmp_path) -> None:
65 target = tmp_path / "notes.txt"
66 plan = RollbackPlan()
67 plan.add_file_modification(str(target), "restored\n")
68
69 writes: list[tuple[str, str]] = []
70 commands: list[str] = []
71
72 async def write_file(path: str, content: str) -> None:
73 writes.append((path, content))
74
75 async def run_command(command: str) -> None:
76 commands.append(command)
77
78 results = await execute_rollback(plan, write_file, run_command)
79
80 assert results == [f"✓ Restored: {target}"]
81 assert writes == [(str(target), "restored\n")]
82 assert commands == []
83 assert plan.actions[0].executed is True