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)