test(clarify-gateway): cover signature, timeout fallback, and notify paths for 100% coverage

Fixes #36531
This commit is contained in:
Christopher-Schulze 2026-06-24 19:15:30 +02:00 committed by Christopher
parent 07be37d996
commit 5265dfe2f5

View file

@ -250,3 +250,82 @@ class TestGatewayTextIntercept:
# Clean up
cm.clear_session("sk-tf")
class TestCoverageGaps:
"""Cover remaining branches: signature(), get_entry miss, find_awaiting
with deleted entry, cancel with None entry, timeout exception, get_notify."""
def setup_method(self):
_clear_clarify_state()
def test_entry_signature(self):
"""_ClarifyEntry.signature() returns the expected dict."""
from tools import clarify_gateway as cm
entry = cm.register("sig1", "sk", "Q?", ["A", "B"])
sig = entry.signature()
assert sig["clarify_id"] == "sig1"
assert sig["session_key"] == "sk"
assert sig["question"] == "Q?"
assert sig["choices"] == ["A", "B"]
def test_entry_signature_no_choices(self):
"""signature() returns None for choices when open-ended."""
from tools import clarify_gateway as cm
entry = cm.register("sig2", "sk", "Q?", None)
sig = entry.signature()
assert sig["choices"] is None
def test_wait_for_response_unknown_id_returns_none(self):
"""wait_for_response on a non-existent id returns None immediately."""
from tools import clarify_gateway as cm
assert cm.wait_for_response("nonexistent-id", timeout=0.1) is None
def test_find_awaiting_skips_deleted_entry(self):
"""get_pending_for_session skips entries that were removed from _entries
but still listed in _session_index."""
from tools import clarify_gateway as cm
cm.register("a1", "sk", "Q?", None)
# Manually remove from _entries but leave in _session_index
with cm._lock:
cm._entries.pop("a1", None)
# No entry to find → returns None
assert cm.get_pending_for_session("sk") is None
def test_clear_session_skips_deleted_entry(self):
"""clear_session skips entries that are None (already removed)."""
from tools import clarify_gateway as cm
cm.register("c1", "sk", "Q?", ["A"])
# Manually remove from _entries but leave in _session_index
with cm._lock:
cm._entries.pop("c1", None)
# Should return 0 cancelled (entry was already gone)
cancelled = cm.clear_session("sk")
assert cancelled == 0
def test_get_clarify_timeout_exception_returns_default(self, monkeypatch):
"""get_clarify_timeout returns 3600 when load_config raises."""
from tools import clarify_gateway as cm
monkeypatch.setattr("hermes_cli.config.load_config",
lambda: (_ for _ in ()).throw(RuntimeError("boom")))
assert cm.get_clarify_timeout() == 3600
def test_get_notify_returns_callback(self):
"""get_notify returns the registered callback."""
from tools import clarify_gateway as cm
cb = lambda entry: None
cm.register_notify("sk-notify", cb)
assert cm.get_notify("sk-notify") is cb
def test_get_notify_returns_none_when_not_registered(self):
"""get_notify returns None for an unregistered session."""
from tools import clarify_gateway as cm
assert cm.get_notify("unregistered") is None