From 91b4d9fb67f5cb5bd04dacd20c32f73d68ebff3b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 6 Sep 2026 03:05:36 -0700 Subject: [PATCH] test(fallback): keep the two persisted-config invariants; PTY harness for the picker error path Trim the salvaged suite to the behaviour contracts: a picker exception or Ctrl+C leaves config.yaml's model exactly as snapshotted (parametrized), and an absent active_provider stays absent through snapshot+restore. The mock-dispatch tests (asserting calls into our own helpers) are dropped. evals/cli_fallback_add_picker_error.py drives the real `hermes fallback add` under a Linux PTY into an ordinary picker OSError and checks the persisted model: stranded on base, restored after. --- evals/cli_fallback_add_picker_error.py | 135 +++++++++++++++++++++++++ hermes_cli/fallback_cmd.py | 3 + tests/hermes_cli/test_fallback_cmd.py | 82 ++------------- 3 files changed, 144 insertions(+), 76 deletions(-) create mode 100644 evals/cli_fallback_add_picker_error.py diff --git a/evals/cli_fallback_add_picker_error.py b/evals/cli_fallback_add_picker_error.py new file mode 100644 index 0000000000..da90b18782 --- /dev/null +++ b/evals/cli_fallback_add_picker_error.py @@ -0,0 +1,135 @@ +"""Drive the real ``hermes fallback add`` under a Linux PTY into an ordinary picker error. + +Run from the checkout with the project Python. No provider requests are sent: the picker +target is a saved custom provider with ``discover_models: false``. The auth store is made +unreadable AFTER the provider menu renders (the pre-picker snapshot has already happened), so +the canonical picker writes the temporary primary route to config.yaml and then fails inside +``deactivate_provider`` with a plain ``PermissionError`` -- not ``SystemExit``. + +The invariant under test: config.yaml ``model`` must equal the pre-picker primary afterwards. +""" +import argparse +import errno +import json +import os +from pathlib import Path +import pty +import re +import select +import signal +import struct +import subprocess +import sys +import tempfile +import termios +import time +import fcntl + +PRIMARY = {"provider": "openrouter", "default": "primary/model-a", + "base_url": "https://openrouter.ai/api/v1", "api_mode": "chat_completions"} +CONFIG = ( + "model:\n provider: openrouter\n default: primary/model-a\n" + " base_url: https://openrouter.ai/api/v1\n api_mode: chat_completions\n" + "custom_providers:\n - name: LocalLab\n base_url: http://127.0.0.1:9/v1\n" + " model: lab-model\n discover_models: false\n models:\n - lab-model\n" + "memory:\n provider: ''\n") + + +def _persisted_model(root: Path, env: dict) -> dict: + """``config.yaml`` ``model`` section as the CLI itself reads it (owner module, same env).""" + out = subprocess.run( + [sys.executable, "-c", "import json; from hermes_cli.config import load_config; " + "print(json.dumps(load_config().get('model')))"], + cwd=root, env=env, capture_output=True, text=True, check=True) + return json.loads(out.stdout.strip().splitlines()[-1]) + + +def run(root: Path, output: Path) -> dict: + with tempfile.TemporaryDirectory(prefix="hermes_test_fallback_") as home: + hh = Path(home) / ".hermes" + hh.mkdir() + (hh / "config.yaml").write_text(CONFIG, encoding="utf-8") + (hh / ".env").write_text("OPENROUTER_API_KEY=local-not-used\n", encoding="utf-8") + auth = hh / "auth.json" + auth.write_text(json.dumps({"version": 1, "providers": {}, "active_provider": "nous"})) + # A stub ``curses`` package forces every menu onto its numbered fallback so the PTY + # exchange is line-oriented (the curses UI is not what is under test here). + shim = Path(home) / "shim" / "curses" + shim.mkdir(parents=True) + (shim / "__init__.py").write_text("raise ImportError('curses disabled for PTY harness')\n") + env = {"PATH": os.environ["PATH"], "HOME": home, "HERMES_HOME": str(hh), + "PYTHONPATH": f"{shim.parent}{os.pathsep}{root}", "PYTHONUNBUFFERED": "1", + "TERM": "dumb", "LANG": "C.UTF-8"} + master, slave = pty.openpty() + fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 50, 120, 0, 0)) + proc = subprocess.Popen([sys.executable, "-m", "hermes_cli.main", "fallback", "add"], + cwd=root, env=env, stdin=slave, stdout=slave, stderr=slave, + start_new_session=True) + os.close(slave) + data = bytearray() + + def pump_until(predicate, timeout=60): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(bytes(data)): + return True + if select.select([master], [], [], 0.1)[0]: + try: + chunk = os.read(master, 65536) + except OSError as exc: + if exc.errno == errno.EIO: + return predicate(bytes(data)) + raise + if not chunk: + return predicate(bytes(data)) + data.extend(chunk) + return predicate(bytes(data)) + + try: + assert pump_until(lambda b: b"Choice [default" in b), data[-2000:] + text = bytes(data).decode(errors="replace") + row = re.search(r"(\d+)\. LocalLab", text) + assert row, text[-3000:] + # Snapshot is done (menu is up); now make the auth store unreadable so the picker's + # own deactivate_provider() fails with an ordinary OSError after writing the model. + auth.chmod(0) + os.write(master, f"{row.group(1)}\r".encode()) + offset = len(data) + assert pump_until(lambda b: b"Choice [" in b[offset:]), data[-2000:] + os.write(master, b"1\r") + exited = pump_until(lambda b: proc.poll() is not None, 90) + proc.wait(timeout=30) + auth.chmod(0o600) + model_after = _persisted_model(root, env) + text = bytes(data).decode(errors="replace") + return {"exited": exited, "returncode": proc.returncode, + "picker_error_surfaced": "PermissionError" in text, + "model_after": model_after, "primary_restored": model_after == PRIMARY, + "auth_active_provider": json.loads(auth.read_text()).get("active_provider"), + "restore_note": "Could not fully restore" in text, + "raw_path": str(output / "fallback-add-picker-error.pty")} + finally: + output.mkdir(parents=True, exist_ok=True) + (output / "fallback-add-picker-error.pty").write_bytes(data) + if proc.poll() is None: + os.killpg(proc.pid, signal.SIGKILL) + proc.wait(timeout=30) + os.close(master) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--root", required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--expect", choices=("stranded", "restored"), required=True) + args = parser.parse_args() + result = run(Path(args.root).resolve(), args.output) + (args.output / "results.json").write_text(json.dumps(result, indent=2) + "\n") + print(json.dumps(result, indent=2)) + assert result["exited"] and result["returncode"] != 0, result + assert result["picker_error_surfaced"], result + assert result["primary_restored"] == (args.expect == "restored"), result + + +if __name__ == "__main__": + main() diff --git a/hermes_cli/fallback_cmd.py b/hermes_cli/fallback_cmd.py index 6867919dff..866fb88d9a 100644 --- a/hermes_cli/fallback_cmd.py +++ b/hermes_cli/fallback_cmd.py @@ -68,6 +68,7 @@ def _restore_auth_active_provider(value: Any) -> None: store["active_provider"] = value _save_auth_store(store) + def _restore_model_cfg(model_before: Any) -> None: """Restore ``config["model"]`` to a previously-captured snapshot.""" from hermes_cli.config import load_config, save_config @@ -93,6 +94,7 @@ def _restore_primary_route(model_before: Any, active_provider_before: Any) -> No details = "; ".join(str(exc) for exc in errors) raise RuntimeError(f"Could not fully restore the primary route: {details}") from errors[0] + def _entries(n: int) -> str: return f"{n} {'entry' if n == 1 else 'entries'}" @@ -191,6 +193,7 @@ def cmd_fallback_add(args) -> None: print(f" Chain is now {_entries(len(chain))} long.\n") print(" Run `hermes fallback list` to view, or `hermes fallback remove` to delete.") + def cmd_fallback_remove(args) -> None: # noqa: ARG001 """Pick an entry from the chain and remove it.""" from hermes_cli.config import save_config diff --git a/tests/hermes_cli/test_fallback_cmd.py b/tests/hermes_cli/test_fallback_cmd.py index 3e0393f282..a20a85293b 100644 --- a/tests/hermes_cli/test_fallback_cmd.py +++ b/tests/hermes_cli/test_fallback_cmd.py @@ -117,20 +117,6 @@ class TestListCommand: # --------------------------------------------------------------------------- class TestAddCommand: - def test_auth_snapshot_failure_aborts_before_picker(self): - from hermes_cli import fallback_cmd - - picker = object() - with patch( - "hermes_cli.auth._load_auth_store", - side_effect=OSError("auth read failed"), - ), patch( - "hermes_cli.main.select_provider_and_model", - picker, - ), patch("hermes_cli.main._require_tty"): - with pytest.raises(OSError, match="auth read failed"): - fallback_cmd.cmd_fallback_add(types.SimpleNamespace()) - def test_add_appends_new_entry(self, isolated_home, capsys): _write_config(isolated_home, { "model": {"provider": "anthropic", "default": "claude-sonnet-4-6"}, @@ -243,65 +229,6 @@ class TestAddCommand: } ] - def test_post_picker_config_read_failure_restores_route(self): - from hermes_cli import fallback_cmd - - primary_model = { - "provider": "anthropic", - "default": "claude-sonnet-4-6", - } - post_read_error = OSError("post-picker config read failed") - reads = iter([{"model": primary_model}, post_read_error]) - - def load_config(): - value = next(reads) - if isinstance(value, BaseException): - raise value - return value - - restore_calls = [] - with patch("hermes_cli.config.load_config", side_effect=load_config), patch( - "hermes_cli.main._require_tty" - ), patch("hermes_cli.main.select_provider_and_model"), patch.object( - fallback_cmd, "_snapshot_auth_active_provider", return_value="old-provider" - ), patch.object( - fallback_cmd, - "_restore_primary_route", - side_effect=lambda model, provider: restore_calls.append((model, provider)), - ): - with pytest.raises(OSError) as exc_info: - fallback_cmd.cmd_fallback_add(types.SimpleNamespace()) - - assert exc_info.value is post_read_error - assert restore_calls == [(primary_model, "old-provider")] - - @pytest.mark.parametrize( - ("failure_type", "message"), - [ - (OSError, "config write failed"), - (KeyboardInterrupt, "config restore interrupted"), - ], - ) - def test_restore_attempts_auth_after_model_restore_failure( - self, failure_type, message - ): - from hermes_cli import fallback_cmd - - auth_calls = [] - with patch.object( - fallback_cmd, - "_restore_model_cfg", - side_effect=failure_type(message), - ), patch.object( - fallback_cmd, - "_restore_auth_active_provider", - side_effect=lambda value: auth_calls.append(value), - ): - with pytest.raises(RuntimeError, match=message): - fallback_cmd._restore_primary_route("old-model", "old-provider") - - assert auth_calls == ["old-provider"] - def test_restore_preserves_absent_active_provider(self): from contextlib import nullcontext @@ -321,9 +248,13 @@ class TestAddCommand: assert "active_provider" not in store + @pytest.mark.parametrize("picker_error", [LookupError("picker failed"), KeyboardInterrupt()], + ids=["exception", "ctrl-c"]) def test_picker_failure_restores_persisted_primary_without_masking_error( - self, isolated_home + self, isolated_home, picker_error ): + """An ordinary picker exception or a Ctrl+C mid-picker must leave config.yaml's + ``model`` exactly as it was before ``fallback add`` started (base only handled SystemExit).""" from hermes_cli import fallback_cmd primary_model = { @@ -333,7 +264,6 @@ class TestAddCommand: "api_mode": "anthropic_messages", } _write_config(isolated_home, {"model": primary_model, "theme": "midnight"}) - picker_error = LookupError("picker failed") def failing_picker(args=None): from hermes_cli.config import load_config, save_config @@ -352,7 +282,7 @@ class TestAddCommand: "hermes_cli.main.select_provider_and_model", side_effect=failing_picker, ), patch("hermes_cli.main._require_tty"): - with pytest.raises(LookupError, match="picker failed") as exc_info: + with pytest.raises(type(picker_error)) as exc_info: fallback_cmd.cmd_fallback_add(types.SimpleNamespace()) assert exc_info.value is picker_error