1  import queue
  2  import threading
  3  import time
  4  from unittest.mock import patch
  5  
  6  import cli as cli_module
  7  import tools.skills_tool as skills_tool_module
  8  from cli import HermesCLI
  9  from hermes_cli.callbacks import prompt_for_secret
 10  from tools.skills_tool import set_secret_capture_callback
 11  
 12  
 13  class _FakeBuffer:
 14      def __init__(self):
 15          self.reset_called = False
 16  
 17      def reset(self):
 18          self.reset_called = True
 19  
 20  
 21  class _FakeApp:
 22      def __init__(self):
 23          self.invalidated = False
 24          self.current_buffer = _FakeBuffer()
 25  
 26      def invalidate(self):
 27          self.invalidated = True
 28  
 29  
 30  def _make_cli_stub(with_app=False):
 31      cli = HermesCLI.__new__(HermesCLI)
 32      cli._app = _FakeApp() if with_app else None
 33      cli._last_invalidate = 0.0
 34      cli._secret_state = None
 35      cli._secret_deadline = 0
 36      return cli
 37  
 38  
 39  def test_secret_capture_callback_can_be_completed_from_cli_state_machine():
 40      cli = _make_cli_stub(with_app=True)
 41      results = []
 42  
 43      with patch("hermes_cli.callbacks.save_env_value_secure") as save_secret:
 44          save_secret.return_value = {
 45              "success": True,
 46              "stored_as": "TENOR_API_KEY",
 47              "validated": False,
 48          }
 49  
 50          thread = threading.Thread(
 51              target=lambda: results.append(
 52                  cli._secret_capture_callback("TENOR_API_KEY", "Tenor API key")
 53              )
 54          )
 55          thread.start()
 56  
 57          deadline = time.time() + 2
 58          while cli._secret_state is None and time.time() < deadline:
 59              time.sleep(0.01)
 60  
 61          assert cli._secret_state is not None
 62          cli._submit_secret_response("super-secret-value")
 63          thread.join(timeout=2)
 64  
 65      assert results[0]["success"] is True
 66      assert results[0]["stored_as"] == "TENOR_API_KEY"
 67      assert results[0]["skipped"] is False
 68  
 69  
 70  def test_cancel_secret_capture_marks_setup_skipped():
 71      cli = _make_cli_stub()
 72      cli._secret_state = {
 73          "response_queue": queue.Queue(),
 74          "var_name": "TENOR_API_KEY",
 75          "prompt": "Tenor API key",
 76          "metadata": {},
 77      }
 78      cli._secret_deadline = 123
 79  
 80      cli._cancel_secret_capture()
 81  
 82      assert cli._secret_state is None
 83      assert cli._secret_deadline == 0
 84  
 85  
 86  def test_secret_capture_uses_getpass_without_tui():
 87      cli = _make_cli_stub()
 88  
 89      with patch("hermes_cli.callbacks.getpass.getpass", return_value="secret-value"), patch(
 90          "hermes_cli.callbacks.save_env_value_secure"
 91      ) as save_secret:
 92          save_secret.return_value = {
 93              "success": True,
 94              "stored_as": "TENOR_API_KEY",
 95              "validated": False,
 96          }
 97          result = prompt_for_secret(cli, "TENOR_API_KEY", "Tenor API key")
 98  
 99      assert result["success"] is True
100      assert result["stored_as"] == "TENOR_API_KEY"
101      assert result["skipped"] is False
102  
103  
104  def test_secret_capture_timeout_clears_hidden_input_buffer():
105      cli = _make_cli_stub(with_app=True)
106      cleared = {"value": False}
107  
108      def clear_buffer():
109          cleared["value"] = True
110  
111      cli._clear_secret_input_buffer = clear_buffer
112  
113      with patch("hermes_cli.callbacks.queue.Queue.get", side_effect=queue.Empty), patch(
114          "hermes_cli.callbacks._time.monotonic",
115          side_effect=[0, 121],
116      ):
117          result = prompt_for_secret(cli, "TENOR_API_KEY", "Tenor API key")
118  
119      assert result["success"] is True
120      assert result["skipped"] is True
121      assert result["reason"] == "timeout"
122      assert cleared["value"] is True
123  
124  
125  def test_cli_chat_registers_secret_capture_callback():
126      clean_config = {
127          "model": {
128              "default": "anthropic/claude-opus-4.6",
129              "base_url": "https://openrouter.ai/api/v1",
130              "provider": "auto",
131          },
132          "display": {"compact": False, "tool_progress": "all"},
133          "agent": {},
134          "terminal": {"env_type": "local"},
135      }
136  
137      with patch("cli.get_tool_definitions", return_value=[]), patch.dict(
138          "os.environ", {"LLM_MODEL": "", "HERMES_MAX_ITERATIONS": ""}, clear=False
139      ), patch.dict(cli_module.__dict__, {"CLI_CONFIG": clean_config}):
140          cli_obj = HermesCLI()
141          with patch.object(cli_obj, "_ensure_runtime_credentials", return_value=False):
142              cli_obj.chat("hello")
143  
144      try:
145          assert skills_tool_module._secret_capture_callback == cli_obj._secret_capture_callback
146      finally:
147          set_secret_capture_callback(None)