mirror of
https://github.com/yusufipk/dikte.git
synced 2026-09-11 10:56:10 +00:00
yusufipek
1df2d03056
Both sides rewrote the doctor: master rebuilt it around ready flags so a fully local setup stops reading as broken, while this branch taught it that Google has a key of its own and that the local model has neither a key nor a program. Kept master's structure and folded the branch in: the two hosted providers answer for their keys, a CLI for its program, and the JSON keeps both the provider-aware "key" and master's "ready". SECRET_KEYS gained groq on master and gemini here; the resolution keeps all four. _conclude was also changed by both: master stopped matching stderr wording and asks _API_TROUBLE instead, this branch renamed its last parameter to the provider's short name so a session is stored under the name it is read back by; the rename now rides on master's body. codex_models() and the settings loaders landed beside the new Gemini ones, so both stay. The new cleanup tests called the CLI fake by its old name, fake_run, which master had renamed to fake_cli.
851 lines
36 KiB
Python
851 lines
36 KiB
Python
"""Handing a dictation to an agent.
|
||
|
||
Three providers behind one setting, so most of this is about the command line
|
||
each of them is given and about the conversation carried between dictations. The
|
||
CLIs are faked at subprocess.Popen: what the tests read is the argument list and
|
||
what the stream of JSON events is turned into.
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
from dikte import assistant
|
||
from tests.support import (DikteTest, FakeCompleted, fake_urlopen,
|
||
only_these_tools)
|
||
|
||
|
||
class FakeCli:
|
||
"""A CLI that prints the given events and exits.
|
||
|
||
Its stderr is not modelled: _stream hands the process a temporary file for
|
||
that, and a mocked Popen leaves the file empty, which is what a quiet CLI
|
||
writes anyway.
|
||
"""
|
||
|
||
def __init__(self, events=(), code=0, noise=()):
|
||
lines = list(noise) + [json.dumps(event) for event in events]
|
||
self.stdout = io.StringIO("\n".join(lines) + "\n")
|
||
self.returncode = code
|
||
self.killed = False
|
||
|
||
def poll(self):
|
||
return self.returncode
|
||
|
||
def wait(self, timeout=None):
|
||
return self.returncode
|
||
|
||
def terminate(self):
|
||
self.killed = True
|
||
|
||
def kill(self):
|
||
self.killed = True
|
||
|
||
|
||
class WedgedCli:
|
||
"""A CLI whose stdout produces nothing, the way a hung process's does.
|
||
|
||
Iterating its stdout blocks until the kill arrives, because that is what
|
||
reading a silent pipe does: only the process ending closes the stream.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.pid = 4242
|
||
self.returncode = None
|
||
self.released = threading.Event()
|
||
self.stdout = self
|
||
|
||
def __iter__(self):
|
||
return self
|
||
|
||
def __next__(self):
|
||
# The 5 second cap is a safety net for the test itself; the kill is
|
||
# what is supposed to end the wait.
|
||
self.released.wait(timeout=5)
|
||
raise StopIteration
|
||
|
||
def poll(self):
|
||
return self.returncode
|
||
|
||
def wait(self, timeout=None):
|
||
return self.returncode
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
|
||
class Provider(DikteTest):
|
||
def test_the_default(self):
|
||
self.assertEqual(assistant.provider(self.config()), "claude")
|
||
|
||
def test_a_provider_this_version_does_not_have(self):
|
||
self.assertEqual(
|
||
assistant.provider(self.config(assistant_provider="ollama")), "claude")
|
||
|
||
def test_each_one_is_recognised(self):
|
||
for name in assistant.PROVIDERS:
|
||
with self.subTest(name=name):
|
||
self.assertEqual(
|
||
assistant.provider(self.config(assistant_provider=name)), name)
|
||
|
||
def test_what_each_one_runs(self):
|
||
self.assertEqual(assistant.executable("claude"), "claude")
|
||
self.assertEqual(assistant.executable("codex"), "codex")
|
||
self.assertEqual(assistant.executable("agy"), "agy")
|
||
self.assertEqual(assistant.executable("openrouter"), "")
|
||
|
||
def test_the_model_recorded_is_the_one_that_answered(self):
|
||
"""The history used to write Claude's setting whoever had answered."""
|
||
self.assertEqual(assistant.model(self.config()), "sonnet")
|
||
self.assertEqual(
|
||
assistant.model(self.config(assistant_provider="codex")), "codex")
|
||
self.assertEqual(
|
||
assistant.model(self.config(assistant_provider="agy")), "agy")
|
||
self.assertEqual(
|
||
assistant.model(self.config(assistant_provider="agy",
|
||
assistant_agy_model="gemini-3.1-pro-low")),
|
||
"gemini-3.1-pro-low")
|
||
self.assertEqual(
|
||
assistant.model(self.config(assistant_provider="openrouter")),
|
||
"google/gemini-3.5-flash")
|
||
|
||
def test_what_each_one_is_called(self):
|
||
self.assertEqual(assistant.display_name(self.config()), "Claude")
|
||
for name, called in (("codex", "Codex"), ("agy", "Antigravity"),
|
||
("openrouter", "OpenRouter")):
|
||
with self.subTest(name=name):
|
||
self.assertEqual(
|
||
assistant.display_name(self.config(assistant_provider=name)),
|
||
called)
|
||
|
||
def test_every_provider_has_a_name_to_be_called_by(self):
|
||
"""_conclude writes its errors in it, so a gap here is a bare id."""
|
||
self.assertEqual(set(assistant.SERVICES), set(assistant.PROVIDERS))
|
||
|
||
|
||
class Effort(unittest.TestCase):
|
||
"""One scale, offered once; a rung a provider lacks lands on its nearest."""
|
||
|
||
def test_the_scales_cover_the_same_settings(self):
|
||
self.assertEqual(set(assistant.CLAUDE_EFFORT), set(assistant.CODEX_EFFORT))
|
||
self.assertEqual(set(assistant.CLAUDE_EFFORT), set(assistant.AGY_EFFORT))
|
||
|
||
def test_neither_codex_nor_agy_has_a_rung_above_high(self):
|
||
for scale in (assistant.CODEX_EFFORT, assistant.AGY_EFFORT):
|
||
self.assertEqual(scale["xhigh"], "high")
|
||
self.assertEqual(scale["max"], "high")
|
||
|
||
def test_none_of_them_asks_for_a_rung_below_low(self):
|
||
# Claude has none; Codex has one, but calls it "minimal" on the older
|
||
# models and "none" on the newer ones, and refuses the wrong word; agy
|
||
# has three rungs and no word for off at all.
|
||
for scale in (assistant.CLAUDE_EFFORT, assistant.CODEX_EFFORT,
|
||
assistant.AGY_EFFORT):
|
||
self.assertEqual(scale["none"], "low")
|
||
self.assertEqual(scale["minimal"], "low")
|
||
|
||
def test_an_empty_setting_asks_for_nothing(self):
|
||
for scale in (assistant.CLAUDE_EFFORT, assistant.CODEX_EFFORT,
|
||
assistant.AGY_EFFORT):
|
||
self.assertEqual(scale.get("", ""), "")
|
||
|
||
|
||
class Session(DikteTest):
|
||
def test_nothing_stored_yet(self):
|
||
self.assertEqual(assistant.read_session("claude", 1800), "")
|
||
self.assertEqual(assistant.read_messages("openrouter", 1800), [])
|
||
self.assertEqual(assistant.stored_provider(), "")
|
||
self.assertIsNone(assistant.session_age())
|
||
|
||
def test_an_id_is_written_and_read_back(self):
|
||
assistant.write_session("claude", "abc-123")
|
||
self.assertEqual(assistant.read_session("claude", 1800), "abc-123")
|
||
self.assertEqual(assistant.stored_provider(), "claude")
|
||
|
||
def test_nobody_picks_up_another_provider_s_thread(self):
|
||
assistant.write_session("claude", "abc-123")
|
||
self.assertEqual(assistant.read_session("codex", 1800), "")
|
||
|
||
def test_a_conversation_that_has_sat_unused_is_dropped(self):
|
||
assistant.write_session("claude", "abc-123")
|
||
with mock.patch.object(time, "time", return_value=time.time() + 3600):
|
||
self.assertEqual(assistant.read_session("claude", 1800), "")
|
||
|
||
def test_a_session_that_never_expires(self):
|
||
assistant.write_session("claude", "abc-123")
|
||
with mock.patch.object(time, "time", return_value=time.time() + 10 ** 6):
|
||
self.assertEqual(assistant.read_session("claude", 0), "abc-123")
|
||
|
||
def test_the_messages_of_the_provider_that_keeps_none(self):
|
||
messages = [{"role": "user", "content": "hi"}]
|
||
assistant.write_session("openrouter", messages=messages)
|
||
self.assertEqual(assistant.read_messages("openrouter", 1800), messages)
|
||
|
||
def test_the_history_window_ends_somewhere(self):
|
||
messages = [{"role": "user", "content": str(index)} for index in range(50)]
|
||
assistant.write_session("openrouter", messages=messages)
|
||
stored = assistant.read_messages("openrouter", 1800)
|
||
self.assertEqual(len(stored), assistant.MAX_HISTORY)
|
||
self.assertEqual(stored[-1]["content"], "49")
|
||
|
||
def test_the_age_of_the_conversation(self):
|
||
assistant.write_session("claude", "abc-123")
|
||
self.assertLess(assistant.session_age(), 5)
|
||
|
||
def test_a_row_with_neither_an_id_nor_messages_has_no_age(self):
|
||
assistant.write_session("claude", "")
|
||
self.assertIsNone(assistant.session_age())
|
||
|
||
def test_clearing(self):
|
||
assistant.write_session("claude", "abc-123")
|
||
assistant.clear_session()
|
||
self.assertEqual(assistant.read_session("claude", 1800), "")
|
||
|
||
def test_clearing_one_that_is_not_there(self):
|
||
assistant.clear_session() # must not raise
|
||
|
||
def test_a_session_file_that_is_not_json(self):
|
||
assistant.SESSION_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
assistant.SESSION_FILE.write_text("{oh dear", encoding="utf-8")
|
||
self.assertEqual(assistant.read_session("claude", 1800), "")
|
||
self.assertEqual(assistant.stored_provider(), "")
|
||
self.assertIsNone(assistant.session_age())
|
||
|
||
|
||
class WorkingDir(DikteTest):
|
||
def test_the_home_directory_by_default(self):
|
||
self.assertEqual(assistant.working_dir(self.config()),
|
||
os.path.expanduser("~"))
|
||
|
||
def test_a_directory_of_your_own(self):
|
||
conf = self.config(assistant_dir=self.root)
|
||
self.assertEqual(assistant.working_dir(conf), self.root)
|
||
|
||
def test_a_tilde_is_expanded(self):
|
||
conf = self.config(assistant_dir="~")
|
||
self.assertEqual(assistant.working_dir(conf), os.path.expanduser("~"))
|
||
|
||
def test_a_directory_that_is_not_there_falls_back(self):
|
||
conf = self.config(assistant_dir="/no/such/place")
|
||
self.assertEqual(assistant.working_dir(conf), os.path.expanduser("~"))
|
||
|
||
|
||
class Labels(DikteTest):
|
||
def test_a_tool_the_table_knows(self):
|
||
self.assertEqual(assistant._claude_label({"name": "Bash"}),
|
||
"Running a command…")
|
||
|
||
def test_a_tool_arriving_from_an_mcp_server_is_named_by_its_server(self):
|
||
self.assertIn("gmail", assistant._claude_label({"name": "mcp__gmail__send"}))
|
||
|
||
def test_a_skill_is_named_by_the_skill(self):
|
||
label = assistant._claude_label(
|
||
{"name": "Skill", "input": {"skill": "calendar"}})
|
||
self.assertIn("calendar", label)
|
||
|
||
def test_a_tool_nobody_wrote_a_line_for(self):
|
||
self.assertIn("SomeNewTool",
|
||
assistant._claude_label({"name": "SomeNewTool"}))
|
||
|
||
def test_a_tool_with_no_name_at_all(self):
|
||
self.assertTrue(assistant._claude_label({}))
|
||
|
||
def test_the_codex_table(self):
|
||
self.assertEqual(assistant._codex_label({"type": "command_execution"}),
|
||
"Running a command…")
|
||
|
||
def test_a_codex_mcp_call(self):
|
||
self.assertIn("gmail", assistant._codex_label(
|
||
{"type": "mcp_tool_call", "server": "gmail"}))
|
||
|
||
def test_a_codex_item_nobody_listed(self):
|
||
self.assertIn("something_new",
|
||
assistant._codex_label({"type": "something_new"}))
|
||
|
||
|
||
class Denials(DikteTest):
|
||
def test_nothing_was_denied(self):
|
||
self.assertEqual(assistant._denial_warning({}), "")
|
||
self.assertEqual(assistant._denial_warning({"permission_denials": []}), "")
|
||
|
||
def test_a_denied_tool_is_named(self):
|
||
warning = assistant._denial_warning(
|
||
{"permission_denials": [{"tool_name": "Bash"}]})
|
||
self.assertIn("Bash", warning)
|
||
|
||
def test_the_same_tool_denied_twice_is_named_once(self):
|
||
warning = assistant._denial_warning({"permission_denials": [
|
||
{"tool_name": "Bash"}, {"tool_name": "Bash"}, {"tool_name": "Write"}]})
|
||
self.assertEqual(warning.count("Bash"), 1)
|
||
self.assertIn("Write", warning)
|
||
|
||
|
||
class LastLine(unittest.TestCase):
|
||
def test_the_last_line_is_the_one_worth_showing(self):
|
||
self.assertEqual(assistant.last_line("warning\n\nreal error\n"),
|
||
"real error")
|
||
self.assertEqual(assistant.last_line(""), "")
|
||
self.assertEqual(assistant.last_line(None), "")
|
||
|
||
|
||
class Conclude(DikteTest):
|
||
def found(self, **changes):
|
||
row = {"answer": "", "warning": "", "session": "", "failure": ""}
|
||
row.update(changes)
|
||
return row
|
||
|
||
def test_an_answer_and_its_session(self):
|
||
answer, warning = assistant._conclude(
|
||
self.found(answer="done", session="abc"), 0, "", "", "claude")
|
||
self.assertEqual(answer, "done")
|
||
self.assertEqual(warning, "")
|
||
self.assertEqual(assistant.read_session("claude", 1800), "abc")
|
||
|
||
def test_each_one_stores_under_its_own_name(self):
|
||
for name, session in (("codex", "t-1"), ("agy", "c-9")):
|
||
with self.subTest(name=name):
|
||
assistant._conclude(self.found(answer="done", session=session),
|
||
0, "", "", name)
|
||
self.assertEqual(assistant.read_session(name, 1800), session)
|
||
|
||
def test_the_error_is_written_in_the_provider_s_own_name(self):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
assistant._conclude(self.found(), 1, "", "", "agy")
|
||
self.assertIn("Antigravity", str(caught.exception))
|
||
|
||
def test_a_non_zero_exit_with_nothing_to_show_for_it(self):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
assistant._conclude(self.found(), 1, "it all went wrong\n", "", "claude")
|
||
self.assertIn("it all went wrong", str(caught.exception))
|
||
|
||
def test_a_session_that_is_gone_is_raised_apart(self):
|
||
with self.assertRaises(assistant._SessionGone):
|
||
assistant._conclude(self.found(), 1, "session abc not found",
|
||
"abc", "claude")
|
||
|
||
def test_the_recovery_no_longer_hangs_on_the_words_the_cli_chose(self):
|
||
# The complaint used to be matched by substring, which a CLI update or
|
||
# another language broke. A resumed run that died with nothing to show
|
||
# is now enough on its own.
|
||
for stderr in ("Oturum bulunamadı", "something else entirely", ""):
|
||
with self.subTest(stderr=stderr):
|
||
with self.assertRaises(assistant._SessionGone):
|
||
assistant._conclude(self.found(), 1, stderr, "abc", "Claude")
|
||
|
||
def test_api_trouble_on_a_resumed_run_is_not_blamed_on_the_session(self):
|
||
# A fresh session cannot cure a spent quota, a signed-out CLI or a dead
|
||
# network: the retry would fail the same way after a second wait, and
|
||
# the user would lose the conversation thread on top.
|
||
for stderr in ("Rate limit exceeded",
|
||
"You are not logged in. Please run /login.",
|
||
"API Error: 401 Unauthorized",
|
||
"fetch failed: ECONNREFUSED 127.0.0.1"):
|
||
with self.subTest(stderr=stderr):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
assistant._conclude(self.found(), 1, stderr, "abc", "Claude")
|
||
self.assertIn(stderr, str(caught.exception))
|
||
|
||
def test_a_session_that_is_gone_only_matters_when_one_was_resumed(self):
|
||
with self.assertRaises(assistant.AssistantError):
|
||
assistant._conclude(self.found(), 1, "session abc not found",
|
||
"", "claude")
|
||
|
||
def test_an_answer_survives_a_non_zero_exit(self):
|
||
answer, _ = assistant._conclude(self.found(answer="done"), 1, "noise",
|
||
"", "claude")
|
||
self.assertEqual(answer, "done")
|
||
|
||
def test_an_answer_on_a_resumed_session_is_kept_rather_than_retried(self):
|
||
answer, _ = assistant._conclude(self.found(answer="done"), 1, "noise",
|
||
"abc", "Claude")
|
||
self.assertEqual(answer, "done")
|
||
|
||
def test_a_reported_failure_with_no_answer(self):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
assistant._conclude(self.found(failure="the model refused"), 0, "",
|
||
"", "claude")
|
||
self.assertIn("refused", str(caught.exception))
|
||
|
||
def test_a_run_that_said_nothing_at_all(self):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
assistant._conclude(self.found(), 0, "", "", "codex")
|
||
self.assertIn("Codex", str(caught.exception))
|
||
|
||
|
||
class Stream(DikteTest):
|
||
def test_a_cli_that_floods_stderr_still_finishes(self):
|
||
# A real subprocess, because the wedge being tested is real plumbing:
|
||
# with stderr on a pipe nobody drains, 200 KB fills the pipe's buffer,
|
||
# the child blocks writing it, and the run hangs until the watchdog
|
||
# timeout. With stderr on a file the run completes at once.
|
||
script = (
|
||
"import sys\n"
|
||
"sys.stderr.write('x' * 200000)\n"
|
||
"sys.stderr.flush()\n"
|
||
"print('{\"type\": \"result\", \"result\": \"done\"}')\n"
|
||
)
|
||
conf = self.config(assistant_timeout=15)
|
||
events = []
|
||
code, stderr = assistant._stream(
|
||
[sys.executable, "-c", script], conf, events.append, None)
|
||
self.assertEqual(code, 0)
|
||
self.assertEqual(len(stderr), 200000)
|
||
self.assertEqual(events[-1]["result"], "done")
|
||
|
||
def test_the_watchdog_takes_the_whole_tree_down_on_timeout(self):
|
||
proc = WedgedCli()
|
||
|
||
def killed(target):
|
||
# What the real kill does, as far as _stream can see: the process
|
||
# ends, and its closing stream releases the blocked read.
|
||
target.returncode = 1
|
||
target.released.set()
|
||
|
||
conf = self.config(assistant_timeout=0)
|
||
with mock.patch.object(subprocess, "Popen", return_value=proc), \
|
||
mock.patch.object(assistant, "kill_tree",
|
||
side_effect=killed) as kill:
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
assistant._stream(["claude"], conf, lambda event: None, None)
|
||
kill.assert_called_once_with(proc)
|
||
self.assertIn("did not finish", str(caught.exception))
|
||
|
||
|
||
class AskClaude(DikteTest):
|
||
def run_ask(self, conf=None, events=None, code=0, noise=(),
|
||
session=""):
|
||
conf = conf or self.config()
|
||
proc = FakeCli(events or [
|
||
{"type": "system", "subtype": "init", "session_id": "abc"},
|
||
{"type": "result", "session_id": "abc", "result": " done "},
|
||
], code=code, noise=noise)
|
||
stages = []
|
||
with only_these_tools("claude", "codex"), \
|
||
mock.patch.object(subprocess, "Popen", return_value=proc) as popen:
|
||
result = assistant._ask_claude(
|
||
"book it", conf, session, stages.append, None)
|
||
return result, popen.call_args.args[0], stages
|
||
|
||
def test_the_answer_comes_back_stripped(self):
|
||
(answer, warning), _, _ = self.run_ask()
|
||
self.assertEqual(answer, "done")
|
||
self.assertEqual(warning, "")
|
||
|
||
def test_the_prompt_goes_in_as_one_argument(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertEqual(cmd[:3], ["claude", "-p", "book it"])
|
||
|
||
def test_the_stream_is_asked_for_so_progress_can_be_shown(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertIn("--output-format", cmd)
|
||
self.assertIn("stream-json", cmd)
|
||
self.assertIn("--verbose", cmd)
|
||
|
||
def test_the_model_and_the_permission_mode_are_passed_on(self):
|
||
conf = self.config(assistant_model="opus",
|
||
assistant_permission_mode="plan")
|
||
_, cmd, _ = self.run_ask(conf)
|
||
self.assertEqual(cmd[cmd.index("--model") + 1], "opus")
|
||
self.assertEqual(cmd[cmd.index("--permission-mode") + 1], "plan")
|
||
|
||
def test_the_instruction_rides_along_as_a_system_prompt(self):
|
||
conf = self.config()
|
||
_, cmd, _ = self.run_ask(conf)
|
||
self.assertEqual(cmd[cmd.index("--append-system-prompt") + 1],
|
||
conf.assistant_prompt())
|
||
|
||
def test_no_effort_asked_for_means_no_flag(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertNotIn("--effort", cmd)
|
||
|
||
def test_an_effort_is_translated_to_the_provider_s_vocabulary(self):
|
||
_, cmd, _ = self.run_ask(self.config(assistant_reasoning="minimal"))
|
||
self.assertEqual(cmd[cmd.index("--effort") + 1], "low")
|
||
|
||
def test_a_conversation_is_resumed(self):
|
||
_, cmd, _ = self.run_ask(session="abc-123")
|
||
self.assertEqual(cmd[cmd.index("--resume") + 1], "abc-123")
|
||
|
||
def test_a_fresh_conversation_resumes_nothing(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertNotIn("--resume", cmd)
|
||
|
||
def test_every_tool_it_picks_up_is_named_in_the_corner(self):
|
||
_, _, stages = self.run_ask(events=[
|
||
{"type": "assistant", "message": {"content": [
|
||
{"type": "tool_use", "name": "WebSearch"}]}},
|
||
{"type": "assistant", "message": {"content": [
|
||
{"type": "tool_use", "name": "Bash"}]}},
|
||
{"type": "result", "result": "done"},
|
||
])
|
||
self.assertEqual(stages, ["Searching the web…", "Running a command…"])
|
||
|
||
def test_a_denied_tool_comes_back_as_a_warning_beside_the_answer(self):
|
||
(answer, warning), _, _ = self.run_ask(events=[
|
||
{"type": "result", "result": "I could not do that.",
|
||
"permission_denials": [{"tool_name": "Bash"}]},
|
||
])
|
||
self.assertEqual(answer, "I could not do that.")
|
||
self.assertIn("Bash", warning)
|
||
|
||
def test_a_run_that_ended_in_an_error(self):
|
||
with self.assertRaises(assistant.AssistantError):
|
||
self.run_ask(events=[{"type": "result", "is_error": True,
|
||
"result": "rate limited"}])
|
||
|
||
def test_the_odd_unstructured_line_among_the_json(self):
|
||
(answer, _), _, _ = self.run_ask(noise=["Loading…", "not json at all"])
|
||
self.assertEqual(answer, "done")
|
||
|
||
def test_a_json_line_that_is_not_an_object(self):
|
||
proc = FakeCli(code=0)
|
||
proc.stdout = io.StringIO('{"type": "result", "result": "done"}\n[1,2]\n')
|
||
with only_these_tools("claude"), \
|
||
mock.patch.object(subprocess, "Popen", return_value=proc):
|
||
answer, _ = assistant._ask_claude("hi", self.config(), "", None, None)
|
||
self.assertEqual(answer, "done")
|
||
|
||
|
||
class AskCodex(DikteTest):
|
||
def run_ask(self, conf=None, events=None, session=""):
|
||
conf = conf or self.config(assistant_provider="codex")
|
||
proc = FakeCli(events or [
|
||
{"type": "thread.started", "thread_id": "t-1"},
|
||
{"type": "item.completed",
|
||
"item": {"type": "agent_message", "text": "done"}},
|
||
])
|
||
stages = []
|
||
with only_these_tools("codex"), \
|
||
mock.patch.object(subprocess, "Popen", return_value=proc) as popen:
|
||
result = assistant._ask_codex("book it", conf, session,
|
||
stages.append, None)
|
||
return result, popen.call_args.args[0], stages
|
||
|
||
def test_the_answer(self):
|
||
(answer, _), _, _ = self.run_ask()
|
||
self.assertEqual(answer, "done")
|
||
|
||
def test_the_instruction_is_kept_apart_from_the_command(self):
|
||
"""Codex takes no system prompt, so the two must not read as one."""
|
||
conf = self.config(assistant_provider="codex")
|
||
_, cmd, _ = self.run_ask(conf)
|
||
body = cmd[-1]
|
||
self.assertTrue(body.startswith(conf.assistant_prompt()))
|
||
self.assertIn("\n\n---\n\n", body)
|
||
self.assertTrue(body.endswith("book it"))
|
||
|
||
def test_there_is_nobody_here_to_approve_anything(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertIn('approval_policy="never"', cmd)
|
||
self.assertIn("--skip-git-repo-check", cmd)
|
||
self.assertIn("--json", cmd)
|
||
|
||
def test_the_sandbox_setting_is_passed_on(self):
|
||
_, cmd, _ = self.run_ask(
|
||
self.config(assistant_provider="codex",
|
||
assistant_codex_sandbox="read-only"))
|
||
self.assertIn('sandbox_mode="read-only"', cmd)
|
||
|
||
def test_no_model_named_means_whatever_codex_is_set_to(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertNotIn("-m", cmd)
|
||
|
||
def test_a_model_of_your_own(self):
|
||
_, cmd, _ = self.run_ask(
|
||
self.config(assistant_provider="codex", assistant_codex_model=" gpt-5 "))
|
||
self.assertEqual(cmd[cmd.index("-m") + 1], "gpt-5")
|
||
|
||
def test_the_effort_lands_on_the_nearest_rung_codex_has(self):
|
||
_, cmd, _ = self.run_ask(
|
||
self.config(assistant_provider="codex", assistant_reasoning="max"))
|
||
self.assertIn('model_reasoning_effort="high"', cmd)
|
||
|
||
def test_a_conversation_is_resumed(self):
|
||
_, cmd, _ = self.run_ask(session="t-1")
|
||
self.assertEqual(cmd[:4], ["codex", "exec", "resume", "t-1"])
|
||
|
||
def test_a_fresh_conversation(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertEqual(cmd[:2], ["codex", "exec"])
|
||
|
||
def test_the_closing_message_is_the_answer(self):
|
||
(answer, _), _, _ = self.run_ask(events=[
|
||
{"type": "item.completed",
|
||
"item": {"type": "agent_message", "text": "let me look"}},
|
||
{"type": "item.completed",
|
||
"item": {"type": "agent_message", "text": "it is on Thursday"}},
|
||
])
|
||
self.assertEqual(answer, "it is on Thursday")
|
||
|
||
def test_the_work_is_narrated_as_it_goes(self):
|
||
_, _, stages = self.run_ask(events=[
|
||
{"type": "item.started", "item": {"type": "command_execution"}},
|
||
{"type": "item.completed",
|
||
"item": {"type": "agent_message", "text": "done"}},
|
||
])
|
||
self.assertEqual(stages, ["Running a command…"])
|
||
|
||
def test_a_turn_that_failed(self):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
self.run_ask(events=[{"type": "turn.failed",
|
||
"error": {"message": "quota exhausted"}}])
|
||
self.assertIn("quota", str(caught.exception))
|
||
|
||
|
||
class AskAgy(DikteTest):
|
||
"""agy's stream is shaped nothing like the other two: the key is `event`,
|
||
the answer arrives whole in `result.response`, and the conversation to
|
||
resume is named in the first line rather than the last."""
|
||
|
||
def run_ask(self, conf=None, events=None, session=""):
|
||
conf = conf or self.config(assistant_provider="agy")
|
||
proc = FakeCli(events or [
|
||
{"event": "init", "conversation_id": "c-9", "init": {"cwd": "/home"}},
|
||
{"event": "result",
|
||
"result": {"conversation_id": "c-9", "status": "SUCCESS",
|
||
"response": " done "}},
|
||
])
|
||
stages = []
|
||
with only_these_tools("agy"), \
|
||
mock.patch.object(subprocess, "Popen", return_value=proc) as popen:
|
||
result = assistant._ask_agy("book it", conf, session,
|
||
stages.append, None)
|
||
return result, popen.call_args.args[0], stages
|
||
|
||
def test_the_answer_comes_back_stripped(self):
|
||
(answer, warning), _, _ = self.run_ask()
|
||
self.assertEqual(answer, "done")
|
||
self.assertEqual(warning, "")
|
||
|
||
def test_the_instruction_is_kept_apart_from_the_command(self):
|
||
"""agy takes no system prompt, so the two must not read as one."""
|
||
conf = self.config(assistant_provider="agy")
|
||
_, cmd, _ = self.run_ask(conf)
|
||
body = cmd[cmd.index("-p") + 1]
|
||
self.assertTrue(body.startswith(conf.assistant_prompt()))
|
||
self.assertIn("\n\n---\n\n", body)
|
||
self.assertTrue(body.endswith("book it"))
|
||
|
||
def test_a_first_command_starts_a_project_of_its_own(self):
|
||
"""Without it agy works in whichever project it was last in."""
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertIn("--new-project", cmd)
|
||
self.assertNotIn("--conversation", cmd)
|
||
|
||
def test_a_second_command_carries_the_conversation_rather_than_starting_one(self):
|
||
_, cmd, _ = self.run_ask(session="c-9")
|
||
self.assertEqual(cmd[cmd.index("--conversation") + 1], "c-9")
|
||
self.assertNotIn("--new-project", cmd)
|
||
|
||
def test_the_conversation_is_kept_under_agy_s_own_name(self):
|
||
self.run_ask()
|
||
self.assertEqual(assistant.read_session("agy", 1800), "c-9")
|
||
|
||
def test_it_is_not_left_to_give_up_before_the_caller_does(self):
|
||
conf = self.config(assistant_provider="agy", assistant_timeout=90)
|
||
_, cmd, _ = self.run_ask(conf)
|
||
self.assertEqual(cmd[cmd.index("--print-timeout") + 1], "90s")
|
||
|
||
def test_no_model_named_means_whatever_agy_is_set_to(self):
|
||
_, cmd, _ = self.run_ask()
|
||
self.assertNotIn("--model", cmd)
|
||
|
||
def test_a_model_of_your_own(self):
|
||
_, cmd, _ = self.run_ask(
|
||
self.config(assistant_provider="agy",
|
||
assistant_agy_model="gemini-3.1-pro-low"))
|
||
self.assertEqual(cmd[cmd.index("--model") + 1], "gemini-3.1-pro-low")
|
||
|
||
def test_a_tool_is_named_in_the_corner_as_it_starts(self):
|
||
_, _, stages = self.run_ask(events=[
|
||
{"event": "step_update",
|
||
"step_update": {"step_type": "tool", "state": "ACTIVE",
|
||
"tool_name": "run_command"}},
|
||
{"event": "step_update",
|
||
"step_update": {"step_type": "tool", "state": "DONE",
|
||
"tool_name": "run_command"}},
|
||
{"event": "result",
|
||
"result": {"status": "SUCCESS", "response": "done"}},
|
||
])
|
||
self.assertEqual(stages, ["Running a command…"])
|
||
|
||
def test_the_two_dozen_browser_tools_are_one_line_between_them(self):
|
||
_, _, stages = self.run_ask(events=[
|
||
{"event": "step_update",
|
||
"step_update": {"step_type": "tool", "state": "ACTIVE",
|
||
"tool_name": "browser_click_element"}},
|
||
{"event": "result",
|
||
"result": {"status": "SUCCESS", "response": "done"}},
|
||
])
|
||
self.assertEqual(stages, ["Working in the browser…"])
|
||
|
||
def test_a_turn_that_did_not_succeed_is_a_failure_rather_than_an_answer(self):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
self.run_ask(events=[
|
||
{"event": "result",
|
||
"result": {"status": "ERROR", "response": "the model refused"}},
|
||
])
|
||
self.assertIn("refused", str(caught.exception))
|
||
|
||
def test_a_failure_with_nothing_to_say_is_still_named(self):
|
||
with self.assertRaises(assistant.AssistantError) as caught:
|
||
self.run_ask(events=[
|
||
{"event": "result", "result": {"status": "ERROR"}},
|
||
])
|
||
self.assertIn("Antigravity", str(caught.exception))
|
||
|
||
|
||
class AskOpenRouter(DikteTest):
|
||
def test_a_question_and_an_answer(self):
|
||
conf = self.config(assistant_provider="openrouter",
|
||
openrouter_api_key="sk-or-test")
|
||
with fake_urlopen({"choices": [{"message": {"content": "on Thursday"}}]}):
|
||
answer, warning = assistant.ask("when is it", conf)
|
||
self.assertEqual(answer, "on Thursday")
|
||
self.assertEqual(warning, "")
|
||
|
||
def test_the_conversation_is_ours_to_keep(self):
|
||
conf = self.config(assistant_provider="openrouter",
|
||
openrouter_api_key="sk-or-test")
|
||
with fake_urlopen({"choices": [{"message": {"content": "on Thursday"}}]}):
|
||
assistant.ask("when is it", conf)
|
||
stored = assistant.read_messages("openrouter", 1800)
|
||
self.assertEqual([row["content"] for row in stored],
|
||
["when is it", "on Thursday"])
|
||
|
||
def test_the_next_command_knows_what_that_means(self):
|
||
conf = self.config(assistant_provider="openrouter",
|
||
openrouter_api_key="sk-or-test")
|
||
assistant.write_session("openrouter", messages=[
|
||
{"role": "user", "content": "when is it"},
|
||
{"role": "assistant", "content": "on Thursday"}])
|
||
with fake_urlopen({"choices": [{"message": {"content": "moved"}}]}) as calls:
|
||
assistant.ask("move it to Friday", conf)
|
||
sent = json.loads(calls[0].data.decode("utf-8"))["messages"]
|
||
self.assertEqual(len(sent), 4) # system, the two stored, the new one
|
||
|
||
def test_an_api_failure_reads_as_an_assistant_failure(self):
|
||
conf = self.config(assistant_provider="openrouter")
|
||
with self.assertRaises(assistant.AssistantError):
|
||
assistant.ask("when is it", conf)
|
||
|
||
|
||
class Ask(DikteTest):
|
||
def test_a_cli_that_is_not_installed_says_where_to_change_it(self):
|
||
with only_these_tools(), \
|
||
self.assertRaises(assistant.AssistantError) as caught:
|
||
assistant.ask("hi", self.config())
|
||
self.assertIn("claude", str(caught.exception))
|
||
self.assertIn("Settings", str(caught.exception))
|
||
|
||
def test_a_session_that_is_gone_is_started_over_without_a_word(self):
|
||
conf = self.config()
|
||
assistant.write_session("claude", "stale-id")
|
||
attempts = []
|
||
|
||
def run(prompt, conf, session, on_stage, should_stop):
|
||
attempts.append(session)
|
||
if session:
|
||
raise assistant._SessionGone()
|
||
return "done", ""
|
||
|
||
with only_these_tools("claude"), \
|
||
mock.patch.object(assistant, "_ask_claude", side_effect=run):
|
||
answer, _ = assistant.ask("hi", conf)
|
||
self.assertEqual(answer, "done")
|
||
self.assertEqual(attempts, ["stale-id", ""])
|
||
self.assertEqual(assistant.stored_provider(), "")
|
||
|
||
def test_a_resumed_run_that_dies_is_retried_without_the_session_flag(self):
|
||
# All the way through the stream this time: the first run exits 1 with
|
||
# no answer and whatever stderr it liked, and the recovery must not
|
||
# depend on those words.
|
||
conf = self.config()
|
||
assistant.write_session("claude", "stale-id")
|
||
procs = iter([
|
||
FakeCli(code=1),
|
||
FakeCli(events=[{"type": "result", "result": "done"}]),
|
||
])
|
||
cmds = []
|
||
|
||
def popen(cmd, **kwargs):
|
||
cmds.append(cmd)
|
||
return next(procs)
|
||
|
||
with only_these_tools("claude"), \
|
||
mock.patch.object(subprocess, "Popen", side_effect=popen):
|
||
answer, _ = assistant.ask("hi", conf)
|
||
self.assertEqual(answer, "done")
|
||
self.assertEqual(cmds[0][cmds[0].index("--resume") + 1], "stale-id")
|
||
self.assertNotIn("--resume", cmds[1])
|
||
|
||
def test_a_run_that_dies_with_an_answer_in_hand_is_not_retried(self):
|
||
conf = self.config()
|
||
assistant.write_session("claude", "stale-id")
|
||
calls = []
|
||
|
||
def popen(cmd, **kwargs):
|
||
calls.append(cmd)
|
||
return FakeCli(events=[{"type": "result", "result": "done"}], code=1)
|
||
|
||
with only_these_tools("claude"), \
|
||
mock.patch.object(subprocess, "Popen", side_effect=popen):
|
||
answer, _ = assistant.ask("hi", conf)
|
||
self.assertEqual(answer, "done")
|
||
self.assertEqual(len(calls), 1)
|
||
|
||
|
||
class CodexModels(DikteTest):
|
||
"""The model list read off `codex debug models`."""
|
||
|
||
CATALOG = {"models": [
|
||
{"slug": "gpt-6-mini", "visibility": "list", "priority": 9},
|
||
{"slug": "gpt-6", "visibility": "list", "priority": 1},
|
||
{"slug": "codex-auto-review", "visibility": "hide", "priority": 3},
|
||
]}
|
||
|
||
def models(self, reply, code=0):
|
||
with only_these_tools("codex"), \
|
||
mock.patch.object(subprocess, "run",
|
||
return_value=FakeCompleted(
|
||
returncode=code, stdout=reply)) as run:
|
||
found = assistant.codex_models()
|
||
self.run_call = run
|
||
return found
|
||
|
||
def test_the_catalog_arrives_best_first_without_the_hidden_ones(self):
|
||
found = self.models(json.dumps(self.CATALOG))
|
||
self.assertEqual(found, ["gpt-6", "gpt-6-mini"])
|
||
self.assertEqual(self.run_call.call_args.args[0],
|
||
["codex", "debug", "models"])
|
||
|
||
def test_a_codex_that_is_not_installed_is_not_run(self):
|
||
with only_these_tools(), \
|
||
mock.patch.object(subprocess, "run") as run:
|
||
self.assertEqual(assistant.codex_models(), [])
|
||
run.assert_not_called()
|
||
|
||
def test_a_codex_too_old_to_have_the_command(self):
|
||
self.assertEqual(self.models("error: unknown subcommand", code=2), [])
|
||
|
||
def test_a_catalog_that_is_not_what_was_expected(self):
|
||
self.assertEqual(self.models(json.dumps(["gpt-6"])), [])
|
||
self.assertEqual(self.models(""), [])
|
||
|
||
def test_a_codex_that_hangs_is_given_up_on(self):
|
||
with only_these_tools("codex"), \
|
||
mock.patch.object(subprocess, "run",
|
||
side_effect=subprocess.TimeoutExpired(
|
||
["codex"], 30)):
|
||
self.assertEqual(assistant.codex_models(), [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|