mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-28 18:19:28 +00:00
10b7ab5cb6
Cross-surface coverage #23768 missed (per review feedback): - tools/clarify_gateway.py: _ClarifyEntry carries a multi_select flag (register() accepts it; signature() exposes it to adapters). _coerce_text_response now parses multi-select replies — comma- or space-separated numbers ('1,3' / '1 3'), exact labels, dedup — into a JSON array string that _parse_multi_select_response decodes into a list. Out-of-range/unknown tokens reject the reply (native button UI) or fall back to custom text (awaiting_text/'Other' mode). - gateway/run.py: _clarify_callback_sync accepts multi_select and registers it on the pending entry. - gateway/platforms/base.py: default numbered-list text fallback tells the user multiple selections are allowed and how to reply. - tui_gateway/server.py: clarify_callback passes multi_select through the clarify.request payload as a hint; renderers without checkbox support ignore the field and remain single-select-compatible. - tests: 13 new gateway tests (flag storage, comma/space/single-number parsing, label matching, out-of-range rejection, dedup, end-to-end resolve, single-select regressions).
558 lines
22 KiB
Python
558 lines
22 KiB
Python
"""Tests for the gateway-side clarify primitive (tools/clarify_gateway.py).
|
|
|
|
The clarify tool needs to ask the user a question and block the agent
|
|
thread until they respond. These tests cover the module-level state
|
|
machine: register, wait, resolve via button, resolve via text-fallback,
|
|
"Other"-button text-capture flip, timeout, session boundary cleanup.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
|
|
|
|
def _clear_clarify_state():
|
|
"""Reset module-level state between tests."""
|
|
from tools import clarify_gateway as cm
|
|
with cm._lock:
|
|
cm._entries.clear()
|
|
cm._session_index.clear()
|
|
cm._notify_cbs.clear()
|
|
|
|
|
|
class TestClarifyPrimitive:
|
|
"""Core register/wait/resolve mechanics."""
|
|
|
|
def setup_method(self):
|
|
_clear_clarify_state()
|
|
|
|
def test_button_choice_resolves_wait(self):
|
|
"""resolve_gateway_clarify unblocks wait_for_response with the chosen string."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id1", "sk1", "Pick one", ["A", "B", "C"])
|
|
|
|
def resolver():
|
|
time.sleep(0.05)
|
|
cm.resolve_gateway_clarify("id1", "B")
|
|
|
|
threading.Thread(target=resolver).start()
|
|
result = cm.wait_for_response("id1", timeout=10.0)
|
|
assert result == "B"
|
|
|
|
def test_open_ended_auto_awaits_text(self):
|
|
"""Clarify with no choices is in text-capture mode immediately."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
entry = cm.register("id2", "sk2", "Free form?", None)
|
|
assert entry.awaiting_text is True
|
|
|
|
# get_pending_for_session returns the entry so the gateway
|
|
# text-intercept can find it.
|
|
pending = cm.get_pending_for_session("sk2")
|
|
assert pending is not None
|
|
assert pending.clarify_id == "id2"
|
|
|
|
def test_button_choice_does_not_auto_await(self):
|
|
"""Multi-choice clarify should NOT be in text-capture mode initially."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
entry = cm.register("id3", "sk3", "Pick", ["X", "Y"])
|
|
assert entry.awaiting_text is False
|
|
assert cm.get_pending_for_session("sk3") is None
|
|
|
|
def test_include_choice_prompts_returns_multi_choice_entry(self):
|
|
"""Gateway typed replies must see active choice prompts too."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id3b", "sk3b", "Pick", ["X", "Y"])
|
|
pending = cm.get_pending_for_session("sk3b", include_choice_prompts=True)
|
|
assert pending is not None
|
|
assert pending.clarify_id == "id3b"
|
|
|
|
def test_resolve_text_response_maps_numeric_choice(self):
|
|
"""Typed numbers should resolve to the canonical choice string."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id3c", "sk3c", "Pick", ["X", "Y"])
|
|
assert cm.resolve_text_response_for_session("sk3c", "2") is True
|
|
assert cm.wait_for_response("id3c", timeout=0.1) == "Y"
|
|
|
|
def test_resolve_text_response_accepts_custom_other_text(self):
|
|
"""Arbitrary typed text should resolve as a custom Other answer when awaiting_text is True."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id3d", "sk3d", "Pick", ["X", "Y"])
|
|
# Flip to text-capture mode (user picked "Other")
|
|
cm.mark_awaiting_text("id3d")
|
|
custom = "None of those are valid options"
|
|
assert cm.resolve_text_response_for_session("sk3d", custom) is True
|
|
assert cm.wait_for_response("id3d", timeout=0.1) == custom
|
|
|
|
def test_resolve_text_rejects_arbitrary_prose_for_native_multi_choice(self):
|
|
"""Native interactive multi-choice clarifies reject arbitrary prose unless awaiting_text is True."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
# Native multi-choice (buttons, not awaiting text)
|
|
cm.register("id-strict", "sk-strict", "Pick one", ["A", "B", "C"])
|
|
|
|
# Arbitrary prose should be rejected
|
|
assert cm.resolve_text_response_for_session("sk-strict", "just checking the visual UI") is False
|
|
assert cm.resolve_text_response_for_session("sk-strict", "present 3 buttons") is False
|
|
|
|
# Numeric choices should still work
|
|
assert cm.resolve_text_response_for_session("sk-strict", "2") is True
|
|
assert cm.wait_for_response("id-strict", timeout=0.1) == "B"
|
|
|
|
# Exact label match should still work
|
|
cm.register("id-strict2", "sk-strict2", "Pick", ["Option Alpha", "Option Beta"])
|
|
assert cm.resolve_text_response_for_session("sk-strict2", "Option Alpha") is True
|
|
assert cm.wait_for_response("id-strict2", timeout=0.1) == "Option Alpha"
|
|
|
|
def test_text_fallback_mode_allows_any_text(self):
|
|
"""Text fallback mode (after base send_clarify calls mark_awaiting_text) accepts any text."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
entry = cm.register("id-tf", "sk-tf", "Pick one", ["A", "B", "C"])
|
|
assert entry.awaiting_text is False
|
|
|
|
# Simulate base send_clarify calling mark_awaiting_text
|
|
cm.mark_awaiting_text("id-tf")
|
|
assert entry.awaiting_text is True
|
|
|
|
# Now arbitrary text is accepted
|
|
custom = "I choose a custom answer"
|
|
assert cm.resolve_text_response_for_session("sk-tf", custom) is True
|
|
assert cm.wait_for_response("id-tf", timeout=0.1) == custom
|
|
|
|
# Numeric choices also work
|
|
cm.register("id-tf2", "sk-tf2", "Pick", ["X", "Y"])
|
|
cm.mark_awaiting_text("id-tf2")
|
|
assert cm.resolve_text_response_for_session("sk-tf2", "1") is True
|
|
assert cm.wait_for_response("id-tf2", timeout=0.1) == "X"
|
|
|
|
def test_other_button_flips_to_text_mode(self):
|
|
"""mark_awaiting_text makes get_pending_for_session find the entry."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id4", "sk4", "Pick", ["X", "Y"])
|
|
assert cm.get_pending_for_session("sk4") is None
|
|
|
|
flipped = cm.mark_awaiting_text("id4")
|
|
assert flipped is True
|
|
|
|
pending = cm.get_pending_for_session("sk4")
|
|
assert pending is not None
|
|
assert pending.clarify_id == "id4"
|
|
|
|
def test_mark_awaiting_text_unknown_id(self):
|
|
"""mark_awaiting_text on a non-existent id returns False."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
assert cm.mark_awaiting_text("nope") is False
|
|
|
|
def test_timeout_returns_none(self):
|
|
"""wait_for_response returns None when no resolve fires within the timeout."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id5", "sk5", "Q?", ["A"])
|
|
result = cm.wait_for_response("id5", timeout=0.2)
|
|
assert result is None
|
|
|
|
def test_resolve_unknown_id_returns_false(self):
|
|
"""resolve_gateway_clarify is idempotent on unknown ids."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
assert cm.resolve_gateway_clarify("nope", "anything") is False
|
|
|
|
def test_resolve_after_wait_completes_is_noop(self):
|
|
"""A late resolve on a finished entry doesn't blow up."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id6", "sk6", "Q?", ["A"])
|
|
# Time out, entry gets cleaned up
|
|
cm.wait_for_response("id6", timeout=0.1)
|
|
# Late button click — should not raise
|
|
result = cm.resolve_gateway_clarify("id6", "A")
|
|
assert result is False
|
|
|
|
def test_clear_session_cancels_pending_entries(self):
|
|
"""clear_session unblocks blocked threads with empty response."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id7", "sk7", "Q?", ["A"])
|
|
|
|
def waiter():
|
|
return cm.wait_for_response("id7", timeout=10.0)
|
|
|
|
with ThreadPoolExecutor(1) as pool:
|
|
fut = pool.submit(waiter)
|
|
time.sleep(0.05)
|
|
cancelled = cm.clear_session("sk7")
|
|
assert cancelled == 1
|
|
result = fut.result(timeout=10.0)
|
|
# clear_session sets response="" then the wait returns it
|
|
assert result == ""
|
|
|
|
def test_has_pending(self):
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id8", "sk8", "Q?", ["A"])
|
|
assert cm.has_pending("sk8") is True
|
|
assert cm.has_pending("nonexistent") is False
|
|
|
|
def test_notify_register_unregister_clears_pending(self):
|
|
"""unregister_notify cancels any pending clarify so threads unwind."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("id9", "sk9", "Q?", ["A"])
|
|
|
|
def waiter():
|
|
return cm.wait_for_response("id9", timeout=10.0)
|
|
|
|
with ThreadPoolExecutor(1) as pool:
|
|
fut = pool.submit(waiter)
|
|
time.sleep(0.05)
|
|
|
|
cm.register_notify("sk9", lambda entry: None)
|
|
cm.unregister_notify("sk9")
|
|
|
|
# unregister_notify calls clear_session; thread unwinds
|
|
result = fut.result(timeout=10.0)
|
|
assert result == ""
|
|
|
|
def test_session_index_isolation(self):
|
|
"""Entries from different sessions don't leak across get_pending lookups."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("idA", "alpha", "Q?", None) # auto-await text
|
|
cm.register("idB", "beta", "Q?", None) # auto-await text
|
|
|
|
a = cm.get_pending_for_session("alpha")
|
|
b = cm.get_pending_for_session("beta")
|
|
assert a is not None and a.clarify_id == "idA"
|
|
assert b is not None and b.clarify_id == "idB"
|
|
|
|
def test_clarify_timeout_config_default(self):
|
|
"""get_clarify_timeout returns a positive int (default 3600)."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
timeout = cm.get_clarify_timeout()
|
|
# Default 3600s OR whatever is in the user's loaded config.
|
|
# Floor check: must be a positive int, not crashed.
|
|
assert isinstance(timeout, int)
|
|
assert timeout > 0
|
|
|
|
|
|
class TestGatewayTextIntercept:
|
|
"""The gateway's _handle_message intercepts text replies to pending clarifies."""
|
|
|
|
def setup_method(self):
|
|
_clear_clarify_state()
|
|
|
|
def test_get_pending_for_session_returns_oldest_text_awaiting(self):
|
|
"""When two clarifies are pending, get_pending_for_session returns the
|
|
first that is awaiting_text (the older one if both)."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
# Older multi-choice (not awaiting text)
|
|
cm.register("first", "sk", "Q1?", ["A"])
|
|
# Newer open-ended (awaiting text)
|
|
cm.register("second", "sk", "Q2?", None)
|
|
|
|
pending = cm.get_pending_for_session("sk")
|
|
# The newer one is awaiting text; the older isn't.
|
|
assert pending is not None
|
|
assert pending.clarify_id == "second"
|
|
|
|
# Now flip the first to text mode too. Both are awaiting text,
|
|
# FIFO returns the older one.
|
|
cm.mark_awaiting_text("first")
|
|
pending2 = cm.get_pending_for_session("sk")
|
|
assert pending2 is not None
|
|
assert pending2.clarify_id == "first"
|
|
def test_text_fallback_enables_awaiting_text_for_multi_choice(self):
|
|
"""When base send_clarify renders choices as text, mark_awaiting_text
|
|
is called so the gateway text-intercept can capture the reply."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
entry = cm.register("id-tf", "sk-tf", "Pick one", ["A", "B", "C"])
|
|
# Initially, multi-choice does NOT await text (button path)
|
|
assert entry.awaiting_text is False
|
|
|
|
# After the base send_clarify text fallback calls mark_awaiting_text:
|
|
flipped = cm.mark_awaiting_text("id-tf")
|
|
assert flipped is True
|
|
|
|
# Now get_pending_for_session should find it
|
|
pending = cm.get_pending_for_session("sk-tf")
|
|
assert pending is not None
|
|
assert pending.clarify_id == "id-tf"
|
|
|
|
# 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
|
|
|
|
|
|
class TestClarifyTimeoutResolution:
|
|
"""resolve_clarify_timeout is the single source of truth for the clarify
|
|
timeout, shared by the CLI, TUI/desktop, and messaging-gateway paths."""
|
|
|
|
def test_canonical_agent_key(self):
|
|
from tools import clarify_gateway as cm
|
|
|
|
assert cm.resolve_clarify_timeout({"agent": {"clarify_timeout": 900}}) == 900
|
|
|
|
def test_legacy_clarify_key_overrides(self):
|
|
"""An explicitly-set legacy top-level clarify.timeout wins, for
|
|
back-compat with users who set it before agent.clarify_timeout existed."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
cfg = {"clarify": {"timeout": 42}, "agent": {"clarify_timeout": 900}}
|
|
assert cm.resolve_clarify_timeout(cfg) == 42
|
|
|
|
def test_default_when_unset(self):
|
|
from tools import clarify_gateway as cm
|
|
|
|
assert cm.resolve_clarify_timeout({}) == 3600
|
|
|
|
def test_non_numeric_falls_back_to_default(self):
|
|
from tools import clarify_gateway as cm
|
|
|
|
assert cm.resolve_clarify_timeout({"agent": {"clarify_timeout": "nope"}}) == 3600
|
|
|
|
def test_non_positive_preserved_as_unlimited_sentinel(self):
|
|
"""<= 0 is passed through verbatim — the waiting loops read it as
|
|
'unlimited', so the resolver must not clamp it to a positive default."""
|
|
from tools import clarify_gateway as cm
|
|
|
|
assert cm.resolve_clarify_timeout({"agent": {"clarify_timeout": 0}}) == 0
|
|
assert cm.resolve_clarify_timeout({"clarify": {"timeout": -1}}) == -1
|
|
|
|
|
|
class TestUnlimitedWait:
|
|
"""timeout <= 0 makes wait_for_response block until the answer arrives
|
|
instead of auto-skipping."""
|
|
|
|
def setup_method(self):
|
|
_clear_clarify_state()
|
|
|
|
def test_zero_timeout_waits_until_resolved(self):
|
|
from tools import clarify_gateway as cm
|
|
|
|
cm.register("u1", "sk", "Q?", ["A", "B"])
|
|
result_box = {}
|
|
|
|
def waiter():
|
|
result_box["r"] = cm.wait_for_response("u1", timeout=0)
|
|
|
|
t = threading.Thread(target=waiter)
|
|
t.start()
|
|
# An unlimited wait cannot finish while nothing resolves it: still
|
|
# running after a comfortable margin (old code auto-skipped at once).
|
|
t.join(timeout=1.5)
|
|
assert t.is_alive()
|
|
|
|
# Once resolved, the unlimited wait returns the real answer.
|
|
cm.resolve_gateway_clarify("u1", "B")
|
|
t.join(timeout=5.0)
|
|
assert not t.is_alive()
|
|
assert result_box["r"] == "B"
|
|
|
|
|
|
class TestMultiSelectTextFallback:
|
|
"""Multi-select clarifies via the gateway text fallback.
|
|
|
|
The adapter's numbered-list fallback asks the user to reply with
|
|
comma/space-separated numbers; _coerce_text_response must map those to a
|
|
JSON array of choice labels (which _parse_multi_select_response on the
|
|
tool side decodes into a list).
|
|
"""
|
|
|
|
def setup_method(self):
|
|
_clear_clarify_state()
|
|
|
|
def _register_multi(self, cid="m1", choices=("A", "B", "C")):
|
|
from tools import clarify_gateway as cm
|
|
entry = cm.register(cid, "sk", "Pick some", list(choices), multi_select=True)
|
|
# Text fallback path always flips awaiting_text on.
|
|
cm.mark_awaiting_text(cid)
|
|
return entry
|
|
|
|
def test_register_stores_multi_select_flag(self):
|
|
entry = self._register_multi()
|
|
assert entry.multi_select is True
|
|
assert entry.signature()["multi_select"] is True
|
|
|
|
def test_register_default_multi_select_false(self):
|
|
from tools import clarify_gateway as cm
|
|
entry = cm.register("s1", "sk", "Q?", ["A"])
|
|
assert entry.multi_select is False
|
|
assert entry.signature()["multi_select"] is False
|
|
|
|
def test_multi_select_without_choices_is_ignored(self):
|
|
"""multi_select on an open-ended clarify is meaningless — dropped."""
|
|
from tools import clarify_gateway as cm
|
|
entry = cm.register("s2", "sk", "Q?", None, multi_select=True)
|
|
assert entry.multi_select is False
|
|
|
|
def test_comma_separated_numbers(self):
|
|
import json
|
|
from tools import clarify_gateway as cm
|
|
entry = self._register_multi()
|
|
coerced = cm._coerce_text_response(entry, "1, 3")
|
|
assert json.loads(coerced) == ["A", "C"]
|
|
|
|
def test_space_separated_numbers(self):
|
|
import json
|
|
from tools import clarify_gateway as cm
|
|
entry = self._register_multi()
|
|
coerced = cm._coerce_text_response(entry, "1 3")
|
|
assert json.loads(coerced) == ["A", "C"]
|
|
|
|
def test_single_number(self):
|
|
import json
|
|
from tools import clarify_gateway as cm
|
|
entry = self._register_multi()
|
|
coerced = cm._coerce_text_response(entry, "2")
|
|
assert json.loads(coerced) == ["B"]
|
|
|
|
def test_choice_labels_comma_separated(self):
|
|
import json
|
|
from tools import clarify_gateway as cm
|
|
entry = self._register_multi()
|
|
coerced = cm._coerce_text_response(entry, "a, C")
|
|
assert json.loads(coerced) == ["A", "C"]
|
|
|
|
def test_out_of_range_number_rejected_but_custom_text_kept(self):
|
|
"""Out-of-range numbers don't parse as a selection; awaiting_text
|
|
mode falls back to accepting the raw text as a custom answer."""
|
|
from tools import clarify_gateway as cm
|
|
entry = self._register_multi()
|
|
assert cm._coerce_multi_select_text(entry, "1, 9") is None
|
|
# awaiting_text (text fallback) keeps the raw reply as custom text
|
|
assert cm._coerce_text_response(entry, "1, 9") == "1, 9"
|
|
|
|
def test_out_of_range_rejected_for_native_button_ui(self):
|
|
"""Without awaiting_text (button UI), a bad selection rejects the
|
|
reply entirely so it flows through as a normal message."""
|
|
from tools import clarify_gateway as cm
|
|
entry = cm.register("m2", "sk", "Pick some", ["A", "B"], multi_select=True)
|
|
assert cm._coerce_text_response(entry, "5") is None
|
|
assert cm._coerce_text_response(entry, "random prose") is None
|
|
|
|
def test_duplicate_selections_deduped(self):
|
|
import json
|
|
from tools import clarify_gateway as cm
|
|
entry = self._register_multi()
|
|
coerced = cm._coerce_text_response(entry, "1, 1, 2")
|
|
assert json.loads(coerced) == ["A", "B"]
|
|
|
|
def test_resolve_text_response_end_to_end(self):
|
|
"""resolve_text_response_for_session delivers the JSON array to the waiter."""
|
|
import json
|
|
from tools import clarify_gateway as cm
|
|
self._register_multi(cid="m3")
|
|
result_box = {}
|
|
|
|
def waiter():
|
|
result_box["r"] = cm.wait_for_response("m3", timeout=5)
|
|
|
|
t = threading.Thread(target=waiter)
|
|
t.start()
|
|
time.sleep(0.05)
|
|
assert cm.resolve_text_response_for_session("sk", "1,2") is True
|
|
t.join(timeout=5)
|
|
assert json.loads(result_box["r"]) == ["A", "B"]
|
|
|
|
def test_single_select_regression_numeric(self):
|
|
"""Single-select coercion unchanged: '2' maps to the choice label string."""
|
|
from tools import clarify_gateway as cm
|
|
entry = cm.register("s3", "sk", "Q?", ["A", "B", "C"])
|
|
assert cm._coerce_text_response(entry, "2") == "B"
|
|
|
|
def test_single_select_regression_label(self):
|
|
from tools import clarify_gateway as cm
|
|
entry = cm.register("s4", "sk", "Q?", ["A", "B"])
|
|
assert cm._coerce_text_response(entry, "b") == "B"
|