1  """Tests for hermes_cli/webhook.py — webhook subscription CLI."""
  2  
  3  import json
  4  import os
  5  import pytest
  6  from argparse import Namespace
  7  from pathlib import Path
  8  
  9  from hermes_cli.webhook import (
 10      webhook_command,
 11      _load_subscriptions,
 12      _save_subscriptions,
 13      _subscriptions_path,
 14      _is_webhook_enabled,
 15  )
 16  
 17  
 18  @pytest.fixture(autouse=True)
 19  def _isolate(tmp_path, monkeypatch):
 20      monkeypatch.setenv("HERMES_HOME", str(tmp_path))
 21      # Default: webhooks enabled (most tests need this)
 22      monkeypatch.setattr(
 23          "hermes_cli.webhook._is_webhook_enabled", lambda: True
 24      )
 25  
 26  
 27  def _make_args(**kwargs):
 28      defaults = {
 29          "webhook_action": None,
 30          "name": "",
 31          "prompt": "",
 32          "events": "",
 33          "description": "",
 34          "skills": "",
 35          "deliver": "log",
 36          "deliver_chat_id": "",
 37          "secret": "",
 38          "payload": "",
 39      }
 40      defaults.update(kwargs)
 41      return Namespace(**defaults)
 42  
 43  
 44  class TestSubscribe:
 45      def test_basic_create(self, capsys):
 46          webhook_command(_make_args(webhook_action="subscribe", name="test-hook"))
 47          out = capsys.readouterr().out
 48          assert "Created" in out
 49          assert "/webhooks/test-hook" in out
 50          subs = _load_subscriptions()
 51          assert "test-hook" in subs
 52  
 53      def test_with_options(self, capsys):
 54          webhook_command(_make_args(
 55              webhook_action="subscribe",
 56              name="gh-issues",
 57              events="issues,pull_request",
 58              prompt="Issue: {issue.title}",
 59              deliver="telegram",
 60              deliver_chat_id="12345",
 61              description="Watch GitHub",
 62          ))
 63          subs = _load_subscriptions()
 64          route = subs["gh-issues"]
 65          assert route["events"] == ["issues", "pull_request"]
 66          assert route["prompt"] == "Issue: {issue.title}"
 67          assert route["deliver"] == "telegram"
 68          assert route["deliver_extra"] == {"chat_id": "12345"}
 69  
 70      def test_custom_secret(self):
 71          webhook_command(_make_args(
 72              webhook_action="subscribe", name="s", secret="my-secret"
 73          ))
 74          assert _load_subscriptions()["s"]["secret"] == "my-secret"
 75  
 76      def test_auto_secret(self):
 77          webhook_command(_make_args(webhook_action="subscribe", name="s"))
 78          secret = _load_subscriptions()["s"]["secret"]
 79          assert len(secret) > 20
 80  
 81      def test_update(self, capsys):
 82          webhook_command(_make_args(webhook_action="subscribe", name="x", prompt="v1"))
 83          webhook_command(_make_args(webhook_action="subscribe", name="x", prompt="v2"))
 84          out = capsys.readouterr().out
 85          assert "Updated" in out
 86          assert _load_subscriptions()["x"]["prompt"] == "v2"
 87  
 88      def test_invalid_name(self, capsys):
 89          webhook_command(_make_args(webhook_action="subscribe", name="bad name!"))
 90          out = capsys.readouterr().out
 91          assert "Error" in out or "Invalid" in out
 92          assert _load_subscriptions() == {}
 93  
 94  
 95  class TestList:
 96      def test_empty(self, capsys):
 97          webhook_command(_make_args(webhook_action="list"))
 98          out = capsys.readouterr().out
 99          assert "No dynamic" in out
100  
101      def test_with_entries(self, capsys):
102          webhook_command(_make_args(webhook_action="subscribe", name="a"))
103          webhook_command(_make_args(webhook_action="subscribe", name="b"))
104          capsys.readouterr()  # clear
105          webhook_command(_make_args(webhook_action="list"))
106          out = capsys.readouterr().out
107          assert "2 webhook" in out
108          assert "a" in out
109          assert "b" in out
110  
111  
112  class TestRemove:
113      def test_remove_existing(self, capsys):
114          webhook_command(_make_args(webhook_action="subscribe", name="temp"))
115          webhook_command(_make_args(webhook_action="remove", name="temp"))
116          out = capsys.readouterr().out
117          assert "Removed" in out
118          assert _load_subscriptions() == {}
119  
120      def test_remove_nonexistent(self, capsys):
121          webhook_command(_make_args(webhook_action="remove", name="nope"))
122          out = capsys.readouterr().out
123          assert "No subscription" in out
124  
125      def test_selective_remove(self):
126          webhook_command(_make_args(webhook_action="subscribe", name="keep"))
127          webhook_command(_make_args(webhook_action="subscribe", name="drop"))
128          webhook_command(_make_args(webhook_action="remove", name="drop"))
129          subs = _load_subscriptions()
130          assert "keep" in subs
131          assert "drop" not in subs
132  
133  
134  class TestPersistence:
135      def test_file_written(self):
136          webhook_command(_make_args(webhook_action="subscribe", name="persist"))
137          path = _subscriptions_path()
138          assert path.exists()
139          data = json.loads(path.read_text())
140          assert "persist" in data
141  
142      def test_corrupted_file(self):
143          path = _subscriptions_path()
144          path.parent.mkdir(parents=True, exist_ok=True)
145          path.write_text("broken{{{")
146          assert _load_subscriptions() == {}
147  
148  
149  class TestWebhookEnabledGate:
150      def test_blocks_when_disabled(self, capsys, monkeypatch):
151          monkeypatch.setattr("hermes_cli.webhook._is_webhook_enabled", lambda: False)
152          webhook_command(_make_args(webhook_action="subscribe", name="blocked"))
153          out = capsys.readouterr().out
154          assert "not enabled" in out.lower()
155          assert "hermes gateway setup" in out
156          assert _load_subscriptions() == {}
157  
158      def test_blocks_list_when_disabled(self, capsys, monkeypatch):
159          monkeypatch.setattr("hermes_cli.webhook._is_webhook_enabled", lambda: False)
160          webhook_command(_make_args(webhook_action="list"))
161          out = capsys.readouterr().out
162          assert "not enabled" in out.lower()
163  
164      def test_allows_when_enabled(self, capsys):
165          # _is_webhook_enabled already patched to True by autouse fixture
166          webhook_command(_make_args(webhook_action="subscribe", name="allowed"))
167          out = capsys.readouterr().out
168          assert "Created" in out
169          assert "allowed" in _load_subscriptions()
170  
171      def test_real_check_disabled(self, monkeypatch):
172          monkeypatch.setattr(
173              "hermes_cli.webhook._get_webhook_config",
174              lambda: {},
175          )
176          monkeypatch.setattr(
177              "hermes_cli.webhook._is_webhook_enabled",
178              lambda: bool({}.get("enabled")),
179          )
180          import hermes_cli.webhook as wh_mod
181          assert wh_mod._is_webhook_enabled() is False
182  
183      def test_real_check_enabled(self, monkeypatch):
184          monkeypatch.setattr(
185              "hermes_cli.webhook._is_webhook_enabled",
186              lambda: True,
187          )
188          import hermes_cli.webhook as wh_mod
189          assert wh_mod._is_webhook_enabled() is True