Update documentation: Add file diff rendering to REPL, bump version to 1.17.0, enhance AGENTS.md, and add test for resume interrupt
This commit is contained in:
+193
@@ -299,6 +299,9 @@ class FakeSecretApp:
|
||||
self.approvals.append(command)
|
||||
return self.approve
|
||||
|
||||
def show_diff(self):
|
||||
pass
|
||||
|
||||
|
||||
class SecretsTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
@@ -2438,6 +2441,196 @@ class WebFetchTests(unittest.TestCase):
|
||||
self.assertIn("never curl or wget", tai.DEFAULT_SYSTEM)
|
||||
|
||||
|
||||
class DiffTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
os.environ["TAI_HOME"] = self.tmp.name
|
||||
|
||||
class FakeArgs:
|
||||
profile = "t"
|
||||
yes = True
|
||||
|
||||
self.config = tai.Config(FakeArgs())
|
||||
self.store = tai.Store(self.config, tai.Seal(self.config.home, "diff-test-1"))
|
||||
self.app = FakeSecretApp(self.store)
|
||||
self.app.profile = "t"
|
||||
self.tools = tai.Tools(self.app)
|
||||
|
||||
def tearDown(self):
|
||||
self.store.close()
|
||||
self.tmp.cleanup()
|
||||
os.environ.pop("TAI_HOME", None)
|
||||
|
||||
def test_highlight_python(self):
|
||||
out = tai.highlight_python("def f(x=1): # hi")
|
||||
self.assertIn(tai.Ansi.MAGENTA + "def" + tai.Ansi.RESET, out)
|
||||
self.assertIn(tai.Ansi.CYAN + "f" + tai.Ansi.RESET, out)
|
||||
self.assertIn(tai.FG_ORANGE + "1" + tai.Ansi.RESET, out)
|
||||
self.assertIn(tai.Ansi.GRAY + "# hi" + tai.Ansi.RESET, out)
|
||||
self.assertEqual(tai.strip_ansi(out), "def f(x=1): # hi")
|
||||
|
||||
def test_highlight_string_with_hash(self):
|
||||
out = tai.highlight_python("x = 'a # b'")
|
||||
self.assertIn(tai.Ansi.YELLOW + "'a # b'" + tai.Ansi.RESET, out)
|
||||
self.assertNotIn(tai.Ansi.GRAY, out)
|
||||
|
||||
def test_render_diff_plain(self):
|
||||
out = tai.render_diff("a.py", "x = 1\n", "x = 2\n", width=60, colors=False)
|
||||
self.assertEqual(out, "│ ── a.py (+1 -1)\n 1 - x = 1\n 1 + x = 2")
|
||||
|
||||
def test_render_diff_colors(self):
|
||||
out = tai.render_diff("a.py", "x = 1\n", "x = 2\n", width=60, colors=True)
|
||||
self.assertIn(tai.BG_ADD, out)
|
||||
self.assertIn(tai.BG_DEL, out)
|
||||
self.assertIn(tai.FG_ORANGE + "2" + tai.Ansi.RESET, out)
|
||||
for line in out.splitlines()[1:]:
|
||||
self.assertEqual(len(tai.strip_ansi(line)), 60)
|
||||
|
||||
def test_render_diff_gap_and_cap(self):
|
||||
old = "".join("line %d\n" % num for num in range(20))
|
||||
new = old.replace("line 0\n", "line zero\n").replace("line 19\n", "line nineteen\n")
|
||||
out = tai.render_diff("a.py", old, new, width=60, colors=False)
|
||||
self.assertIn("···", out)
|
||||
big = "".join("row %d\n" % num for num in range(200))
|
||||
capped = tai.render_diff("a.py", "", big, width=60, colors=False)
|
||||
self.assertIn("more lines hidden", capped)
|
||||
self.assertEqual(len(capped.splitlines()), 1 + tai.DIFF_MAX_LINES + 1)
|
||||
|
||||
def test_render_diff_shapes(self):
|
||||
self.assertEqual(tai.render_diff("x.py", "a\n", "a\n"), "")
|
||||
created = tai.render_diff("x.py", None, "a\nb\n", width=60, colors=False)
|
||||
self.assertIn("(new file, 2 lines)", created.splitlines()[0])
|
||||
removed = tai.render_diff("x.py", "a\nb\n", None, width=60, colors=False)
|
||||
self.assertIn("(deleted, 2 lines)", removed.splitlines()[0])
|
||||
plain = tai.render_diff("notes.txt", "a\n", "b\n", width=60, colors=True)
|
||||
self.assertIn(tai.BG_ADD, plain)
|
||||
self.assertNotIn(tai.Ansi.MAGENTA, plain)
|
||||
self.assertNotIn(tai.Ansi.YELLOW, plain)
|
||||
|
||||
def test_render_diff_docstring_fence(self):
|
||||
out = tai.render_diff("a.py", "", '"""\ndoc body\n"""\nx = 1\n', width=60, colors=True)
|
||||
self.assertIn(tai.Ansi.YELLOW + "doc body" + tai.Ansi.RESET, out)
|
||||
|
||||
def test_tools_note_preview(self):
|
||||
path = os.path.join(self.tmp.name, "note.py")
|
||||
self.assertTrue(self.tools.dispatch("write_file", json.dumps({"path": path, "content": "x = 1\n"})).startswith("wrote "))
|
||||
preview = self.tools.diff_preview
|
||||
self.assertEqual((preview["path"], preview["old"], preview["new"]), (path, None, "x = 1\n"))
|
||||
self.tools.dispatch("read_file", json.dumps({"path": path}))
|
||||
self.assertTrue(self.tools.dispatch("edit_file", json.dumps({"path": path, "find": "x = 1", "replace": "x = 2"})).startswith("edited "))
|
||||
preview = self.tools.diff_preview
|
||||
self.assertEqual((preview["old"], preview["new"]), ("x = 1\n", "x = 2\n"))
|
||||
self.assertTrue(self.tools.dispatch("delete_file", json.dumps({"path": path})).startswith("deleted "))
|
||||
preview = self.tools.diff_preview
|
||||
self.assertEqual((preview["old"], preview["new"]), ("x = 2\n", None))
|
||||
|
||||
def test_show_result_renders_and_clears(self):
|
||||
agent = tai.Agent(self.config, self.store, persist=False, quiet=False)
|
||||
agent.tools.diff_preview = {"path": "a.py", "old": "x = 1\n", "new": "x = 2\n"}
|
||||
with mock.patch("builtins.print") as shown:
|
||||
agent.show_result("ok", 1)
|
||||
agent.show_result("ok", 1)
|
||||
printed = "\n".join(str(call.args[0]) for call in shown.call_args_list)
|
||||
self.assertEqual(printed.count("── a.py"), 1)
|
||||
self.assertEqual(printed.count("└─"), 2)
|
||||
self.assertNotIn("\033", printed)
|
||||
self.assertIsNone(agent.tools.diff_preview)
|
||||
agent.quiet = True
|
||||
agent.tools.diff_preview = {"path": "a.py", "old": "x = 1\n", "new": "x = 2\n"}
|
||||
with mock.patch("builtins.print") as shown:
|
||||
agent.show_result("ok", 1)
|
||||
shown.assert_not_called()
|
||||
|
||||
|
||||
class ResumeTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
os.environ["TAI_HOME"] = self.tmp.name
|
||||
|
||||
class FakeArgs:
|
||||
profile = "t"
|
||||
yes = True
|
||||
|
||||
self.config = tai.Config(FakeArgs())
|
||||
self.store = tai.Store(self.config, tai.Seal(self.config.home, "resume-test-1"))
|
||||
|
||||
def tearDown(self):
|
||||
self.store.close()
|
||||
self.tmp.cleanup()
|
||||
os.environ.pop("TAI_HOME", None)
|
||||
|
||||
def text_reply(self, text="done"):
|
||||
return {"role": "assistant", "content": text, "reasoning": "", "tool_calls": [], "backend": "x"}
|
||||
|
||||
def test_atomic_session_survives_failed_replace(self):
|
||||
first = [{"role": "user", "content": "first"}]
|
||||
self.store.save_session("t", first)
|
||||
with mock.patch("os.replace", side_effect=OSError("disk full")):
|
||||
self.store.save_session("t", [{"role": "user", "content": "second"}])
|
||||
self.assertEqual([item["content"] for item in self.store.load_session("t")], ["first"])
|
||||
|
||||
def test_corrupt_session_quarantined(self):
|
||||
path = self.store.session_path("t")
|
||||
self.store.save_session("t", [{"role": "user", "content": "good"}])
|
||||
with open(path, "w", encoding="utf-8") as handle:
|
||||
handle.write("{not json")
|
||||
self.assertEqual(self.store.load_session("t"), [])
|
||||
leftovers = [entry for entry in os.listdir(self.config.profiles_dir) if ".corrupt-" in entry]
|
||||
self.assertEqual(len(leftovers), 1)
|
||||
with open(os.path.join(self.config.profiles_dir, leftovers[0]), encoding="utf-8") as handle:
|
||||
self.assertEqual(handle.read(), "{not json")
|
||||
|
||||
def test_normal_turn_closes_state(self):
|
||||
agent = tai.Agent(self.config, self.store, persist=True, quiet=True)
|
||||
with mock.patch.object(agent.chat, "complete", return_value=self.text_reply()):
|
||||
agent.run_turn("hello", capture=True)
|
||||
state = self.store.load_turn_state("t", "main")
|
||||
self.assertFalse(state["open"])
|
||||
self.assertEqual(state["goal"], "hello")
|
||||
|
||||
def test_crash_resume_end_to_end(self):
|
||||
agent = tai.Agent(self.config, self.store, persist=True, quiet=True)
|
||||
tool_reply = {"role": "assistant", "content": "", "reasoning": "", "tool_calls": [{"id": "1", "name": "shell", "arguments": json.dumps({"command": "echo resume-mark"})}], "backend": "x"}
|
||||
with mock.patch.object(agent.chat, "complete", side_effect=[dict(tool_reply), KeyboardInterrupt()]):
|
||||
with self.assertRaises(KeyboardInterrupt):
|
||||
agent.run_turn("do the thing", capture=True)
|
||||
state = self.store.load_turn_state("t", "main")
|
||||
self.assertTrue(state["open"])
|
||||
self.assertEqual(state["goal"], "do the thing")
|
||||
self.store.close()
|
||||
store2 = tai.Store(self.config, tai.Seal(self.config.home, "resume-test-1"))
|
||||
self.store = store2
|
||||
agent2 = tai.Agent(self.config, store2, persist=True, quiet=True)
|
||||
texts = [item.get("content") or "" for item in agent2.messages]
|
||||
self.assertTrue(any("do the thing" in text for text in texts))
|
||||
self.assertTrue(any("resume-mark" in text for text in texts))
|
||||
with mock.patch.object(agent2.chat, "complete", return_value=self.text_reply("finished")):
|
||||
with mock.patch("builtins.print"):
|
||||
answer = tai.resume_turn(agent2)
|
||||
self.assertEqual(answer, "finished")
|
||||
self.assertFalse(store2.load_turn_state("t", "main")["open"])
|
||||
|
||||
def test_resume_nothing_open(self):
|
||||
agent = tai.Agent(self.config, self.store, persist=True, quiet=True)
|
||||
with mock.patch.object(agent.chat, "complete") as called:
|
||||
with mock.patch("builtins.print") as shown:
|
||||
self.assertEqual(tai.resume_turn(agent), "")
|
||||
called.assert_not_called()
|
||||
printed = "\n".join(str(call.args[0]) for call in shown.call_args_list)
|
||||
self.assertIn("nothing to resume", printed)
|
||||
|
||||
def test_checkpoint_idle_writes_no_state(self):
|
||||
agent = tai.Agent(self.config, self.store, persist=True, quiet=True)
|
||||
agent.checkpoint()
|
||||
self.assertEqual(self.store.load_turn_state("t", "main"), {})
|
||||
self.assertFalse(os.path.exists(self.store.turn_state_path("t", "main")))
|
||||
|
||||
def test_parser_continue(self):
|
||||
args = tai.build_parser().parse_args(["--continue"])
|
||||
self.assertTrue(args.resume)
|
||||
self.assertFalse(tai.build_parser().parse_args([]).resume)
|
||||
|
||||
|
||||
class DenialTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
|
||||
Reference in New Issue
Block a user