Suriyakumarvijayanayagam
58bfa07385
- Tool.to_schema() emits input_schema (Anthropic Messages API) instead of OpenAI-style parameters - QuestionManager: remove questions from _pending on answer and timeout, reject double answers, add list_pending() - core/skills/__init__.py so skills are importable as a package - revert optional-import fallbacks in logger/message_bus: nats-py and loguru are hard requirements, a broken install should fail loudly - tests for wait_for_answer happy path and timeout cleanup Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012AJLYcbTHCe45fyFnMfEin
83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""Unit tests for ToolRegistry and Interactive Question Asking."""
|
||
import pytest
|
||
import asyncio
|
||
from core.tool_registry import Tool, ToolParameter, ToolRegistry
|
||
from core.interactive_question import QuestionManager, QuestionOption, question_manager
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_tool_registry_registration_and_execution():
|
||
registry = ToolRegistry()
|
||
|
||
async def sample_handler(order_id: str, count: int = 1):
|
||
return {"order": order_id, "items": count}
|
||
|
||
tool = Tool(
|
||
name="test_tool",
|
||
description="A test tool",
|
||
handler=sample_handler,
|
||
parameters=[
|
||
ToolParameter(name="order_id", type="string", description="Order ID"),
|
||
ToolParameter(name="count", type="integer", description="Count", required=False, default=1),
|
||
],
|
||
)
|
||
registry.register(tool)
|
||
|
||
assert registry.get("test_tool") is not None
|
||
schemas = registry.get_schemas()
|
||
assert len(schemas) == 1
|
||
assert schemas[0]["name"] == "test_tool"
|
||
assert "order_id" in schemas[0]["input_schema"]["properties"]
|
||
assert "parameters" not in schemas[0] # Anthropic shape, not OpenAI
|
||
|
||
result = await registry.execute_tool("test_tool", order_id="ORD-100")
|
||
assert result == {"order": "ORD-100", "items": 1}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_question_manager_flow():
|
||
qm = QuestionManager()
|
||
q = qm.create_question(
|
||
prompt="Select target batch:",
|
||
options=[
|
||
{"id": "morning", "label": "Morning Batch", "badge": "12 AM–9 AM"},
|
||
{"id": "afternoon", "label": "Afternoon Batch", "badge": "9 AM–4 PM"},
|
||
],
|
||
allow_custom_input=False,
|
||
)
|
||
|
||
assert q.question_id.startswith("q-")
|
||
assert len(q.options) == 2
|
||
assert q.options[0].badge == "12 AM–9 AM"
|
||
assert not q.answered
|
||
|
||
assert qm.list_pending() == [q]
|
||
assert qm.answer_question(q.question_id, "morning") is True
|
||
assert q.answered is True
|
||
assert q.answer == "morning"
|
||
assert qm.get_pending(q.question_id) is None # cleaned up, not leaked
|
||
assert qm.answer_question(q.question_id, "afternoon") is False # double-answer rejected
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_question_manager_wait_and_answer():
|
||
qm = QuestionManager()
|
||
q = qm.create_question(prompt="Reassign?", options=["yes", "no"], allow_custom_input=False)
|
||
|
||
async def operator():
|
||
await asyncio.sleep(0.01)
|
||
qm.answer_question(q.question_id, "yes")
|
||
|
||
asyncio.create_task(operator())
|
||
assert await qm.wait_for_answer(q, timeout_s=1.0) == "yes"
|
||
assert qm.get_pending(q.question_id) is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_question_manager_timeout_cleans_up():
|
||
qm = QuestionManager()
|
||
q = qm.create_question(prompt="Reassign?", options=["yes", "no"])
|
||
with pytest.raises(TimeoutError):
|
||
await qm.wait_for_answer(q, timeout_s=0.01)
|
||
assert qm.get_pending(q.question_id) is None
|