from __future__ import annotations import base64 import unittest from pathlib import Path from peer import Peer, output class TreeVerbTests(unittest.IsolatedAsyncioTestCase): """The tree verbs are bound in the cell and act on the session's workspace.""" async def test_verbs_are_bound_without_an_import_and_resolve_the_tree_once(self): async with Peer() as peer: tree = Path(peer.directory.name) (tree / "src").mkdir(exist_ok=True) (tree / "src" / "note.md").write_text("first line\nTODO: find me\n") seen = {"inspect": 0, "runs": []} async def service(message): request = message["request"] if request["method"] == "runtime.inspect": seen["inspect"] += 1 return { "kind": "ok", "value": { "source_root": {"path": str(tree), "exists": True, "writable": False} }, } if request["method"] == "workspace.run": seen["runs"].append(request) return { "kind": "ok", "value": { "argv": request["argv"], "cwd": request["cwd"], "exit_code": 0, "stdout_base64": base64.b64encode(b"clean\n").decode(), "stderr_base64": base64.b64encode(b"").decode(), "duration_ms": 3, "completed": True, "timed_out": False, "cancelled": False, "stdout_truncated": False, "stderr_truncated": False, }, } raise AssertionError(f"unexpected request: {request}") # No import line: these names are already bound in the cell. code = ( "found = await search('TODO: find me', 'src')\n" "print([(m.path.name, m.line) for m in found])\n" "note = await open_file('src/note.md')\n" "print(note.relative, (await note.text()).splitlines()[0])\n" "completed = await run(('git', 'status'), timeout=5)\n" "print(completed.exit_code, completed.stdout.decode().strip())\n" ) result = await peer.execute(code, service) self.assertEqual(result["status"], "ok", output(result)) printed = output(result) self.assertIn("('note.md', 2)", printed) self.assertIn("first line", printed) self.assertIn("0 clean", printed) # Three verbs ran; the workspace was resolved once and reused, so the host is not asked # for the same path on every call. self.assertEqual(seen["inspect"], 1) self.assertEqual(seen["runs"][0]["argv"], ["git", "status"]) self.assertEqual(seen["runs"][0]["cwd"], ".") async def test_a_patch_anchors_on_the_line_hashes_the_read_showed(self): # The anchor is copied out of the view, so the model never reproduces the line it changes; # both ends of a range are verified against the file as it is now. async with Peer() as peer: tree = Path(peer.directory.name) (tree / "f.py").write_text('def greet(name):\n msg = "hi"\n print(msg)\n') async def service(message): request = message["request"] if request["method"] == "runtime.inspect": return { "kind": "ok", "value": { "source_root": {"path": str(tree), "exists": True, "writable": False} }, } raise AssertionError(f"unexpected request: {request}") viewed = await peer.execute( "page = await open_file('f.py')\nprint(await page.hashlines())\n", service ) printed = output(viewed) anchor = next(line.split("|")[0] for line in printed.splitlines() if line.startswith("2:")) replaced = await peer.execute( f"page = await open_file('f.py')\nawait page.edit('{anchor}', ' msg = \"hello\"')\n", service, ) stale = await peer.execute( f"page = await open_file('f.py')\nawait page.edit('{anchor}', ' msg = \"again\"')\n", service, ) # replace takes both arguments: leaving `new` out is a TypeError, not a delete. missing_new = await peer.execute( "page = await open_file('f.py')\nawait page.replace(' msg = \"hello\"')\n", service ) tail = await peer.execute( "page = await open_file('f.py')\nprint((await page.text()).splitlines()[-1])\n", service ) # Read while the peer still owns the tree: its context manager removes it on exit. on_disk = (tree / "f.py").read_text() self.assertEqual("2:", anchor[:2], printed) self.assertEqual(replaced["status"], "ok", output(replaced)) self.assertEqual(on_disk, 'def greet(name):\n msg = "hello"\n print(msg)\n') # The line moved under it, so the same anchor is refused rather than applied again. self.assertEqual(stale["status"], "error", output(stale)) self.assertIn("changed since it was read", (stale.get("error") or "")) self.assertEqual(missing_new["status"], "error", output(missing_new)) self.assertIn("TypeError", (missing_new.get("error") or "")) self.assertIn("print(msg)", output(tail)) async def test_delete_refuses_a_directory_it_was_not_told_to_take(self): async with Peer() as peer: tree = Path(peer.directory.name) (tree / "build" / "deep").mkdir(parents=True) (tree / "build" / "one.o").write_text("x") (tree / "build" / "deep" / "two.o").write_text("x") (tree / "scratch.md").write_text("x") async def service(message): request = message["request"] if request["method"] == "runtime.inspect": return { "kind": "ok", "value": { "source_root": {"path": str(tree), "exists": True, "writable": False} }, } raise AssertionError(f"unexpected request: {request}") refused = await peer.execute("await delete('build')\n", service) root = await peer.execute("await delete('.')\n", service) # Captured before the recursive delete below, not after it. survived_refusals = (tree / "build" / "deep" / "two.o").exists() gone = await peer.execute("await delete('build', recursive=True)\nprint('gone')\n", service) file_gone = await peer.execute("await delete('scratch.md')\nprint(await (await open_file('scratch.md')).exists())\n", service) # `force` is narrow: it forgives a path that is already gone, and nothing else. gone_twice = await peer.execute("await delete('scratch.md')\n", service) forgiven = await peer.execute("await delete('scratch.md', force=True)\nprint('forgiven')\n", service) self.assertEqual(refused["status"], "error", output(refused)) # The refusal says how many entries are in the way, so the next call is informed, not hopeful. self.assertIn("is a directory with 2 entries", (refused.get("error") or "")) self.assertIn("pass recursive=True", (refused.get("error") or "")) self.assertTrue(survived_refusals) self.assertEqual(root["status"], "error", output(root)) self.assertIn("refuses the root of the tree", (root.get("error") or "")) self.assertEqual(gone["status"], "ok", output(gone)) self.assertFalse((tree / "build").exists()) self.assertIn("False", output(file_gone)) self.assertEqual(gone_twice["status"], "error", output(gone_twice)) self.assertEqual(forgiven["status"], "ok", output(forgiven)) async def test_write_takes_bytes_and_reads_back_exactly(self): async with Peer() as peer: tree = Path(peer.directory.name) raw = bytes(range(256)) async def service(message): request = message["request"] if request["method"] == "runtime.inspect": return { "kind": "ok", "value": { "source_root": {"path": str(tree), "exists": True, "writable": False} }, } raise AssertionError(f"unexpected request: {request}") written = await peer.execute( "payload = bytes(range(256))\n" "blob = await open_file('blob.bin')\n" "print('before', await blob.exists())\n" "await blob.write(payload)\n" "print('after', await blob.exists())\n" "print(len(await blob.read()), (await blob.read()) == payload)\n", service ) on_disk = (tree / "blob.bin").read_bytes() self.assertEqual(written["status"], "ok", output(written)) # Opening does not say whether the file is there; exists() does, before and after the write. self.assertIn("before False", output(written)) self.assertIn("after True", output(written)) self.assertIn("256 True", output(written)) self.assertEqual(on_disk, raw) async def test_an_edit_is_one_occurrence_or_an_error(self): async with Peer() as peer: tree = Path(peer.directory.name) (tree / "note.md").write_text("same\nsame\n") async def service(message): request = message["request"] if request["method"] == "runtime.inspect": return { "kind": "ok", "value": { "source_root": {"path": str(tree), "exists": True, "writable": False} }, } raise AssertionError(f"unexpected request: {request}") result = await peer.execute( "try:\n" " note = await open_file('note.md')\n" " await note.replace('same', 'other')\n" "except ValueError as error:\n" " print('refused:', error)\n", service, ) self.assertEqual(result["status"], "ok", output(result)) self.assertIn("refused:", output(result)) # Nothing was written: an ambiguous edit is refused, not applied somewhere. self.assertEqual((tree / "note.md").read_text(), "same\nsame\n") async def test_a_listing_says_what_things_are_and_what_it_skipped(self): async with Peer() as peer: tree = Path(peer.directory.name) (tree / "src").mkdir(exist_ok=True) (tree / "src" / "keep.py").write_text("needle\n") (tree / "target" / "build").mkdir(parents=True, exist_ok=True) (tree / "target" / "build" / "noise.py").write_text("needle\n") async def service(message): request = message["request"] if request["method"] == "runtime.inspect": return { "kind": "ok", "value": { "source_root": {"path": str(tree), "exists": True, "writable": False} }, } raise AssertionError(f"unexpected request: {request}") result = await peer.execute( "listing = await ls('src')\n" "print([(e.relative, e.is_dir, e.size is not None) for e in listing])\n" "print(await ls('src/keep.py'))\n" "hits = await search('needle', '.')\n" "print(sorted(m.relative for m in hits), hits.truncated, hits.pruned)\n" "wider = await search('needle', '.', include_pruned=True)\n" "print(sorted(m.relative for m in wider))\n" "capped = await search('needle', 'src', max_results=1)\n" "print(len(capped), capped.truncated)\n" "ones = await ls('.', recursive=True, match='*.py')\n" "print([e.relative for e in ones], ones.truncated)\n", service, ) self.assertEqual(result["status"], "ok", output(result)) printed = output(result) self.assertIn("[('src/keep.py', False, True)]", printed) # ls of a file names that file instead of raising at the caller. self.assertIn("src/keep.py", printed) # Build output is not walked, and the skip is reported rather than silent. self.assertIn("['src/keep.py'] False ('target',)", printed) self.assertIn("['src/keep.py', 'target/build/noise.py']", printed) # A capped search says it was capped. self.assertIn("1 True", printed) # A name glob walks the tree the caller meant, without reaching into build output. self.assertIn("['src/keep.py'] False", printed)