mirror of
https://github.com/yusufipk/dikte.git
synced 2026-09-11 10:56:10 +00:00
yusufipk
034c2bec7e
Cleanup was the one step with only one place to run. Speech to text has three providers behind a setting and the agent has three behind another, but the model that drops the "eee"s out of a sentence was always a request to OpenRouter, which meant a second key on a machine that already pays for a model and already hands whole dictations to it as commands. Claude Code and Codex can rewrite a sentence as easily as they can put something in your calendar, and now they may. cleanup.py is where that choice lives, so worker, the file transcriber and the meeting all ask the same question rather than each building the same OpenRouter request. What comes out of a CLI that failed is a CleanupError, which is an ApiError, because to the chain a cleanup that failed is a cleanup that failed however it was run: the raw transcript is still pasted and the reason still shows in the corner, unchanged. Neither CLI is given anything it does not need for the job. No tools, no MCP servers, no session to resume, and the home directory rather than wherever the agent is pointed, since a project's instructions have opinions about how text should be written and none of them are about this transcript. The transcript goes in fenced the same way the OpenRouter call fences it, because it is material rather than an instruction however much of it reads like one. Claude takes the cleanup rules as its whole system prompt; Codex has no system prompt of its own, so they ride in front of the text, and its answer is read from the file it writes on the way out rather than from a stdout that also carries a header, its thinking and a token count. The cost is seconds. OpenRouter answers in about one, a CLI in six or seven, because each one opens a whole session to do it. That is the trade the box says out loud, and the default has not moved: OpenRouter cleans up until you say otherwise. Codex's two lowest thinking levels now ask for "low". "minimal" was its bottom rung until the newer models replaced it with "none", and each of them answers the other's word with a 400, which the agent has been quietly hitting too. In the settings window the model box belongs to whoever is chosen rather than meaning three different things in turn, since an OpenRouter id and a Claude alias do not belong in the same field, and under it is the same "found it or not" line the agent tab has. dikte doctor asks about the program instead of the key when a CLI does the cleaning, and the history records which model actually did it.
539 lines
22 KiB
Python
539 lines
22 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 time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import assistant
|
|
from tests.support import DikteTest, fake_urlopen, only_these_tools
|
|
|
|
|
|
class FakeCli:
|
|
"""A CLI that prints the given events and exits."""
|
|
|
|
def __init__(self, events=(), code=0, stderr="", noise=()):
|
|
lines = list(noise) + [json.dumps(event) for event in events]
|
|
self.stdout = io.StringIO("\n".join(lines) + "\n")
|
|
self.stderr = io.StringIO(stderr)
|
|
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 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("openrouter"), "")
|
|
|
|
def test_what_each_one_is_called(self):
|
|
self.assertEqual(assistant.display_name(self.config()), "Claude")
|
|
self.assertEqual(
|
|
assistant.display_name(self.config(assistant_provider="codex")), "Codex")
|
|
self.assertEqual(
|
|
assistant.display_name(self.config(assistant_provider="openrouter")),
|
|
"OpenRouter")
|
|
|
|
|
|
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))
|
|
|
|
def test_codex_has_no_rung_above_high(self):
|
|
self.assertEqual(assistant.CODEX_EFFORT["xhigh"], "high")
|
|
self.assertEqual(assistant.CODEX_EFFORT["max"], "high")
|
|
|
|
def test_neither_one_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.
|
|
for scale in (assistant.CLAUDE_EFFORT, assistant.CODEX_EFFORT):
|
|
self.assertEqual(scale["none"], "low")
|
|
self.assertEqual(scale["minimal"], "low")
|
|
|
|
def test_an_empty_setting_asks_for_nothing(self):
|
|
self.assertEqual(assistant.CLAUDE_EFFORT.get("", ""), "")
|
|
self.assertEqual(assistant.CODEX_EFFORT.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 SessionMissing(unittest.TestCase):
|
|
def test_a_session_that_is_gone(self):
|
|
for text in ("Error: session abc not found",
|
|
"No conversation with that id",
|
|
"unknown thread: abc"):
|
|
with self.subTest(text=text):
|
|
self.assertTrue(assistant._session_missing(text))
|
|
|
|
def test_an_unrelated_failure(self):
|
|
for text in ("", "network unreachable", "session limit exceeded"):
|
|
with self.subTest(text=text):
|
|
self.assertFalse(assistant._session_missing(text))
|
|
|
|
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_codex_stores_under_its_own_name(self):
|
|
assistant._conclude(self.found(answer="done", session="t-1"), 0, "",
|
|
"", "Codex")
|
|
self.assertEqual(assistant.read_session("codex", 1800), "t-1")
|
|
|
|
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_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_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 AskClaude(DikteTest):
|
|
def run_ask(self, conf=None, events=None, code=0, stderr="", 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, stderr=stderr, 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 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(), "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|