ive harnessed the harness
Something went wrong. Try again.
2.6 kB · 57 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758from pathlib import Pathimport unittestfrom peer import Peer
class WorkspaceTraceTests(unittest.IsolatedAsyncioTestCase): async def test_receipts_survive_failure_and_next_cell_only_contains_its_own_edit(self): async with Peer() as peer: root = Path(peer.directory.name) result = await peer.execute(f"""from pathlib import Pathfrom klbr.workspace import Workspacew = Workspace(Path({str(root)!r}))f = await w.open_file('index.html')await f.write('wrong title\\nbody\\n')await f.replace('body', 'retained body')raise ValueError('intentional failure after writing')""") self.assertEqual(result['status'], 'error') change, = result['trace']['changes'] self.assertEqual(change['path'], 'index.html') self.assertIn('+wrong title', change['diff']) self.assertIn('+retained body', change['diff']) self.assertEqual(change['added'], 2) fixed = await peer.execute("await f.replace('wrong title', 'right title')") change, = fixed['trace']['changes'] self.assertEqual((change['added'], change['removed']), (1, 1)) self.assertIn('-wrong title', change['diff']) self.assertIn('+right title', change['diff']) refused = await peer.execute("await f.replace('not present', 'oops')") self.assertEqual(refused['status'], 'error') self.assertEqual(refused['trace']['changes'], [])
async def test_multiple_files_bounds_and_no_phantom_changes(self): async with Peer() as peer: root = Path(peer.directory.name) result = await peer.execute(f"""from pathlib import Pathfrom klbr.workspace import Workspacew = Workspace(Path({str(root)!r}))a = await w.open_file('a.txt')b = await w.open_file('b.txt')await a.write('a\\n')await b.write(b'\\x00binary')await a.text()await w.ls()""") self.assertEqual(result['status'], 'ok') self.assertEqual([c['path'] for c in result['trace']['changes']], ['a.txt', 'b.txt']) self.assertTrue(result['trace']['changes'][1]['truncated']) self.assertEqual([a['kind'] for a in result['trace']['activities']], ['read', 'list']) unchanged = await peer.execute("await a.write('a\\n')") self.assertEqual(unchanged['trace']['changes'], []) huge = await peer.execute("await a.write('x' * 100000)") self.assertTrue(huge['trace']['changes'][0]['truncated']) syntax = await peer.execute("await a.write('unterminated") self.assertEqual(syntax['trace']['changes'], [])