Merge branch 'main' into bb/gui
This commit is contained in:
emozilla
@@ -0,0 +1,210 @@
|
||||
"""Skip the per-file shell linter when LSP will handle the same file.
|
||||
|
||||
The per-file ``npx tsc --noEmit FILE.ts`` shell linter cannot see
|
||||
``tsconfig.json`` (a documented ``tsc`` quirk: explicit file args bypass
|
||||
the project config), so it defaults to no-lib / ES5 and floods the
|
||||
agent's lint field with phantom "Cannot find 'Promise' / 'Map' / 'Set' /
|
||||
'ReadonlySet' / 'Iterable' / 'imul' / …" errors on every edit — up to
|
||||
25K tokens per patch. The LSP tier (``tsserver`` via
|
||||
typescript-language-server) reads tsconfig correctly and surfaces real
|
||||
diagnostics in the ``lsp_diagnostics`` field of the WriteResult /
|
||||
PatchResult.
|
||||
|
||||
These tests pin the contract:
|
||||
|
||||
- When LSP is active AND ``enabled_for(path)`` for a ``.ts`` / ``.go``
|
||||
/ ``.rs`` file, ``_check_lint`` returns ``skipped`` without invoking
|
||||
the shell linter at all.
|
||||
- When LSP is inactive or disabled-for-path, the shell linter runs
|
||||
exactly as before (regression guard for the default config).
|
||||
- The skip only applies to extensions in
|
||||
``_SHELL_LINTER_LSP_REDUNDANT`` — Python ``py_compile`` and
|
||||
``node --check`` keep running unconditionally because they're fast,
|
||||
file-local, and correct.
|
||||
- ``.tsx`` is intentionally NOT in either ``LINTERS`` or
|
||||
``_SHELL_LINTER_LSP_REDUNDANT``: it had no ``LINTERS`` entry
|
||||
pre-PR (so it was already implicitly ``skipped`` via the
|
||||
``ext not in LINTERS`` branch) and adding one would have inherited
|
||||
``.ts``'s broken ``tsc --noEmit FILE`` invocation for LSP-disabled
|
||||
users. When LSP IS enabled, ``.tsx`` is still covered by
|
||||
typescript-language-server via ``_maybe_lsp_diagnostics`` — the
|
||||
diagnostics show up on ``lsp_diagnostics``, not ``lint``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _make_fops():
|
||||
from tools.environments.local import LocalEnvironment
|
||||
from tools.file_operations import ShellFileOperations
|
||||
return ShellFileOperations(LocalEnvironment())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ext", [".ts", ".go", ".rs"])
|
||||
def test_shell_linter_skipped_when_lsp_will_handle(ext, tmp_path):
|
||||
"""When LSP is active and enabled_for(path), shell linter is skipped.
|
||||
|
||||
The shell linter's _exec must NOT be called — that's the whole
|
||||
point. We assert by patching ``_exec`` to raise, so any accidental
|
||||
invocation surfaces as a test failure.
|
||||
"""
|
||||
fops = _make_fops()
|
||||
src = tmp_path / f"bad{ext}"
|
||||
src.write_text("intentionally invalid content\n")
|
||||
|
||||
def _exec_must_not_run(*args, **kwargs): # pragma: no cover
|
||||
raise AssertionError(
|
||||
"shell linter was invoked despite LSP claiming the file"
|
||||
)
|
||||
|
||||
with patch.object(fops, "_lsp_will_handle", return_value=True), \
|
||||
patch.object(fops, "_exec", side_effect=_exec_must_not_run), \
|
||||
patch.object(fops, "_has_command", return_value=True):
|
||||
result = fops._check_lint(str(src))
|
||||
|
||||
assert result.skipped is True
|
||||
assert "LSP" in (result.message or "")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ext", [".ts", ".go", ".rs"])
|
||||
def test_shell_linter_runs_when_lsp_inactive(ext, tmp_path):
|
||||
"""When LSP is inactive (default config, no service, remote backend, ...),
|
||||
the shell linter runs as before — no behavior change."""
|
||||
fops = _make_fops()
|
||||
src = tmp_path / f"clean{ext}"
|
||||
src.write_text("// content\n")
|
||||
|
||||
fake_result = MagicMock()
|
||||
fake_result.exit_code = 0
|
||||
fake_result.stdout = ""
|
||||
|
||||
with patch.object(fops, "_lsp_will_handle", return_value=False), \
|
||||
patch.object(fops, "_exec", return_value=fake_result) as exec_mock, \
|
||||
patch.object(fops, "_has_command", return_value=True):
|
||||
result = fops._check_lint(str(src))
|
||||
|
||||
# _exec must have been called — proving the shell linter ran.
|
||||
assert exec_mock.called, "shell linter did NOT run when LSP was inactive"
|
||||
assert result.success is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ext", [".py", ".js"])
|
||||
def test_lsp_does_not_skip_non_redundant_extensions(ext, tmp_path):
|
||||
"""``py_compile`` and ``node --check`` keep running even when an LSP
|
||||
server (pyright/pylsp/typescript-language-server-for-JS) is active —
|
||||
they're fast, file-local, and correct, so there's no upside to
|
||||
suppressing them.
|
||||
"""
|
||||
fops = _make_fops()
|
||||
src = tmp_path / f"clean{ext}"
|
||||
src.write_text("# valid\n" if ext == ".py" else "// valid\n")
|
||||
|
||||
fake_result = MagicMock()
|
||||
fake_result.exit_code = 0
|
||||
fake_result.stdout = ""
|
||||
|
||||
# Even with LSP claiming the file, the shell linter must still run
|
||||
# for these extensions.
|
||||
with patch.object(fops, "_lsp_will_handle", return_value=True), \
|
||||
patch.object(fops, "_exec", return_value=fake_result) as exec_mock, \
|
||||
patch.object(fops, "_has_command", return_value=True):
|
||||
fops._check_lint(str(src))
|
||||
|
||||
assert exec_mock.called, (
|
||||
f"shell linter for {ext} did not run despite being in the "
|
||||
"'always-run' set (py_compile / node --check)"
|
||||
)
|
||||
|
||||
|
||||
def test_lsp_will_handle_returns_false_when_service_is_none(tmp_path):
|
||||
"""``_lsp_will_handle`` must return False when the LSP service hasn't
|
||||
been initialized — otherwise we'd accidentally skip the shell linter
|
||||
on systems where LSP isn't configured at all."""
|
||||
fops = _make_fops()
|
||||
src = tmp_path / "foo.ts"
|
||||
src.write_text("const x = 1\n")
|
||||
|
||||
with patch.object(fops, "_lsp_local_only", return_value=True), \
|
||||
patch("agent.lsp.get_service", return_value=None):
|
||||
assert fops._lsp_will_handle(str(src)) is False
|
||||
|
||||
|
||||
def test_lsp_will_handle_returns_false_on_remote_backend(tmp_path):
|
||||
"""LSP servers run on the host process — remote backends (Docker,
|
||||
SSH, Modal, …) keep files inside the sandbox where the host LSP
|
||||
can't reach them. ``_lsp_will_handle`` must short-circuit before
|
||||
calling into the service in that case."""
|
||||
fops = _make_fops()
|
||||
src = tmp_path / "foo.ts"
|
||||
src.write_text("const x = 1\n")
|
||||
|
||||
with patch.object(fops, "_lsp_local_only", return_value=False), \
|
||||
patch("agent.lsp.get_service") as get_service_mock:
|
||||
result = fops._lsp_will_handle(str(src))
|
||||
|
||||
assert result is False
|
||||
# Importantly: we never even consulted the service.
|
||||
assert not get_service_mock.called
|
||||
|
||||
|
||||
def test_lsp_will_handle_swallows_enabled_for_exception(tmp_path):
|
||||
"""A flaky LSP service must never break the shell-linter fallback —
|
||||
if ``enabled_for`` raises, we treat the file as "not handled" so the
|
||||
shell linter still runs."""
|
||||
fops = _make_fops()
|
||||
src = tmp_path / "foo.ts"
|
||||
src.write_text("const x = 1\n")
|
||||
|
||||
fake_svc = MagicMock()
|
||||
fake_svc.enabled_for.side_effect = RuntimeError("server crashed")
|
||||
|
||||
with patch.object(fops, "_lsp_local_only", return_value=True), \
|
||||
patch("agent.lsp.get_service", return_value=fake_svc):
|
||||
assert fops._lsp_will_handle(str(src)) is False
|
||||
|
||||
|
||||
def test_tsx_stays_out_of_linters_table_for_default_compatibility():
|
||||
"""Regression: keep ``.tsx`` out of ``LINTERS`` so users with LSP
|
||||
DISABLED don't suddenly get the broken ``npx tsc --noEmit FILE.tsx``
|
||||
invocation that ``.ts`` historically used to get.
|
||||
|
||||
Pre-PR behavior: ``.tsx`` had no entry in ``LINTERS``, so it fell
|
||||
through to ``ext not in LINTERS`` → ``LintResult(skipped=True,
|
||||
message="No linter for .tsx files")``. This PR preserves that for
|
||||
the default config.
|
||||
|
||||
When LSP IS enabled, ``.tsx`` is still covered by the LSP tier via
|
||||
``_maybe_lsp_diagnostics`` (typescript-language-server claims
|
||||
``.tsx`` in its extensions list) — the diagnostics show up in the
|
||||
``lsp_diagnostics`` field, not the ``lint`` field.
|
||||
"""
|
||||
from tools.file_operations import LINTERS, _SHELL_LINTER_LSP_REDUNDANT
|
||||
|
||||
assert ".tsx" not in LINTERS
|
||||
assert ".tsx" not in _SHELL_LINTER_LSP_REDUNDANT
|
||||
|
||||
|
||||
def test_tsx_default_check_lint_returns_skipped(tmp_path):
|
||||
"""End-to-end: ``.tsx`` files get ``LintResult(skipped=True)`` from
|
||||
``_check_lint`` regardless of LSP status — this is the no-regression
|
||||
contract that addresses Copilot review #3271017282."""
|
||||
fops = _make_fops()
|
||||
src = tmp_path / "foo.tsx"
|
||||
src.write_text("export const X = () => <div/>\n")
|
||||
|
||||
# Even with LSP claiming the file, no shell linter runs for .tsx
|
||||
# because there's no LINTERS entry — the ``ext not in LINTERS``
|
||||
# branch fires before the LSP short-circuit is consulted.
|
||||
with patch.object(fops, "_lsp_will_handle", return_value=True), \
|
||||
patch.object(fops, "_exec") as exec_mock:
|
||||
result = fops._check_lint(str(src))
|
||||
|
||||
assert result.skipped is True
|
||||
assert not exec_mock.called, "no shell linter should run for .tsx"
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -46,6 +46,26 @@ class TestChatCompletionsBasic:
|
||||
assert "codex_reasoning_items" in msgs[0]
|
||||
assert "codex_message_items" in msgs[0]
|
||||
|
||||
def test_convert_messages_strips_tool_name(self, transport):
|
||||
"""Internal `tool_name` (used for FTS indexing in the SQLite store) is
|
||||
not part of the OpenAI Chat Completions schema. Strict providers like
|
||||
Moonshot/Kimi reject it with HTTP 400 'Extra inputs are not permitted'.
|
||||
"""
|
||||
msgs = [
|
||||
{"role": "user", "content": "hi"},
|
||||
{"role": "assistant", "content": None,
|
||||
"tool_calls": [{"id": "call_1", "type": "function",
|
||||
"function": {"name": "execute_code", "arguments": "{}"}}]},
|
||||
{"role": "tool", "tool_call_id": "call_1", "tool_name": "execute_code",
|
||||
"content": "result"},
|
||||
]
|
||||
result = transport.convert_messages(msgs)
|
||||
assert "tool_name" not in result[2]
|
||||
assert result[2]["content"] == "result"
|
||||
assert result[2]["tool_call_id"] == "call_1"
|
||||
# Original list untouched (deepcopy-on-demand)
|
||||
assert msgs[2]["tool_name"] == "execute_code"
|
||||
|
||||
|
||||
class TestChatCompletionsBuildKwargs:
|
||||
|
||||
|
||||
@@ -160,30 +160,6 @@ class TestBranchCommandCLI:
|
||||
assert agent.reset_session_state.called
|
||||
assert agent._last_flushed_db_idx == 4 # len(conversation_history)
|
||||
|
||||
def test_branch_updates_agent_session_log_file(self, cli_instance, session_db, tmp_path):
|
||||
"""Branching must redirect the agent's session_log_file to the new session's path."""
|
||||
from cli import HermesCLI
|
||||
from pathlib import Path
|
||||
|
||||
logs_dir = tmp_path / "sessions"
|
||||
logs_dir.mkdir()
|
||||
|
||||
agent = MagicMock()
|
||||
agent._last_flushed_db_idx = 0
|
||||
agent.logs_dir = logs_dir
|
||||
agent.session_log_file = logs_dir / f"session_{cli_instance.session_id}.json"
|
||||
cli_instance.agent = agent
|
||||
|
||||
old_log_file = agent.session_log_file
|
||||
HermesCLI._handle_branch_command(cli_instance, "/branch")
|
||||
|
||||
new_session_id = cli_instance.session_id
|
||||
expected_log = logs_dir / f"session_{new_session_id}.json"
|
||||
assert agent.session_log_file == expected_log, (
|
||||
"session_log_file must point to the branch session, not the original"
|
||||
)
|
||||
assert agent.session_log_file != old_log_file
|
||||
|
||||
def test_branch_sets_resumed_flag(self, cli_instance, session_db):
|
||||
"""Branch should set _resumed=True to prevent auto-title generation."""
|
||||
from cli import HermesCLI
|
||||
|
||||
@@ -1,11 +1,18 @@
|
||||
"""Tests for CLI browser CDP auto-launch helpers."""
|
||||
|
||||
from contextlib import redirect_stdout
|
||||
from io import StringIO
|
||||
import os
|
||||
from queue import Queue
|
||||
import subprocess
|
||||
from unittest.mock import patch
|
||||
|
||||
from cli import HermesCLI
|
||||
from hermes_cli.browser_connect import manual_chrome_debug_command
|
||||
from hermes_cli.browser_connect import (
|
||||
get_chrome_debug_candidates,
|
||||
is_browser_debug_ready,
|
||||
manual_chrome_debug_command,
|
||||
)
|
||||
|
||||
|
||||
def _assert_chrome_debug_cmd(cmd, expected_chrome, expected_port):
|
||||
@@ -19,7 +26,35 @@ def _assert_chrome_debug_cmd(cmd, expected_chrome, expected_port):
|
||||
assert "chrome-debug" in user_data_args[0]
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
status = 200
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
|
||||
class TestChromeDebugLaunch:
|
||||
def test_browser_debug_ready_requires_http_cdp_endpoint(self):
|
||||
requested = []
|
||||
|
||||
def fake_urlopen(url, timeout):
|
||||
requested.append(url)
|
||||
if url.endswith("/json/version"):
|
||||
return _FakeResponse()
|
||||
raise OSError("unexpected probe")
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
assert is_browser_debug_ready("http://127.0.0.1:9222", timeout=0.1) is True
|
||||
|
||||
assert requested == ["http://127.0.0.1:9222/json/version"]
|
||||
|
||||
def test_browser_debug_ready_rejects_non_cdp_listener(self):
|
||||
with patch("urllib.request.urlopen", side_effect=OSError("not cdp")):
|
||||
assert is_browser_debug_ready("http://127.0.0.1:9222", timeout=0.1) is False
|
||||
|
||||
def test_windows_launch_uses_browser_found_on_path(self):
|
||||
captured = {}
|
||||
|
||||
@@ -72,6 +107,98 @@ class TestChromeDebugLaunch:
|
||||
assert command is not None
|
||||
assert command.startswith("/usr/bin/chromium --remote-debugging-port=9222")
|
||||
|
||||
def test_linux_candidates_prefer_chrome_before_brave_when_both_exist(self):
|
||||
chrome = "/usr/bin/google-chrome"
|
||||
brave = "/usr/bin/brave-browser"
|
||||
|
||||
def fake_which(name):
|
||||
return {"google-chrome": chrome, "brave-browser": brave}.get(name)
|
||||
|
||||
with patch("hermes_cli.browser_connect.shutil.which", side_effect=fake_which), \
|
||||
patch("hermes_cli.browser_connect.os.path.isfile", side_effect=lambda path: path in {chrome, brave}):
|
||||
candidates = get_chrome_debug_candidates("Linux")
|
||||
command = manual_chrome_debug_command(9222, "Linux")
|
||||
|
||||
assert candidates[:2] == [chrome, brave]
|
||||
assert command is not None
|
||||
assert command.startswith(f"{chrome} --remote-debugging-port=9222")
|
||||
|
||||
def test_linux_candidates_prefer_chrome_install_path_before_brave_on_path(self):
|
||||
chrome = "/opt/google/chrome/chrome"
|
||||
brave = "/usr/bin/brave-browser"
|
||||
|
||||
with patch("hermes_cli.browser_connect.shutil.which", side_effect=lambda name: brave if name == "brave-browser" else None), \
|
||||
patch("hermes_cli.browser_connect.os.path.isfile", side_effect=lambda path: path in {chrome, brave}):
|
||||
candidates = get_chrome_debug_candidates("Linux")
|
||||
|
||||
assert candidates[:2] == [chrome, brave]
|
||||
|
||||
def test_windows_candidates_prefer_chrome_install_path_before_brave_on_path(self, monkeypatch):
|
||||
program_files = r"C:\Program Files"
|
||||
chrome = os.path.join(program_files, "Google", "Chrome", "Application", "chrome.exe")
|
||||
brave = r"C:\Brave\brave.exe"
|
||||
|
||||
monkeypatch.setenv("ProgramFiles", program_files)
|
||||
monkeypatch.delenv("ProgramFiles(x86)", raising=False)
|
||||
monkeypatch.delenv("LOCALAPPDATA", raising=False)
|
||||
|
||||
with patch("hermes_cli.browser_connect.shutil.which", side_effect=lambda name: brave if name == "brave.exe" else None), \
|
||||
patch("hermes_cli.browser_connect.os.path.isfile", side_effect=lambda path: path in {chrome, brave}):
|
||||
candidates = get_chrome_debug_candidates("Windows")
|
||||
|
||||
assert candidates[:2] == [chrome, brave]
|
||||
|
||||
def test_linux_candidates_include_arch_brave_install_path(self):
|
||||
brave = "/opt/brave-bin/brave"
|
||||
|
||||
with patch("hermes_cli.browser_connect.shutil.which", return_value=None), \
|
||||
patch("hermes_cli.browser_connect.os.path.isfile", side_effect=lambda path: path == brave):
|
||||
candidates = get_chrome_debug_candidates("Linux")
|
||||
command = manual_chrome_debug_command(9222, "Linux")
|
||||
|
||||
assert candidates == [brave]
|
||||
assert command is not None
|
||||
assert command.startswith(f"{brave} --remote-debugging-port=9222")
|
||||
|
||||
def test_linux_candidates_include_brave_binary_name(self):
|
||||
brave = "/usr/bin/brave"
|
||||
|
||||
with patch("hermes_cli.browser_connect.shutil.which", side_effect=lambda name: brave if name == "brave" else None), \
|
||||
patch("hermes_cli.browser_connect.os.path.isfile", side_effect=lambda path: path == brave):
|
||||
candidates = get_chrome_debug_candidates("Linux")
|
||||
command = manual_chrome_debug_command(9222, "Linux")
|
||||
|
||||
assert candidates == [brave]
|
||||
assert command is not None
|
||||
assert command.startswith(f"{brave} --remote-debugging-port=9222")
|
||||
|
||||
def test_linux_candidates_include_official_brave_and_edge_stable_paths(self):
|
||||
brave = "/usr/bin/brave-browser-stable"
|
||||
edge = "/usr/bin/microsoft-edge-stable"
|
||||
|
||||
with patch("hermes_cli.browser_connect.shutil.which", return_value=None), \
|
||||
patch("hermes_cli.browser_connect.os.path.isfile", side_effect=lambda path: path in {brave, edge}):
|
||||
candidates = get_chrome_debug_candidates("Linux")
|
||||
|
||||
assert candidates == [brave, edge]
|
||||
|
||||
def test_launch_tries_next_browser_when_first_candidate_fails(self):
|
||||
brave = "/usr/bin/brave-browser"
|
||||
chrome = "/usr/bin/google-chrome"
|
||||
attempts = []
|
||||
|
||||
def fake_popen(cmd, **kwargs):
|
||||
attempts.append(cmd[0])
|
||||
if cmd[0] == brave:
|
||||
raise OSError("broken brave install")
|
||||
return object()
|
||||
|
||||
with patch("hermes_cli.browser_connect.get_chrome_debug_candidates", return_value=[brave, chrome]), \
|
||||
patch("subprocess.Popen", side_effect=fake_popen):
|
||||
assert HermesCLI._try_launch_chrome_debug(9222, "Linux") is True
|
||||
|
||||
assert attempts == [brave, chrome]
|
||||
|
||||
def test_manual_command_uses_wsl_windows_chrome_when_available(self):
|
||||
chrome = "/mnt/c/Program Files/Google/Chrome/Application/chrome.exe"
|
||||
|
||||
@@ -99,3 +226,28 @@ class TestChromeDebugLaunch:
|
||||
with patch("hermes_cli.browser_connect.shutil.which", return_value=None), \
|
||||
patch("hermes_cli.browser_connect.os.path.isfile", return_value=False):
|
||||
assert manual_chrome_debug_command(9222, "Linux") is None
|
||||
|
||||
def test_connect_context_note_allows_expected_browser_use(self, monkeypatch):
|
||||
"""`/browser connect` is an instruction to use the CDP browser.
|
||||
|
||||
The queued context note must not tell the model to wait for a second
|
||||
permission step or imply that the attached browser is the user's main
|
||||
everyday Chrome profile.
|
||||
"""
|
||||
cli = HermesCLI.__new__(HermesCLI)
|
||||
cli._pending_input = Queue()
|
||||
monkeypatch.delenv("BROWSER_CDP_URL", raising=False)
|
||||
|
||||
with patch("cli.is_browser_debug_ready", return_value=True), \
|
||||
patch("tools.browser_tool.cleanup_all_browsers"), \
|
||||
patch("tools.browser_tool._ensure_cdp_supervisor"), \
|
||||
redirect_stdout(StringIO()):
|
||||
cli._handle_browser_command("/browser connect")
|
||||
|
||||
note = cli._pending_input.get_nowait()
|
||||
assert "Chromium-family" in note
|
||||
assert "dev/debug" in note
|
||||
assert "using browser tools for their current browser-related request is expected" in note
|
||||
assert "live Chrome browser" not in note
|
||||
assert "real browser" not in note
|
||||
assert "Please await their instruction" not in note
|
||||
|
||||
@@ -74,7 +74,6 @@ class _Codex401ThenSuccessAgent(run_agent.AIAgent):
|
||||
self._cleanup_task_resources = lambda task_id: None
|
||||
self._persist_session = lambda messages, history=None: None
|
||||
self._save_trajectory = lambda messages, user_message, completed: None
|
||||
self._save_session_log = lambda messages: None
|
||||
|
||||
def _try_refresh_codex_client_credentials(self, *, force: bool = True) -> bool:
|
||||
type(self).refresh_attempts += 1
|
||||
|
||||
@@ -0,0 +1,223 @@
|
||||
"""Tests for ``hermes migrate xai`` — apply path with ruamel round-trip."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.xai_retirement import (
|
||||
RetirementIssue,
|
||||
apply_migration,
|
||||
find_retired_xai_refs,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
def trap_config(tmp_path: Path) -> Path:
|
||||
"""A config.yaml with retired models AND comments to verify round-trip."""
|
||||
p = tmp_path / "config.yaml"
|
||||
p.write_text(
|
||||
"# Hermes config (sample)\n"
|
||||
"principal:\n"
|
||||
" provider: xai # the main model\n"
|
||||
" model: grok-4-1-fast-non-reasoning # retiring May 15\n"
|
||||
" temperature: 0.5\n"
|
||||
"auxiliary:\n"
|
||||
" vision:\n"
|
||||
" provider: xai\n"
|
||||
" model: grok-4-fast-reasoning # retiring\n"
|
||||
" compression:\n"
|
||||
" provider: openai # not affected\n"
|
||||
" model: gpt-4o-mini\n"
|
||||
"delegation:\n"
|
||||
" model: grok-code-fast-1 # retiring\n"
|
||||
"plugins:\n"
|
||||
" image_gen:\n"
|
||||
" xai:\n"
|
||||
" model: grok-imagine-image-pro # retiring\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return p
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clean_config(tmp_path: Path) -> Path:
|
||||
p = tmp_path / "config.yaml"
|
||||
p.write_text(
|
||||
"principal:\n"
|
||||
" provider: xai\n"
|
||||
" model: grok-4.3\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return p
|
||||
|
||||
|
||||
def _parse(path: Path) -> dict:
|
||||
"""Load with ruamel for assertion convenience."""
|
||||
from ruamel.yaml import YAML
|
||||
yaml = YAML(typ="rt")
|
||||
with path.open("r", encoding="utf-8") as fh:
|
||||
return yaml.load(fh)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dry-run / no-op
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestNoOpPaths:
|
||||
def test_clean_config_returns_unchanged_result(self, clean_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(clean_config))
|
||||
assert issues == []
|
||||
result = apply_migration(clean_config, issues)
|
||||
assert result.config_changed is False
|
||||
assert result.backup_path is None
|
||||
# File untouched
|
||||
assert "grok-4.3" in clean_config.read_text(encoding="utf-8")
|
||||
|
||||
def test_empty_issues_list_is_noop(self, trap_config: Path):
|
||||
original = trap_config.read_text(encoding="utf-8")
|
||||
result = apply_migration(trap_config, issues=[])
|
||||
assert result.config_changed is False
|
||||
assert trap_config.read_text(encoding="utf-8") == original
|
||||
|
||||
def test_missing_file_raises(self, tmp_path: Path):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
apply_migration(tmp_path / "absent.yaml", issues=[
|
||||
RetirementIssue(
|
||||
config_path="principal.model",
|
||||
current_model="grok-3",
|
||||
replacement="grok-4.3",
|
||||
)
|
||||
])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Apply: surgical replacement
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyReplacement:
|
||||
def test_replaces_principal_model(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
result = apply_migration(trap_config, issues)
|
||||
assert result.config_changed is True
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["principal"]["model"] == "grok-4.3"
|
||||
|
||||
def test_adds_reasoning_effort_for_non_reasoning_variant(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
# Principal was grok-4-1-fast-non-reasoning → reasoning_effort: "none"
|
||||
assert cfg["principal"]["reasoning_effort"] == "none"
|
||||
|
||||
def test_replaces_auxiliary_vision(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["auxiliary"]["vision"]["model"] == "grok-4.3"
|
||||
|
||||
def test_replaces_delegation(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["delegation"]["model"] == "grok-4.3"
|
||||
|
||||
def test_replaces_image_gen_plugin(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
assert cfg["plugins"]["image_gen"]["xai"]["model"] == "grok-imagine-image-quality"
|
||||
|
||||
def test_does_not_touch_unrelated_slots(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
cfg = _parse(trap_config)
|
||||
# auxiliary.compression was never xAI, must remain untouched
|
||||
assert cfg["auxiliary"]["compression"]["model"] == "gpt-4o-mini"
|
||||
assert cfg["auxiliary"]["compression"]["provider"] == "openai"
|
||||
# principal.temperature must survive
|
||||
assert cfg["principal"]["temperature"] == 0.5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Round-trip preservation (the hard part)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRoundTripPreservation:
|
||||
def test_preserves_top_of_file_comment(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
text = trap_config.read_text(encoding="utf-8")
|
||||
assert "# Hermes config (sample)" in text
|
||||
|
||||
def test_preserves_inline_comments_on_unmodified_lines(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
text = trap_config.read_text(encoding="utf-8")
|
||||
assert "# the main model" in text
|
||||
assert "# not affected" in text
|
||||
|
||||
def test_preserves_top_level_key_order(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues)
|
||||
text = trap_config.read_text(encoding="utf-8")
|
||||
order = [
|
||||
text.index("principal:"),
|
||||
text.index("auxiliary:"),
|
||||
text.index("delegation:"),
|
||||
text.index("plugins:"),
|
||||
]
|
||||
assert order == sorted(order)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Backup behaviour
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestBackup:
|
||||
def test_backup_is_written_by_default(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
original = trap_config.read_text(encoding="utf-8")
|
||||
result = apply_migration(trap_config, issues)
|
||||
assert result.backup_path is not None
|
||||
assert result.backup_path.exists()
|
||||
assert result.backup_path.read_text(encoding="utf-8") == original
|
||||
|
||||
def test_backup_filename_prefixed(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
result = apply_migration(trap_config, issues)
|
||||
assert result.backup_path is not None
|
||||
assert result.backup_path.name.startswith("config.yaml.bak-pre-migrate-xai-")
|
||||
|
||||
def test_no_backup_when_disabled(self, trap_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(trap_config))
|
||||
result = apply_migration(trap_config, issues, backup=False)
|
||||
assert result.backup_path is None
|
||||
# No bak file in the directory
|
||||
assert not list(trap_config.parent.glob("*.bak-pre-migrate-xai-*"))
|
||||
|
||||
def test_no_backup_when_no_changes(self, clean_config: Path):
|
||||
issues = find_retired_xai_refs(_parse(clean_config))
|
||||
result = apply_migration(clean_config, issues, backup=True)
|
||||
assert result.backup_path is None # nothing to back up
|
||||
assert not list(clean_config.parent.glob("*.bak-pre-migrate-xai-*"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Idempotence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestIdempotence:
|
||||
def test_apply_twice_is_safe(self, trap_config: Path):
|
||||
# First pass: replace
|
||||
issues_1 = find_retired_xai_refs(_parse(trap_config))
|
||||
apply_migration(trap_config, issues_1)
|
||||
# Second pass: nothing to do
|
||||
issues_2 = find_retired_xai_refs(_parse(trap_config))
|
||||
assert issues_2 == []
|
||||
result_2 = apply_migration(trap_config, issues_2)
|
||||
assert result_2.config_changed is False
|
||||
@@ -1,6 +1,7 @@
|
||||
"""_tui_need_npm_install: auto npm when node_modules is behind the lockfile."""
|
||||
|
||||
import os
|
||||
import types
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -120,3 +121,75 @@ def test_no_install_prebuilt_bundle_mode(tmp_path: Path, main_mod) -> None:
|
||||
"""dist/entry.js present and no package-lock.json → prebuilt bundle, skip npm install."""
|
||||
_touch_tui_entry(tmp_path)
|
||||
assert main_mod._tui_need_npm_install(tmp_path) is False
|
||||
|
||||
|
||||
def test_need_rebuild_when_tui_bundle_missing(tmp_path: Path, main_mod) -> None:
|
||||
(tmp_path / "src").mkdir()
|
||||
(tmp_path / "src" / "entry.tsx").write_text("console.log('src')")
|
||||
|
||||
assert main_mod._tui_need_rebuild(tmp_path) is True
|
||||
|
||||
|
||||
def test_no_rebuild_when_tui_bundle_newer_than_inputs(tmp_path: Path, main_mod) -> None:
|
||||
_touch_tui_entry(tmp_path)
|
||||
src = tmp_path / "src"
|
||||
src.mkdir()
|
||||
(src / "entry.tsx").write_text("console.log('src')")
|
||||
os.utime(src / "entry.tsx", (100, 100))
|
||||
os.utime(tmp_path / "dist" / "entry.js", (200, 200))
|
||||
|
||||
assert main_mod._tui_need_rebuild(tmp_path) is False
|
||||
|
||||
|
||||
def test_rebuild_when_tui_source_newer_than_bundle(tmp_path: Path, main_mod) -> None:
|
||||
_touch_tui_entry(tmp_path)
|
||||
src = tmp_path / "src"
|
||||
src.mkdir()
|
||||
(src / "entry.tsx").write_text("console.log('src')")
|
||||
os.utime(tmp_path / "dist" / "entry.js", (100, 100))
|
||||
os.utime(src / "entry.tsx", (200, 200))
|
||||
|
||||
assert main_mod._tui_need_rebuild(tmp_path) is True
|
||||
|
||||
|
||||
def test_make_tui_argv_skips_build_only_on_termux_when_fresh(
|
||||
tmp_path: Path, main_mod, monkeypatch
|
||||
) -> None:
|
||||
_touch_tui_entry(tmp_path)
|
||||
monkeypatch.setenv("TERMUX_VERSION", "1")
|
||||
monkeypatch.setattr(main_mod, "_tui_need_npm_install", lambda _root: False)
|
||||
monkeypatch.setattr(main_mod, "_tui_need_rebuild", lambda _root: False)
|
||||
monkeypatch.setattr(main_mod.shutil, "which", lambda name: f"/bin/{name}")
|
||||
|
||||
def fail_run(*_args, **_kwargs):
|
||||
raise AssertionError("fresh Termux TUI launch must not rebuild")
|
||||
|
||||
monkeypatch.setattr(main_mod.subprocess, "run", fail_run)
|
||||
|
||||
argv, cwd = main_mod._make_tui_argv(tmp_path, tui_dev=False)
|
||||
|
||||
assert argv == ["/bin/node", str(tmp_path / "dist" / "entry.js")]
|
||||
assert cwd == tmp_path
|
||||
|
||||
|
||||
def test_make_tui_argv_keeps_desktop_always_build_behaviour(
|
||||
tmp_path: Path, main_mod, monkeypatch
|
||||
) -> None:
|
||||
_touch_tui_entry(tmp_path)
|
||||
monkeypatch.delenv("TERMUX_VERSION", raising=False)
|
||||
monkeypatch.setenv("PREFIX", "/usr")
|
||||
monkeypatch.setattr(main_mod, "_tui_need_npm_install", lambda _root: False)
|
||||
monkeypatch.setattr(main_mod, "_tui_need_rebuild", lambda _root: False)
|
||||
monkeypatch.setattr(main_mod.shutil, "which", lambda name: f"/bin/{name}")
|
||||
calls = []
|
||||
|
||||
def fake_run(*args, **kwargs):
|
||||
calls.append((args, kwargs))
|
||||
return types.SimpleNamespace(returncode=0, stdout="", stderr="")
|
||||
|
||||
monkeypatch.setattr(main_mod.subprocess, "run", fake_run)
|
||||
|
||||
main_mod._make_tui_argv(tmp_path, tui_dev=False)
|
||||
|
||||
assert calls
|
||||
assert calls[0][0][0] == ["/bin/npm", "run", "build"]
|
||||
|
||||
@@ -251,6 +251,38 @@ def test_main_top_level_tui_accepts_toolsets(monkeypatch, main_mod):
|
||||
assert captured == {"toolsets": "web,terminal", "tui": True}
|
||||
|
||||
|
||||
def test_termux_fast_tui_launch_uses_light_parser(monkeypatch, main_mod):
|
||||
captured = {}
|
||||
|
||||
monkeypatch.setenv("TERMUX_VERSION", "1")
|
||||
monkeypatch.setattr(
|
||||
sys, "argv", ["hermes", "--tui", "--toolsets", "web,terminal"]
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
main_mod,
|
||||
"cmd_chat",
|
||||
lambda args: captured.update({"toolsets": args.toolsets, "tui": args.tui}),
|
||||
)
|
||||
|
||||
assert main_mod._try_termux_fast_tui_launch() is True
|
||||
assert captured == {"toolsets": "web,terminal", "tui": True}
|
||||
|
||||
|
||||
def test_termux_fast_tui_launch_skips_help(monkeypatch, main_mod):
|
||||
monkeypatch.setenv("TERMUX_VERSION", "1")
|
||||
monkeypatch.setattr(sys, "argv", ["hermes", "--tui", "--help"])
|
||||
|
||||
assert main_mod._try_termux_fast_tui_launch() is False
|
||||
|
||||
|
||||
def test_fast_tui_launch_is_termux_only(monkeypatch, main_mod):
|
||||
monkeypatch.delenv("TERMUX_VERSION", raising=False)
|
||||
monkeypatch.setenv("PREFIX", "/usr")
|
||||
monkeypatch.setattr(sys, "argv", ["hermes", "--tui"])
|
||||
|
||||
assert main_mod._try_termux_fast_tui_launch() is False
|
||||
|
||||
|
||||
def test_main_top_level_oneshot_accepts_toolsets(monkeypatch, main_mod):
|
||||
captured = {}
|
||||
|
||||
|
||||
@@ -0,0 +1,275 @@
|
||||
"""Unit tests for hermes_cli.xai_retirement (May 15, 2026 model retirement)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.xai_retirement import (
|
||||
MIGRATION_GUIDE_URL,
|
||||
RETIREMENT_DATE,
|
||||
RetirementIssue,
|
||||
_RETIRED_MODELS,
|
||||
_looks_like_xai,
|
||||
_normalize,
|
||||
find_retired_xai_refs,
|
||||
format_issue,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _paths(issues):
|
||||
return [i.config_path for i in issues]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _normalize / _looks_like_xai
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestNormalize:
|
||||
def test_strips_x_ai_prefix(self):
|
||||
assert _normalize("x-ai/grok-4") == "grok-4"
|
||||
|
||||
def test_strips_xai_prefix(self):
|
||||
assert _normalize("xai/grok-4-fast") == "grok-4-fast"
|
||||
|
||||
def test_lowercases(self):
|
||||
assert _normalize("Grok-Code-Fast-1") == "grok-code-fast-1"
|
||||
|
||||
def test_no_prefix_passthrough(self):
|
||||
assert _normalize("grok-4.3") == "grok-4.3"
|
||||
|
||||
def test_strips_whitespace(self):
|
||||
assert _normalize(" grok-4 ") == "grok-4"
|
||||
|
||||
|
||||
class TestLooksLikeXai:
|
||||
def test_grok_prefix(self):
|
||||
assert _looks_like_xai("grok-4")
|
||||
assert _looks_like_xai("x-ai/grok-4-1-fast")
|
||||
|
||||
def test_non_grok_returns_false(self):
|
||||
assert not _looks_like_xai("gpt-4")
|
||||
assert not _looks_like_xai("claude-sonnet-4-6")
|
||||
assert not _looks_like_xai("openrouter/openai/gpt-4")
|
||||
|
||||
def test_none_or_empty(self):
|
||||
assert not _looks_like_xai(None)
|
||||
assert not _looks_like_xai("")
|
||||
assert not _looks_like_xai(" ")
|
||||
|
||||
def test_non_string(self):
|
||||
assert not _looks_like_xai(42)
|
||||
assert not _looks_like_xai({"model": "grok-4"})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# find_retired_xai_refs — config scanning
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestFindRetiredEdgeCases:
|
||||
def test_empty_config_no_issues(self):
|
||||
assert find_retired_xai_refs({}) == []
|
||||
|
||||
def test_non_dict_config_returns_empty(self):
|
||||
assert find_retired_xai_refs(None) == [] # type: ignore[arg-type]
|
||||
assert find_retired_xai_refs("nope") == [] # type: ignore[arg-type]
|
||||
|
||||
def test_no_xai_models_no_issues(self):
|
||||
cfg = {
|
||||
"principal": {"provider": "openai", "model": "gpt-4o"},
|
||||
"auxiliary": {"vision": {"model": "claude-sonnet-4-6"}},
|
||||
"delegation": {"model": "openai/o3"},
|
||||
}
|
||||
assert find_retired_xai_refs(cfg) == []
|
||||
|
||||
def test_xai_valid_model_not_flagged(self):
|
||||
cfg = {
|
||||
"principal": {"model": "grok-4.3"},
|
||||
"auxiliary": {
|
||||
"vision": {"model": "grok-4.20-0309-reasoning"},
|
||||
"fast": {"model": "grok-4-fast"},
|
||||
"fast_1": {"model": "grok-4-1-fast"},
|
||||
"bare": {"model": "grok-4"},
|
||||
},
|
||||
}
|
||||
assert find_retired_xai_refs(cfg) == []
|
||||
|
||||
|
||||
class TestFindRetiredPerSlot:
|
||||
def test_principal_retired(self):
|
||||
cfg = {"principal": {"model": "grok-code-fast-1"}}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert len(issues) == 1
|
||||
assert issues[0].config_path == "principal.model"
|
||||
assert issues[0].current_model == "grok-code-fast-1"
|
||||
assert issues[0].replacement == "grok-4.3"
|
||||
assert issues[0].reasoning_effort is None
|
||||
|
||||
def test_principal_with_x_ai_prefix(self):
|
||||
cfg = {"principal": {"model": "x-ai/grok-4-1-fast-non-reasoning"}}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert len(issues) == 1
|
||||
assert issues[0].current_model == "x-ai/grok-4-1-fast-non-reasoning"
|
||||
assert issues[0].replacement == "grok-4.3"
|
||||
assert issues[0].reasoning_effort == "none"
|
||||
|
||||
def test_auxiliary_multiple_slots(self):
|
||||
cfg = {
|
||||
"auxiliary": {
|
||||
"vision": {"model": "grok-4-fast-reasoning"},
|
||||
"compression": {"model": "grok-code-fast-1"},
|
||||
"curator": {"model": "grok-4.3"}, # not retired
|
||||
"approval": {"model": "gpt-4o-mini"}, # not xAI
|
||||
}
|
||||
}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert sorted(_paths(issues)) == [
|
||||
"auxiliary.compression.model",
|
||||
"auxiliary.vision.model",
|
||||
]
|
||||
|
||||
def test_auxiliary_unknown_slot_still_scanned(self):
|
||||
cfg = {"auxiliary": {"future_slot_xyz": {"model": "grok-3"}}}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert len(issues) == 1
|
||||
assert issues[0].config_path == "auxiliary.future_slot_xyz.model"
|
||||
|
||||
def test_delegation_retired(self):
|
||||
cfg = {"delegation": {"model": "grok-4-fast-reasoning"}}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert _paths(issues) == ["delegation.model"]
|
||||
|
||||
def test_tts_xai_retired(self):
|
||||
cfg = {"tts": {"xai": {"model": "grok-imagine-image-pro"}}}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert _paths(issues) == ["tts.xai.model"]
|
||||
assert issues[0].replacement == "grok-imagine-image-quality"
|
||||
|
||||
def test_image_gen_plugin_retired(self):
|
||||
cfg = {
|
||||
"plugins": {
|
||||
"image_gen": {
|
||||
"xai": {"model": "grok-imagine-image-pro"}
|
||||
}
|
||||
}
|
||||
}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert _paths(issues) == ["plugins.image_gen.xai.model"]
|
||||
assert issues[0].replacement == "grok-imagine-image-quality"
|
||||
|
||||
def test_full_trap_config(self):
|
||||
cfg = {
|
||||
"principal": {"model": "grok-4-1-fast-non-reasoning"},
|
||||
"auxiliary": {"vision": {"model": "grok-4-fast-reasoning"}},
|
||||
"delegation": {"model": "grok-code-fast-1"},
|
||||
"tts": {"xai": {"model": "grok-3"}}, # text model in TTS slot, but valid path
|
||||
"plugins": {"image_gen": {"xai": {"model": "grok-imagine-image-pro"}}},
|
||||
}
|
||||
issues = find_retired_xai_refs(cfg)
|
||||
assert len(issues) == 5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Migration semantics
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestMigrationSemantics:
|
||||
def test_non_reasoning_variant_recommends_reasoning_effort_none(self):
|
||||
cfg = {"principal": {"model": "grok-4-fast-non-reasoning"}}
|
||||
issue = find_retired_xai_refs(cfg)[0]
|
||||
assert issue.reasoning_effort == "none"
|
||||
|
||||
def test_reasoning_variant_no_extra_param(self):
|
||||
cfg = {"principal": {"model": "grok-4-1-fast-reasoning"}}
|
||||
issue = find_retired_xai_refs(cfg)[0]
|
||||
assert issue.reasoning_effort is None
|
||||
|
||||
def test_grok_3_maps_to_grok_4_3(self):
|
||||
cfg = {"principal": {"model": "grok-3"}}
|
||||
issue = find_retired_xai_refs(cfg)[0]
|
||||
assert issue.replacement == "grok-4.3"
|
||||
|
||||
def test_imagine_pro_maps_to_imagine_quality(self):
|
||||
cfg = {"plugins": {"image_gen": {"xai": {"model": "grok-imagine-image-pro"}}}}
|
||||
issue = find_retired_xai_refs(cfg)[0]
|
||||
assert issue.replacement == "grok-imagine-image-quality"
|
||||
|
||||
def test_all_retired_have_replacement(self):
|
||||
for name, entry in _RETIRED_MODELS.items():
|
||||
assert entry.get("replacement"), f"{name} has no replacement"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# format_issue
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestFormatIssue:
|
||||
def test_basic_format(self):
|
||||
issue = RetirementIssue(
|
||||
config_path="principal.model",
|
||||
current_model="grok-3",
|
||||
replacement="grok-4.3",
|
||||
)
|
||||
s = format_issue(issue)
|
||||
assert "principal.model" in s
|
||||
assert "'grok-3'" in s
|
||||
assert "'grok-4.3'" in s
|
||||
|
||||
def test_includes_reasoning_effort_when_set(self):
|
||||
issue = RetirementIssue(
|
||||
config_path="principal.model",
|
||||
current_model="grok-4-fast-non-reasoning",
|
||||
replacement="grok-4.3",
|
||||
reasoning_effort="none",
|
||||
)
|
||||
s = format_issue(issue)
|
||||
assert 'reasoning_effort: "none"' in s
|
||||
|
||||
def test_omits_reasoning_effort_when_none(self):
|
||||
issue = RetirementIssue(
|
||||
config_path="principal.model",
|
||||
current_model="grok-code-fast-1",
|
||||
replacement="grok-4.3",
|
||||
reasoning_effort=None,
|
||||
)
|
||||
s = format_issue(issue)
|
||||
assert "reasoning_effort" not in s
|
||||
|
||||
def test_includes_note_when_set(self):
|
||||
issue = RetirementIssue(
|
||||
config_path="principal.model",
|
||||
current_model="grok-3",
|
||||
replacement="grok-4.3",
|
||||
note="ambiguous variant",
|
||||
)
|
||||
s = format_issue(issue)
|
||||
assert "[note: ambiguous variant]" in s
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module-level constants sanity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestModuleConstants:
|
||||
def test_retirement_date_is_may_15(self):
|
||||
assert "May 15, 2026" == RETIREMENT_DATE
|
||||
|
||||
def test_migration_guide_url_points_to_xai(self):
|
||||
assert MIGRATION_GUIDE_URL.startswith("https://docs.x.ai/")
|
||||
assert "may-15" in MIGRATION_GUIDE_URL.lower()
|
||||
|
||||
def test_retired_models_keyset_matches_doc(self):
|
||||
# Snapshot test: if xAI's list changes we want CI to flag it.
|
||||
expected = {
|
||||
"grok-4-0709",
|
||||
"grok-4-fast-reasoning",
|
||||
"grok-4-fast-non-reasoning",
|
||||
"grok-4-1-fast-reasoning",
|
||||
"grok-4-1-fast-non-reasoning",
|
||||
"grok-code-fast-1",
|
||||
"grok-3",
|
||||
"grok-imagine-image-pro",
|
||||
}
|
||||
assert set(_RETIRED_MODELS.keys()) == expected
|
||||
@@ -110,8 +110,6 @@ class TestFlushDeduplication:
|
||||
db = SessionDB(db_path=db_path)
|
||||
|
||||
agent = self._make_agent(db)
|
||||
# Stub out _save_session_log to avoid file I/O
|
||||
agent._save_session_log = MagicMock()
|
||||
|
||||
conversation_history = [{"role": "user", "content": "old"}]
|
||||
messages = list(conversation_history) + [
|
||||
|
||||
@@ -52,7 +52,7 @@ def _make_agent(monkeypatch, api_mode, provider, response_fn):
|
||||
kw.update(skip_context_files=True, skip_memory=True, max_iterations=4)
|
||||
super().__init__(*a, **kw)
|
||||
self._cleanup_task_resources = self._persist_session = lambda *a, **k: None
|
||||
self._save_trajectory = self._save_session_log = lambda *a, **k: None
|
||||
self._save_trajectory = lambda *a, **k: None
|
||||
|
||||
def run_conversation(self, msg, conversation_history=None, task_id=None):
|
||||
self._interruptible_api_call = lambda kw: response_fn()
|
||||
|
||||
@@ -9,11 +9,7 @@ def _agent_with_stubbed_persistence():
|
||||
agent._persist_user_message_override = None
|
||||
agent._session_db = None
|
||||
agent._session_messages = []
|
||||
agent.saved_session_logs = []
|
||||
agent.flushed_session_db_messages = []
|
||||
agent._save_session_log = lambda messages: agent.saved_session_logs.append(
|
||||
[m.copy() for m in messages]
|
||||
)
|
||||
agent._flush_messages_to_session_db = lambda messages, conversation_history=None: (
|
||||
agent.flushed_session_db_messages.append([m.copy() for m in messages])
|
||||
)
|
||||
@@ -60,7 +56,7 @@ def test_persist_session_strips_trailing_empty_recovery_scaffolding():
|
||||
assert messages == [
|
||||
{"role": "user", "content": "run the task"},
|
||||
]
|
||||
assert agent.saved_session_logs[-1] == messages
|
||||
assert agent.flushed_session_db_messages[-1] == messages
|
||||
assert all(not msg.get("_empty_recovery_synthetic") for msg in messages)
|
||||
|
||||
|
||||
@@ -77,7 +73,7 @@ def test_persist_session_keeps_unmarked_terminal_empty_response():
|
||||
{"role": "user", "content": "run the task"},
|
||||
{"role": "assistant", "content": "(empty)"},
|
||||
]
|
||||
assert agent.saved_session_logs[-1] == messages
|
||||
assert agent.flushed_session_db_messages[-1] == messages
|
||||
|
||||
|
||||
def test_persist_session_strips_marked_terminal_empty_sentinel():
|
||||
@@ -94,5 +90,5 @@ def test_persist_session_strips_marked_terminal_empty_sentinel():
|
||||
AIAgent._persist_session(agent, messages, conversation_history=[])
|
||||
|
||||
assert messages == [{"role": "user", "content": "continue"}]
|
||||
assert agent.saved_session_logs[-1] == messages
|
||||
assert agent.flushed_session_db_messages[-1] == messages
|
||||
assert all(not msg.get("_empty_terminal_sentinel") for msg in messages)
|
||||
|
||||
@@ -554,23 +554,50 @@ class TestExtractReasoning:
|
||||
assert result == "from structured field"
|
||||
|
||||
|
||||
class TestCleanSessionContent:
|
||||
def test_none_passthrough(self):
|
||||
assert AIAgent._clean_session_content(None) is None
|
||||
class TestSessionJsonSnapshotOptIn:
|
||||
"""Regression: per-session JSON snapshot writer is opt-in via config.
|
||||
|
||||
def test_scratchpad_converted(self):
|
||||
text = "<REASONING_SCRATCHPAD>think</REASONING_SCRATCHPAD> answer"
|
||||
result = AIAgent._clean_session_content(text)
|
||||
assert "<REASONING_SCRATCHPAD>" not in result
|
||||
assert "<think>" in result
|
||||
state.db is canonical (PR #29182). ``sessions.write_json_snapshots``
|
||||
defaults to False, so the agent must NOT write ``session_{sid}.json``
|
||||
files by default — that behavior caused multi-GB sessions directories
|
||||
on heavy users. Users can opt back in for external tooling that reads
|
||||
the JSON files directly.
|
||||
"""
|
||||
|
||||
def test_extra_newlines_cleaned(self):
|
||||
text = "\n\n\n<think>x</think>\n\n\nafter"
|
||||
result = AIAgent._clean_session_content(text)
|
||||
# Should not have excessive newlines around think block
|
||||
assert "\n\n\n" not in result
|
||||
# Content after think block must be preserved
|
||||
assert "after" in result
|
||||
def test_session_json_disabled_by_default(self, agent):
|
||||
# Default config: writer is gated off.
|
||||
assert getattr(agent, "_session_json_enabled", False) is False, (
|
||||
"sessions.write_json_snapshots must default to False"
|
||||
)
|
||||
|
||||
def test_save_session_log_noops_when_disabled(self, agent, tmp_path):
|
||||
# When disabled, calling the method must not write any file even
|
||||
# if logs_dir is writable and messages are non-empty.
|
||||
agent._session_json_enabled = False
|
||||
agent.logs_dir = tmp_path
|
||||
agent._session_messages = [{"role": "user", "content": "hello"}]
|
||||
agent._save_session_log()
|
||||
# No session_*.json must appear under logs_dir.
|
||||
assert list(tmp_path.glob("session_*.json")) == []
|
||||
|
||||
def test_save_session_log_writes_when_enabled(self, agent, tmp_path):
|
||||
# Opt-in path: with the flag on and a session_id, the writer must
|
||||
# produce ``session_{sid}.json`` under logs_dir.
|
||||
agent._session_json_enabled = True
|
||||
agent.logs_dir = tmp_path
|
||||
messages = [{"role": "user", "content": "hello"}]
|
||||
agent._save_session_log(messages)
|
||||
expected = tmp_path / f"session_{agent.session_id}.json"
|
||||
assert expected.exists(), (
|
||||
"Opt-in writer must produce session_{sid}.json under logs_dir"
|
||||
)
|
||||
|
||||
def test_logs_dir_retained_for_request_dumps(self, agent):
|
||||
# logs_dir is kept unconditionally because
|
||||
# agent_runtime_helpers.dump_api_request_debug still writes
|
||||
# request_dump_*.json there (debug breadcrumb path), independent of
|
||||
# the session JSON opt-in.
|
||||
assert hasattr(agent, "logs_dir")
|
||||
|
||||
|
||||
class TestGetMessagesUpToLastAssistant:
|
||||
@@ -1901,7 +1928,6 @@ class TestExecuteToolCalls:
|
||||
agent._interruptible_api_call = _fake_api_call
|
||||
agent._persist_session = lambda *args, **kwargs: None
|
||||
agent._save_trajectory = lambda *args, **kwargs: None
|
||||
agent._save_session_log = lambda *args, **kwargs: None
|
||||
|
||||
captured = io.StringIO()
|
||||
agent._print_fn = lambda *args, **kw: print(*args, file=captured, **kw)
|
||||
@@ -4253,22 +4279,6 @@ class TestSafeWriter:
|
||||
assert inner.getvalue() == "test"
|
||||
|
||||
|
||||
class TestSaveSessionLogAtomicWrite:
|
||||
def test_uses_shared_atomic_json_helper(self, agent, tmp_path):
|
||||
agent.session_log_file = tmp_path / "session.json"
|
||||
messages = [{"role": "user", "content": "hello"}]
|
||||
|
||||
with patch("run_agent.atomic_json_write", create=True) as mock_atomic_write:
|
||||
agent._save_session_log(messages)
|
||||
|
||||
mock_atomic_write.assert_called_once()
|
||||
call_args = mock_atomic_write.call_args
|
||||
assert call_args.args[0] == agent.session_log_file
|
||||
payload = call_args.args[1]
|
||||
assert payload["session_id"] == agent.session_id
|
||||
assert payload["messages"] == messages
|
||||
assert call_args.kwargs["indent"] == 2
|
||||
assert call_args.kwargs["default"] is str
|
||||
|
||||
|
||||
# ===================================================================
|
||||
@@ -5056,12 +5066,9 @@ class TestPersistUserMessageOverride:
|
||||
{"role": "assistant", "content": "Hi!"},
|
||||
]
|
||||
|
||||
with patch.object(agent, "_save_session_log") as mock_save:
|
||||
agent._persist_session(messages, [])
|
||||
agent._persist_session(messages, [])
|
||||
|
||||
assert messages[0]["content"] == "Hello there"
|
||||
saved_messages = mock_save.call_args.args[0]
|
||||
assert saved_messages[0]["content"] == "Hello there"
|
||||
first_db_write = agent._session_db.append_message.call_args_list[0].kwargs
|
||||
assert first_db_write["content"] == "Hello there"
|
||||
|
||||
|
||||
@@ -54,7 +54,6 @@ def _build_agent(monkeypatch):
|
||||
agent._cleanup_task_resources = lambda task_id: None
|
||||
agent._persist_session = lambda messages, history=None: None
|
||||
agent._save_trajectory = lambda messages, user_message, completed: None
|
||||
agent._save_session_log = lambda messages: None
|
||||
return agent
|
||||
|
||||
|
||||
@@ -75,7 +74,6 @@ def _build_copilot_agent(monkeypatch, *, model="gpt-5.4"):
|
||||
agent._cleanup_task_resources = lambda task_id: None
|
||||
agent._persist_session = lambda messages, history=None: None
|
||||
agent._save_trajectory = lambda messages, user_message, completed: None
|
||||
agent._save_session_log = lambda messages: None
|
||||
return agent
|
||||
|
||||
|
||||
@@ -335,7 +333,6 @@ def test_build_api_kwargs_codex_clamps_minimal_effort(monkeypatch):
|
||||
agent._cleanup_task_resources = lambda task_id: None
|
||||
agent._persist_session = lambda messages, history=None: None
|
||||
agent._save_trajectory = lambda messages, user_message, completed: None
|
||||
agent._save_session_log = lambda messages: None
|
||||
|
||||
kwargs = agent._build_api_kwargs(
|
||||
[
|
||||
@@ -365,7 +362,6 @@ def test_build_api_kwargs_codex_preserves_supported_efforts(monkeypatch):
|
||||
agent._cleanup_task_resources = lambda task_id: None
|
||||
agent._persist_session = lambda messages, history=None: None
|
||||
agent._save_trajectory = lambda messages, user_message, completed: None
|
||||
agent._save_session_log = lambda messages: None
|
||||
|
||||
kwargs = agent._build_api_kwargs(
|
||||
[
|
||||
@@ -594,7 +590,6 @@ def _build_xai_oauth_agent(monkeypatch):
|
||||
agent._cleanup_task_resources = lambda task_id: None
|
||||
agent._persist_session = lambda messages, history=None: None
|
||||
agent._save_trajectory = lambda messages, user_message, completed: None
|
||||
agent._save_session_log = lambda messages: None
|
||||
return agent
|
||||
|
||||
|
||||
|
||||
@@ -3995,7 +3995,7 @@ def test_browser_manage_connect_sets_env_and_cleans_twice(monkeypatch):
|
||||
|
||||
assert resp["result"]["connected"] is True
|
||||
assert resp["result"]["url"] == "http://127.0.0.1:9222"
|
||||
assert resp["result"]["messages"] == ["Chrome is already listening on port 9222"]
|
||||
assert resp["result"]["messages"] == ["Chromium-family browser is already listening on port 9222"]
|
||||
assert os.environ.get("BROWSER_CDP_URL") == "http://127.0.0.1:9222"
|
||||
# First cleanup runs against the OLD env (none here), second against the NEW.
|
||||
assert cleanup_calls == ["", "http://127.0.0.1:9222"]
|
||||
@@ -4015,7 +4015,7 @@ def test_browser_manage_connect_defaults_to_loopback(monkeypatch):
|
||||
|
||||
assert resp["result"]["connected"] is True
|
||||
assert resp["result"]["url"] == "http://127.0.0.1:9222"
|
||||
assert resp["result"]["messages"] == ["Chrome is already listening on port 9222"]
|
||||
assert resp["result"]["messages"] == ["Chromium-family browser is already listening on port 9222"]
|
||||
assert urls[0] == "http://127.0.0.1:9222/json/version"
|
||||
|
||||
|
||||
@@ -4058,10 +4058,10 @@ def test_browser_manage_connect_default_local_reports_launch_hint(monkeypatch):
|
||||
assert resp["result"]["url"] == "http://127.0.0.1:9222"
|
||||
assert (
|
||||
resp["result"]["messages"][0]
|
||||
== "Chrome isn't running with remote debugging — attempting to launch..."
|
||||
== "Chromium-family browser isn't running with remote debugging — attempting to launch..."
|
||||
)
|
||||
assert any(
|
||||
"No Chrome/Chromium executable was found" in line
|
||||
"No supported Chromium-family browser executable was found" in line
|
||||
for line in resp["result"]["messages"]
|
||||
)
|
||||
assert any(
|
||||
@@ -4188,8 +4188,8 @@ def test_browser_manage_connect_default_local_retries_after_launch(monkeypatch):
|
||||
assert resp["result"]["connected"] is True
|
||||
assert resp["result"]["url"] == "http://127.0.0.1:9222"
|
||||
assert resp["result"]["messages"] == [
|
||||
"Chrome isn't running with remote debugging — attempting to launch...",
|
||||
"Chrome launched and listening on port 9222",
|
||||
"Chromium-family browser isn't running with remote debugging — attempting to launch...",
|
||||
"Chromium-family browser launched and listening on port 9222",
|
||||
]
|
||||
assert os.environ["BROWSER_CDP_URL"] == "http://127.0.0.1:9222"
|
||||
|
||||
|
||||
@@ -509,3 +509,141 @@ class TestParseErrorSignalling:
|
||||
ops, err = parse_v4a_patch(patch)
|
||||
assert err is None
|
||||
assert len(ops) == 1
|
||||
|
||||
|
||||
class TestV4ALspDiagnosticsPropagation:
|
||||
"""V4A patches must surface ``WriteResult.lsp_diagnostics`` from the
|
||||
underlying ``write_file`` calls on ``PatchResult.lsp_diagnostics``.
|
||||
|
||||
Without explicit propagation the LSP tier's output gets silently
|
||||
dropped on the V4A code path — see Copilot review #3271017295 on
|
||||
PR #29054. The shell-linter LSP skip introduced by that PR makes
|
||||
this gap visible: a ``.ts`` / ``.go`` / ``.rs`` V4A patch with LSP
|
||||
active would otherwise return ``lint = {f: {skipped: True, ...}}``
|
||||
and zero diagnostics from any channel.
|
||||
"""
|
||||
|
||||
def _build_ops_writing(self, path: str, content: str):
|
||||
"""Build a single ADD operation that writes ``content`` to ``path``."""
|
||||
# Use the V4A parser so we don't have to construct PatchOperation
|
||||
# / Hunk / Line objects by hand.
|
||||
lines = "\n".join(f"+{line}" for line in content.splitlines())
|
||||
patch_text = (
|
||||
"*** Begin Patch\n"
|
||||
f"*** Add File: {path}\n"
|
||||
f"{lines}\n"
|
||||
"*** End Patch"
|
||||
)
|
||||
ops, err = parse_v4a_patch(patch_text)
|
||||
assert err is None, err
|
||||
return ops
|
||||
|
||||
def test_lsp_diagnostics_propagated_from_write_file_on_add(self):
|
||||
"""ADD op: ``WriteResult.lsp_diagnostics`` flows through to
|
||||
``PatchResult.lsp_diagnostics``."""
|
||||
ops = self._build_ops_writing("foo.ts", "const x: number = 1\n")
|
||||
|
||||
diag_block = (
|
||||
"<diagnostics file=\"foo.ts\">\n"
|
||||
"ERROR [1:7] some diagnostic\n"
|
||||
"</diagnostics>"
|
||||
)
|
||||
|
||||
class FakeFileOps:
|
||||
def write_file(self, path, content):
|
||||
return SimpleNamespace(error=None, lsp_diagnostics=diag_block)
|
||||
|
||||
def _check_lint(self, path):
|
||||
return SimpleNamespace(to_dict=lambda: {"skipped": True})
|
||||
|
||||
result = apply_v4a_operations(ops, FakeFileOps())
|
||||
|
||||
assert result.success is True
|
||||
assert result.lsp_diagnostics == diag_block
|
||||
|
||||
def test_lsp_diagnostics_propagated_from_write_file_on_update(self):
|
||||
"""UPDATE op: ``WriteResult.lsp_diagnostics`` flows through to
|
||||
``PatchResult.lsp_diagnostics``."""
|
||||
patch_text = (
|
||||
"*** Begin Patch\n"
|
||||
"*** Update File: bar.ts\n"
|
||||
"-old\n"
|
||||
"+new\n"
|
||||
"*** End Patch"
|
||||
)
|
||||
ops, err = parse_v4a_patch(patch_text)
|
||||
assert err is None
|
||||
|
||||
diag_block = (
|
||||
"<diagnostics file=\"bar.ts\">\n"
|
||||
"ERROR [3:1] something\n"
|
||||
"</diagnostics>"
|
||||
)
|
||||
|
||||
class FakeFileOps:
|
||||
def read_file_raw(self, path):
|
||||
return SimpleNamespace(content="ctx\nold\nctx\n", error=None)
|
||||
|
||||
def write_file(self, path, content):
|
||||
return SimpleNamespace(error=None, lsp_diagnostics=diag_block)
|
||||
|
||||
def _check_lint(self, path):
|
||||
return SimpleNamespace(to_dict=lambda: {"skipped": True})
|
||||
|
||||
result = apply_v4a_operations(ops, FakeFileOps())
|
||||
|
||||
assert result.success is True
|
||||
assert result.lsp_diagnostics == diag_block
|
||||
|
||||
def test_lsp_diagnostics_none_when_no_blocks_emitted(self):
|
||||
"""When no underlying ``write_file`` produced diagnostics, the
|
||||
aggregated field stays ``None`` (so it doesn't get serialized
|
||||
as an empty string in ``PatchResult.to_dict``)."""
|
||||
ops = self._build_ops_writing("foo.py", "x = 1\n")
|
||||
|
||||
class FakeFileOps:
|
||||
def write_file(self, path, content):
|
||||
# lsp_diagnostics omitted entirely (older WriteResult shape).
|
||||
return SimpleNamespace(error=None)
|
||||
|
||||
def _check_lint(self, path):
|
||||
return SimpleNamespace(to_dict=lambda: {"success": True})
|
||||
|
||||
result = apply_v4a_operations(ops, FakeFileOps())
|
||||
|
||||
assert result.success is True
|
||||
assert result.lsp_diagnostics is None
|
||||
|
||||
def test_lsp_diagnostics_combined_across_multiple_files(self):
|
||||
"""When several files in one V4A patch produce diagnostics,
|
||||
each block appears in the combined output so per-file attribution
|
||||
is preserved."""
|
||||
patch_text = (
|
||||
"*** Begin Patch\n"
|
||||
"*** Add File: a.ts\n"
|
||||
"+const a = 1\n"
|
||||
"*** Add File: b.ts\n"
|
||||
"+const b = 2\n"
|
||||
"*** End Patch"
|
||||
)
|
||||
ops, err = parse_v4a_patch(patch_text)
|
||||
assert err is None
|
||||
|
||||
per_file = {
|
||||
"a.ts": "<diagnostics file=\"a.ts\">\nERR a\n</diagnostics>",
|
||||
"b.ts": "<diagnostics file=\"b.ts\">\nERR b\n</diagnostics>",
|
||||
}
|
||||
|
||||
class FakeFileOps:
|
||||
def write_file(self, path, content):
|
||||
return SimpleNamespace(error=None, lsp_diagnostics=per_file[path])
|
||||
|
||||
def _check_lint(self, path):
|
||||
return SimpleNamespace(to_dict=lambda: {"skipped": True})
|
||||
|
||||
result = apply_v4a_operations(ops, FakeFileOps())
|
||||
|
||||
assert result.success is True
|
||||
assert result.lsp_diagnostics is not None
|
||||
assert per_file["a.ts"] in result.lsp_diagnostics
|
||||
assert per_file["b.ts"] in result.lsp_diagnostics
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Tests for xAI TTS speech-tag handling."""
|
||||
|
||||
from unittest.mock import Mock
|
||||
|
||||
from tools.tts_tool import _apply_xai_auto_speech_tags, _generate_xai_tts
|
||||
|
||||
|
||||
def test_apply_xai_auto_speech_tags_adds_light_pause_after_first_sentence():
|
||||
text = "Bonjour Monsieur Talbot. Ceci est un test de réponse vocale."
|
||||
|
||||
assert _apply_xai_auto_speech_tags(text) == (
|
||||
"Bonjour Monsieur Talbot. [pause] Ceci est un test de réponse vocale."
|
||||
)
|
||||
|
||||
|
||||
def test_apply_xai_auto_speech_tags_preserves_explicit_tags():
|
||||
text = "Bonjour. [pause] <whisper>Déjà balisé.</whisper>"
|
||||
|
||||
assert _apply_xai_auto_speech_tags(text) == text
|
||||
|
||||
|
||||
def test_apply_xai_auto_speech_tags_preserves_all_documented_xai_tags():
|
||||
text = "Bonjour Monsieur Talbot. [sigh] <slow>Je parle lentement.</slow> <emphasis>Important.</emphasis>"
|
||||
|
||||
assert _apply_xai_auto_speech_tags(text) == text
|
||||
|
||||
|
||||
def test_generate_xai_tts_sends_auto_speech_tags_when_enabled(tmp_path, monkeypatch):
|
||||
captured = {}
|
||||
|
||||
class FakeResponse:
|
||||
content = b"mp3"
|
||||
|
||||
def raise_for_status(self):
|
||||
pass
|
||||
|
||||
def fake_post(url, headers, json, timeout):
|
||||
captured["url"] = url
|
||||
captured["headers"] = headers
|
||||
captured["json"] = json
|
||||
captured["timeout"] = timeout
|
||||
return FakeResponse()
|
||||
|
||||
monkeypatch.setenv("XAI_API_KEY", "test-xai-key")
|
||||
monkeypatch.setattr("requests.post", fake_post)
|
||||
|
||||
out = tmp_path / "out.mp3"
|
||||
_generate_xai_tts(
|
||||
"Bonjour Monsieur Talbot. Ceci est un test.",
|
||||
str(out),
|
||||
{"xai": {"voice_id": "ara", "language": "fr", "auto_speech_tags": True}},
|
||||
)
|
||||
|
||||
assert out.read_bytes() == b"mp3"
|
||||
assert captured["url"] == "https://api.x.ai/v1/tts"
|
||||
assert captured["json"]["voice_id"] == "ara"
|
||||
assert captured["json"]["language"] == "fr"
|
||||
assert captured["json"]["text"] == "Bonjour Monsieur Talbot. [pause] Ceci est un test."
|
||||
|
||||
|
||||
def test_generate_xai_tts_leaves_text_plain_by_default(tmp_path, monkeypatch):
|
||||
captured = {}
|
||||
|
||||
fake_response = Mock()
|
||||
fake_response.content = b"mp3"
|
||||
fake_response.raise_for_status.return_value = None
|
||||
|
||||
def fake_post(url, headers, json, timeout):
|
||||
captured["json"] = json
|
||||
return fake_response
|
||||
|
||||
monkeypatch.setenv("XAI_API_KEY", "test-xai-key")
|
||||
monkeypatch.setattr("requests.post", fake_post)
|
||||
|
||||
_generate_xai_tts(
|
||||
"Bonjour Monsieur Talbot. Ceci est un test.",
|
||||
str(tmp_path / "out.mp3"),
|
||||
{"xai": {"voice_id": "ara", "language": "fr"}},
|
||||
)
|
||||
|
||||
assert captured["json"]["text"] == "Bonjour Monsieur Talbot. Ceci est un test."
|
||||
Reference in New Issue
Block a user