jim-bo/cli-textual-demo

1"""Tests for the write_file and edit_file tools.2 3Covers the pure functions (cli_textual.tools.{write_file,edit_file}) including4the ported Aider forgiving matcher, plus the manager.py wrappers' event5lifecycle (called directly with a mock ctx, like test_agent_tools.py).6"""7import asyncio8 9import pytest10from unittest.mock import MagicMock11 12pytestmark = pytest.mark.timeout(5)13 14from cli_textual.agents.manager import edit_file as edit_file_wrapper15from cli_textual.agents.manager import write_file as write_file_wrapper16from cli_textual.core.chat_events import (17    AgentToolEnd,18    AgentToolOutput,19    AgentToolStart,20    ChatDeps,21)22from cli_textual.tools.base import ToolResult23from cli_textual.tools.edit_file import edit_file24from cli_textual.tools.write_file import write_file25 26 27# ---------------------------------------------------------------------------28# Helpers29# ---------------------------------------------------------------------------30 31def make_ctx() -> tuple:32    event_queue: asyncio.Queue = asyncio.Queue()33    input_queue: asyncio.Queue = asyncio.Queue()34    deps = ChatDeps(event_queue=event_queue, input_queue=input_queue)35    ctx = MagicMock()36    ctx.deps = deps37    return ctx, event_queue38 39 40async def drain(q: asyncio.Queue) -> list:41    items = []42    while not q.empty():43        items.append(q.get_nowait())44    return items45 46 47# ---------------------------------------------------------------------------48# write_file (pure)49# ---------------------------------------------------------------------------50 51@pytest.mark.asyncio52async def test_write_file_creates_new_file(tmp_path):53    result = await write_file("hello.py", "print('hi')\n", workspace_root=tmp_path)54    assert isinstance(result, ToolResult)55    assert not result.is_error56    assert (tmp_path / "hello.py").read_text() == "print('hi')\n"57    assert "Created" in result.output58 59 60@pytest.mark.asyncio61async def test_write_file_creates_parent_dirs(tmp_path):62    result = await write_file("a/b/c.txt", "deep", workspace_root=tmp_path)63    assert not result.is_error64    assert (tmp_path / "a" / "b" / "c.txt").read_text() == "deep"65 66 67@pytest.mark.asyncio68async def test_write_file_overwrites_existing(tmp_path):69    (tmp_path / "f.txt").write_text("old")70    result = await write_file("f.txt", "new", workspace_root=tmp_path)71    assert not result.is_error72    assert "Updated" in result.output73    assert (tmp_path / "f.txt").read_text() == "new"74 75 76@pytest.mark.asyncio77async def test_write_file_rejects_path_outside_workspace(tmp_path):78    result = await write_file("../escape.txt", "x", workspace_root=tmp_path)79    assert result.is_error80    assert "outside workspace" in result.output81    assert not (tmp_path.parent / "escape.txt").exists()82 83 84# ---------------------------------------------------------------------------85# edit_file (pure) — exact semantics86# ---------------------------------------------------------------------------87 88@pytest.mark.asyncio89async def test_edit_file_unique_replace(tmp_path):90    f = tmp_path / "m.py"91    f.write_text("def foo():\n    return 1\n")92    result = await edit_file("m.py", "return 1", "return 2", workspace_root=tmp_path)93    assert not result.is_error94    assert f.read_text() == "def foo():\n    return 2\n"95 96 97@pytest.mark.asyncio98async def test_edit_file_not_found_errors(tmp_path):99    f = tmp_path / "m.py"100    f.write_text("a = 1\n")101    result = await edit_file("m.py", "nonexistent text", "x", workspace_root=tmp_path)102    assert result.is_error103    assert "not found" in result.output104    assert f.read_text() == "a = 1\n"  # unchanged105 106 107@pytest.mark.asyncio108async def test_edit_file_multiple_matches_errors(tmp_path):109    f = tmp_path / "m.py"110    f.write_text("x = 1\nx = 1\n")111    result = await edit_file("m.py", "x = 1", "x = 2", workspace_root=tmp_path)112    assert result.is_error113    assert "2 matches" in result.output114    assert f.read_text() == "x = 1\nx = 1\n"  # unchanged115 116 117@pytest.mark.asyncio118async def test_edit_file_replace_all(tmp_path):119    f = tmp_path / "m.py"120    f.write_text("x = 1\nx = 1\nx = 1\n")121    result = await edit_file("m.py", "x = 1", "x = 2", replace_all=True, workspace_root=tmp_path)122    assert not result.is_error123    assert f.read_text() == "x = 2\nx = 2\nx = 2\n"124 125 126@pytest.mark.asyncio127async def test_edit_file_delete_text(tmp_path):128    f = tmp_path / "m.py"129    f.write_text("keep\nremove me\nkeep\n")130    result = await edit_file("m.py", "remove me\n", "", workspace_root=tmp_path)131    assert not result.is_error132    assert f.read_text() == "keep\nkeep\n"133 134 135@pytest.mark.asyncio136async def test_edit_file_identical_strings_errors(tmp_path):137    f = tmp_path / "m.py"138    f.write_text("a = 1\n")139    result = await edit_file("m.py", "a = 1", "a = 1", workspace_root=tmp_path)140    assert result.is_error141    assert "identical" in result.output142 143 144@pytest.mark.asyncio145async def test_edit_file_empty_old_string_errors(tmp_path):146    f = tmp_path / "m.py"147    f.write_text("a = 1\n")148    result = await edit_file("m.py", "", "x", workspace_root=tmp_path)149    assert result.is_error150    assert "empty" in result.output151    assert f.read_text() == "a = 1\n"  # unchanged152 153 154@pytest.mark.asyncio155async def test_edit_file_non_utf8_errors(tmp_path):156    f = tmp_path / "bin.dat"157    f.write_bytes(b"\xff\xfe\x00bad bytes")158    result = await edit_file("bin.dat", "bad", "good", workspace_root=tmp_path)159    assert result.is_error160    assert "UTF-8" in result.output161    assert f.read_bytes() == b"\xff\xfe\x00bad bytes"  # untouched162 163 164@pytest.mark.asyncio165async def test_edit_file_missing_file_errors(tmp_path):166    result = await edit_file("ghost.py", "a", "b", workspace_root=tmp_path)167    assert result.is_error168    assert "not found" in result.output169 170 171@pytest.mark.asyncio172async def test_edit_file_rejects_path_outside_workspace(tmp_path):173    result = await edit_file("../escape.py", "a", "b", workspace_root=tmp_path)174    assert result.is_error175    assert "outside workspace" in result.output176 177 178# ---------------------------------------------------------------------------179# edit_file (pure) — forgiving matcher (ported from Aider)180# ---------------------------------------------------------------------------181 182@pytest.mark.asyncio183async def test_edit_file_tolerates_leading_whitespace_drift(tmp_path):184    """old_string given with no indentation still matches an indented block."""185    f = tmp_path / "m.py"186    f.write_text("class A:\n    def foo(self):\n        return 1\n")187    # Model supplies the body without the 8-space indentation.188    result = await edit_file(189        "m.py",190        "def foo(self):\n    return 1",191        "def foo(self):\n    return 99",192        workspace_root=tmp_path,193    )194    assert not result.is_error, result.output195    assert "return 99" in f.read_text()196 197 198@pytest.mark.asyncio199async def test_edit_file_handles_dotdotdot_elision(tmp_path):200    """`...` between anchors lets the model elide the unchanged middle."""201    f = tmp_path / "m.py"202    f.write_text("start\nline1\nline2\nline3\nend\n")203    result = await edit_file(204        "m.py",205        "start\n...\nend",206        "START\n...\nEND",207        workspace_root=tmp_path,208    )209    assert not result.is_error, result.output210    text = f.read_text()211    assert text.startswith("START")212    assert "END" in text213    assert "line2" in text  # middle preserved214 215 216# ---------------------------------------------------------------------------217# Wrapper event lifecycle218# ---------------------------------------------------------------------------219 220@pytest.mark.asyncio221async def test_write_file_wrapper_emits_lifecycle(tmp_path, monkeypatch):222    monkeypatch.chdir(tmp_path)223    ctx, event_queue = make_ctx()224    await write_file_wrapper(ctx, path="new.txt", content="hi")225    events = await drain(event_queue)226    types = [type(e) for e in events]227    assert AgentToolStart in types228    assert AgentToolOutput in types229    assert AgentToolEnd in types230    start = next(i for i, e in enumerate(events) if isinstance(e, AgentToolStart))231    out = next(i for i, e in enumerate(events) if isinstance(e, AgentToolOutput))232    end = next(i for i, e in enumerate(events) if isinstance(e, AgentToolEnd))233    assert start < out < end234    assert (tmp_path / "new.txt").read_text() == "hi"235 236 237@pytest.mark.asyncio238async def test_edit_file_wrapper_emits_lifecycle_and_error_status(tmp_path, monkeypatch):239    monkeypatch.chdir(tmp_path)240    (tmp_path / "e.txt").write_text("a = 1\n")241    ctx, event_queue = make_ctx()242    # No-match edit → wrapper should still emit lifecycle, with error status.243    await edit_file_wrapper(ctx, path="e.txt", old_string="zzz", new_string="b")244    events = await drain(event_queue)245    end = next(e for e in events if isinstance(e, AgentToolEnd))246    assert end.result == "error"247    output = next(e for e in events if isinstance(e, AgentToolOutput))248    assert output.is_error249