Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into fix/skills-yaml-frontmatter

Browse Source
This commit is contained in:
chengyongru
2026-04-15 16:56:23 +08:00
committed by GitHub
parent a1b544fd23 6829b8b475
commit 015833e34b
44 changed files with 3121 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/agent/test_context_prompt_cache.py
+52
View File
@@ -219,3 +219,55 @@ def test_subagent_result_does_not_create_consecutive_assistant_messages(tmp_path
for left, right in zip(messages, messages[1:]):
assert not (left.get("role") == right.get("role") == "assistant")
def test_always_skills_excluded_from_skills_index(tmp_path) -> None:
"""Always skills should appear in Active Skills but NOT in the skills index."""
workspace = _make_workspace(tmp_path)
builder = ContextBuilder(workspace)
prompt = builder.build_system_prompt()
# memory skill should be in Active Skills section
assert "# Active Skills" in prompt
assert "### Skill: memory" in prompt
# memory skill should NOT appear in the skills index
skills_section = prompt.split("# Skills\n", 1)
if len(skills_section) > 1:
index_text = skills_section[1].split("\n\n---")[0]
assert "**memory**" not in index_text
def test_template_memory_md_is_skipped(tmp_path) -> None:
"""MEMORY.md matching the bundled template should not inject the Memory section."""
workspace = _make_workspace(tmp_path)
from nanobot.utils.helpers import sync_workspace_templates
sync_workspace_templates(workspace, silent=True)
builder = ContextBuilder(workspace)
prompt = builder.build_system_prompt()
# The "# Memory\n\n## Long-term Memory" block is produced only by
# build_system_prompt() when MEMORY.md is injected. The memory skill
# also contains "# Memory" but is followed by "## Structure", not
# "## Long-term Memory".
assert "# Memory\n\n## Long-term Memory" not in prompt
assert "This file is automatically updated by nanobot" not in prompt
def test_customized_memory_md_is_injected(tmp_path) -> None:
"""A Dream-populated MEMORY.md should be injected normally."""
workspace = _make_workspace(tmp_path)
from nanobot.utils.helpers import sync_workspace_templates
sync_workspace_templates(workspace, silent=True)
(workspace / "memory" / "MEMORY.md").write_text(
"# Long-term Memory\n\nUser prefers dark mode.\n", encoding="utf-8"
)
builder = ContextBuilder(workspace)
prompt = builder.build_system_prompt()
assert "# Memory\n\n## Long-term Memory" in prompt
assert "User prefers dark mode" in prompt
tests/agent/test_loop_save_turn.py
+109
View File
@@ -1,3 +1,4 @@
import asyncio
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock
@@ -308,3 +309,111 @@ async def test_next_turn_after_crash_closes_pending_user_turn_before_new_input(t
{"role": "assistant", "content": "new answer"},
]
assert AgentLoop._PENDING_USER_TURN_KEY not in session.metadata
@pytest.mark.asyncio
async def test_stop_preserves_runtime_checkpoint_for_next_turn(tmp_path: Path) -> None:
from nanobot.command.builtin import cmd_stop
from nanobot.command.router import CommandContext
loop = _make_full_loop(tmp_path)
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign]
checkpoint_saved = asyncio.Event()
async def interrupted_run_agent_loop(_initial_messages, *, session=None, **_kwargs):
assert session is not None
loop._set_runtime_checkpoint(
session,
{
"assistant_message": {
"role": "assistant",
"content": "working",
"tool_calls": [
{
"id": "call_done",
"type": "function",
"function": {"name": "read_file", "arguments": "{}"},
},
{
"id": "call_pending",
"type": "function",
"function": {"name": "exec", "arguments": "{}"},
},
],
},
"completed_tool_results": [
{
"role": "tool",
"tool_call_id": "call_done",
"name": "read_file",
"content": "ok",
}
],
"pending_tool_calls": [
{
"id": "call_pending",
"type": "function",
"function": {"name": "exec", "arguments": "{}"},
}
],
},
)
checkpoint_saved.set()
await asyncio.Event().wait()
loop._run_agent_loop = interrupted_run_agent_loop # type: ignore[method-assign]
first_msg = InboundMessage(channel="feishu", sender_id="u1", chat_id="c4", content="keep progress")
task = asyncio.create_task(loop._process_message(first_msg))
loop._active_tasks[first_msg.session_key] = [task]
await asyncio.wait_for(checkpoint_saved.wait(), timeout=1.0)
stop_msg = InboundMessage(channel="feishu", sender_id="u1", chat_id="c4", content="/stop")
stop_ctx = CommandContext(msg=stop_msg, session=None, key=stop_msg.session_key, raw="/stop", loop=loop)
stop_result = await cmd_stop(stop_ctx)
assert "Stopped 1 task" in stop_result.content
assert task.done()
loop.sessions.invalidate("feishu:c4")
interrupted = loop.sessions.get_or_create("feishu:c4")
assert interrupted.metadata.get(AgentLoop._PENDING_USER_TURN_KEY) is True
assert interrupted.metadata.get(AgentLoop._RUNTIME_CHECKPOINT_KEY) is not None
async def resumed_run_agent_loop(initial_messages, **_kwargs):
return (
"next answer",
None,
[*initial_messages, {"role": "assistant", "content": "next answer"}],
"stop",
False,
)
loop._run_agent_loop = resumed_run_agent_loop # type: ignore[method-assign]
result = await loop._process_message(
InboundMessage(channel="feishu", sender_id="u1", chat_id="c4", content="continue here")
)
assert result is not None
assert result.content == "next answer"
session = loop.sessions.get_or_create("feishu:c4")
assert [
{k: v for k, v in m.items() if k in {"role", "content", "tool_call_id", "name"}}
for m in session.messages
] == [
{"role": "user", "content": "keep progress"},
{"role": "assistant", "content": "working"},
{"role": "tool", "tool_call_id": "call_done", "name": "read_file", "content": "ok"},
{
"role": "tool",
"tool_call_id": "call_pending",
"name": "exec",
"content": "Error: Task interrupted before this tool finished.",
},
{"role": "user", "content": "continue here"},
{"role": "assistant", "content": "next answer"},
]
assert AgentLoop._PENDING_USER_TURN_KEY not in session.metadata
assert AgentLoop._RUNTIME_CHECKPOINT_KEY not in session.metadata
tests/agent/test_runner.py
+396 -19
View File
@@ -18,6 +18,16 @@ from nanobot.providers.base import LLMResponse, ToolCallRequest
_MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars
def _make_injection_callback(queue: asyncio.Queue):
"""Return an async callback that drains *queue* into a list of dicts."""
async def inject_cb():
items = []
while not queue.empty():
items.append(await queue.get())
return items
return inject_cb
def _make_loop(tmp_path):
from nanobot.agent.loop import AgentLoop
from nanobot.bus.queue import MessageBus
@@ -679,11 +689,20 @@ async def test_runner_keeps_going_when_tool_result_persistence_fails():
class _DelayTool(Tool):
def __init__(self, name: str, *, delay: float, read_only: bool, shared_events: list[str]):
def __init__(
self,
name: str,
*,
delay: float,
read_only: bool,
shared_events: list[str],
exclusive: bool = False,
):
self._name = name
self._delay = delay
self._read_only = read_only
self._shared_events = shared_events
self._exclusive = exclusive
@property
def name(self) -> str:
@@ -701,6 +720,10 @@ class _DelayTool(Tool):
def read_only(self) -> bool:
return self._read_only
@property
def exclusive(self) -> bool:
return self._exclusive
async def execute(self, **kwargs):
self._shared_events.append(f"start:{self._name}")
await asyncio.sleep(self._delay)
@@ -746,6 +769,48 @@ async def test_runner_batches_read_only_tools_before_exclusive_work():
assert shared_events[-2:] == ["start:write_a", "end:write_a"]
@pytest.mark.asyncio
async def test_runner_does_not_batch_exclusive_read_only_tools():
from nanobot.agent.runner import AgentRunSpec, AgentRunner
tools = ToolRegistry()
shared_events: list[str] = []
read_a = _DelayTool("read_a", delay=0.03, read_only=True, shared_events=shared_events)
read_b = _DelayTool("read_b", delay=0.03, read_only=True, shared_events=shared_events)
ddg_like = _DelayTool(
"ddg_like",
delay=0.01,
read_only=True,
shared_events=shared_events,
exclusive=True,
)
tools.register(read_a)
tools.register(ddg_like)
tools.register(read_b)
runner = AgentRunner(MagicMock())
await runner._execute_tools(
AgentRunSpec(
initial_messages=[],
tools=tools,
model="test-model",
max_iterations=1,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
concurrent_tools=True,
),
[
ToolCallRequest(id="ro1", name="read_a", arguments={}),
ToolCallRequest(id="ddg1", name="ddg_like", arguments={}),
ToolCallRequest(id="ro2", name="read_b", arguments={}),
],
{},
)
assert shared_events[0] == "start:read_a"
assert shared_events.index("end:read_a") < shared_events.index("start:ddg_like")
assert shared_events.index("end:ddg_like") < shared_events.index("start:read_b")
@pytest.mark.asyncio
async def test_runner_blocks_repeated_external_fetches():
from nanobot.agent.runner import AgentRunSpec, AgentRunner
@@ -1888,12 +1953,7 @@ async def test_checkpoint1_injects_after_tool_execution():
tools.execute = AsyncMock(return_value="file content")
injection_queue = asyncio.Queue()
async def inject_cb():
items = []
while not injection_queue.empty():
items.append(await injection_queue.get())
return items
inject_cb = _make_injection_callback(injection_queue)
# Put a follow-up message in the queue before the run starts
await injection_queue.put(
@@ -1951,12 +2011,7 @@ async def test_checkpoint2_injects_after_final_response_with_resuming_stream():
tools.get_definitions.return_value = []
injection_queue = asyncio.Queue()
async def inject_cb():
items = []
while not injection_queue.empty():
items.append(await injection_queue.get())
return items
inject_cb = _make_injection_callback(injection_queue)
# Inject a follow-up that arrives during the first response
await injection_queue.put(
@@ -2005,12 +2060,7 @@ async def test_checkpoint2_preserves_final_response_in_history_before_followup()
tools.get_definitions.return_value = []
injection_queue = asyncio.Queue()
async def inject_cb():
items = []
while not injection_queue.empty():
items.append(await injection_queue.get())
return items
inject_cb = _make_injection_callback(injection_queue)
await injection_queue.put(
InboundMessage(channel="cli", sender_id="u", chat_id="c", content="follow-up question")
@@ -2410,3 +2460,330 @@ async def test_dispatch_republishes_leftover_queue_messages(tmp_path):
contents = [m.content for m in msgs]
assert "leftover-1" in contents
assert "leftover-2" in contents
@pytest.mark.asyncio
async def test_drain_injections_on_fatal_tool_error():
"""Pending injections should be drained even when a fatal tool error occurs."""
from nanobot.agent.runner import AgentRunSpec, AgentRunner
from nanobot.bus.events import InboundMessage
provider = MagicMock()
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
if call_count["n"] == 1:
return LLMResponse(
content="",
tool_calls=[ToolCallRequest(id="c1", name="exec", arguments={"cmd": "bad"})],
usage={},
)
# Second call: respond normally to the injected follow-up
return LLMResponse(content="reply to follow-up", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(side_effect=RuntimeError("tool exploded"))
injection_queue = asyncio.Queue()
inject_cb = _make_injection_callback(injection_queue)
await injection_queue.put(
InboundMessage(channel="cli", sender_id="u", chat_id="c", content="follow-up after error")
)
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[{"role": "user", "content": "hello"}],
tools=tools,
model="test-model",
max_iterations=5,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
fail_on_tool_error=True,
injection_callback=inject_cb,
))
assert result.had_injections is True
assert result.final_content == "reply to follow-up"
# The injection should be in the messages history
injected = [
m for m in result.messages
if m.get("role") == "user" and m.get("content") == "follow-up after error"
]
assert len(injected) == 1
@pytest.mark.asyncio
async def test_drain_injections_on_llm_error():
"""Pending injections should be drained when the LLM returns an error finish_reason."""
from nanobot.agent.runner import AgentRunSpec, AgentRunner
from nanobot.bus.events import InboundMessage
provider = MagicMock()
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
if call_count["n"] == 1:
return LLMResponse(
content=None,
tool_calls=[],
finish_reason="error",
usage={},
)
# Second call: respond normally to the injected follow-up
return LLMResponse(content="recovered answer", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
injection_queue = asyncio.Queue()
inject_cb = _make_injection_callback(injection_queue)
await injection_queue.put(
InboundMessage(channel="cli", sender_id="u", chat_id="c", content="follow-up after LLM error")
)
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[
{"role": "user", "content": "hello"},
{"role": "assistant", "content": "previous response"},
{"role": "user", "content": "trigger error"},
],
tools=tools,
model="test-model",
max_iterations=5,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
injection_callback=inject_cb,
))
assert result.had_injections is True
assert result.final_content == "recovered answer"
injected = [
m for m in result.messages
if m.get("role") == "user" and "follow-up after LLM error" in str(m.get("content", ""))
]
assert len(injected) == 1
@pytest.mark.asyncio
async def test_drain_injections_on_empty_final_response():
"""Pending injections should be drained when the runner exits due to empty response."""
from nanobot.agent.runner import AgentRunSpec, AgentRunner, _MAX_EMPTY_RETRIES
from nanobot.bus.events import InboundMessage
provider = MagicMock()
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
if call_count["n"] <= _MAX_EMPTY_RETRIES + 1:
return LLMResponse(content="", tool_calls=[], usage={})
# After retries exhausted + injection drain, respond normally
return LLMResponse(content="answer after empty", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
injection_queue = asyncio.Queue()
inject_cb = _make_injection_callback(injection_queue)
await injection_queue.put(
InboundMessage(channel="cli", sender_id="u", chat_id="c", content="follow-up after empty")
)
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[
{"role": "user", "content": "hello"},
{"role": "assistant", "content": "previous response"},
{"role": "user", "content": "trigger empty"},
],
tools=tools,
model="test-model",
max_iterations=10,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
injection_callback=inject_cb,
))
assert result.had_injections is True
assert result.final_content == "answer after empty"
injected = [
m for m in result.messages
if m.get("role") == "user" and "follow-up after empty" in str(m.get("content", ""))
]
assert len(injected) == 1
@pytest.mark.asyncio
async def test_drain_injections_on_max_iterations():
"""Pending injections should be drained when the runner hits max_iterations.
Unlike other error paths, max_iterations cannot continue the loop, so
injections are appended to messages but not processed by the LLM.
The key point is they are consumed from the queue to prevent re-publish.
"""
from nanobot.agent.runner import AgentRunSpec, AgentRunner
from nanobot.bus.events import InboundMessage
provider = MagicMock()
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
return LLMResponse(
content="",
tool_calls=[ToolCallRequest(id=f"c{call_count['n']}", name="read_file", arguments={"path": "x"})],
usage={},
)
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="file content")
injection_queue = asyncio.Queue()
inject_cb = _make_injection_callback(injection_queue)
await injection_queue.put(
InboundMessage(channel="cli", sender_id="u", chat_id="c", content="follow-up after max iters")
)
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[{"role": "user", "content": "hello"}],
tools=tools,
model="test-model",
max_iterations=2,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
injection_callback=inject_cb,
))
assert result.stop_reason == "max_iterations"
assert result.had_injections is True
# The injection was consumed from the queue (preventing re-publish)
assert injection_queue.empty()
# The injection message is appended to conversation history
injected = [
m for m in result.messages
if m.get("role") == "user" and m.get("content") == "follow-up after max iters"
]
assert len(injected) == 1
@pytest.mark.asyncio
async def test_drain_injections_set_flag_when_followup_arrives_after_last_iteration():
"""Late follow-ups drained in max_iterations should still flip had_injections."""
from nanobot.agent.hook import AgentHook
from nanobot.agent.runner import AgentRunSpec, AgentRunner
from nanobot.bus.events import InboundMessage
provider = MagicMock()
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
return LLMResponse(
content="",
tool_calls=[ToolCallRequest(id=f"c{call_count['n']}", name="read_file", arguments={"path": "x"})],
usage={},
)
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="file content")
injection_queue = asyncio.Queue()
inject_cb = _make_injection_callback(injection_queue)
class InjectOnLastAfterIterationHook(AgentHook):
def __init__(self) -> None:
self.after_iteration_calls = 0
async def after_iteration(self, context) -> None:
self.after_iteration_calls += 1
if self.after_iteration_calls == 2:
await injection_queue.put(
InboundMessage(
channel="cli",
sender_id="u",
chat_id="c",
content="late follow-up after max iters",
)
)
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[{"role": "user", "content": "hello"}],
tools=tools,
model="test-model",
max_iterations=2,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
injection_callback=inject_cb,
hook=InjectOnLastAfterIterationHook(),
))
assert result.stop_reason == "max_iterations"
assert result.had_injections is True
assert injection_queue.empty()
injected = [
m for m in result.messages
if m.get("role") == "user" and m.get("content") == "late follow-up after max iters"
]
assert len(injected) == 1
@pytest.mark.asyncio
async def test_injection_cycle_cap_on_error_path():
"""Injection cycles should be capped even when every iteration hits an LLM error."""
from nanobot.agent.runner import AgentRunSpec, AgentRunner, _MAX_INJECTION_CYCLES
from nanobot.bus.events import InboundMessage
provider = MagicMock()
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
return LLMResponse(
content=None,
tool_calls=[],
finish_reason="error",
usage={},
)
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
drain_count = {"n": 0}
async def inject_cb():
drain_count["n"] += 1
if drain_count["n"] <= _MAX_INJECTION_CYCLES:
return [InboundMessage(channel="cli", sender_id="u", chat_id="c", content=f"msg-{drain_count['n']}")]
return []
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[
{"role": "user", "content": "hello"},
{"role": "assistant", "content": "previous"},
{"role": "user", "content": "trigger error"},
],
tools=tools,
model="test-model",
max_iterations=20,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
injection_callback=inject_cb,
))
assert result.had_injections is True
# Should cap: _MAX_INJECTION_CYCLES drained rounds + 1 final round that breaks
assert call_count["n"] == _MAX_INJECTION_CYCLES + 1
assert drain_count["n"] == _MAX_INJECTION_CYCLES
tests/channels/test_base_channel.py
+12
View File
@@ -23,3 +23,15 @@ def test_is_allowed_requires_exact_match() -> None:
assert channel.is_allowed("allow@email.com") is True
assert channel.is_allowed("attacker|allow@email.com") is False
def test_is_allowed_supports_dict_allow_from_alias() -> None:
channel = _DummyChannel({"allowFrom": ["alice"]}, MessageBus())
assert channel.is_allowed("alice") is True
def test_is_allowed_denies_empty_dict_allow_from() -> None:
channel = _DummyChannel({"allow_from": []}, MessageBus())
assert channel.is_allowed("alice") is False
tests/channels/test_channel_plugins.py
+23 -1
View File
@@ -646,7 +646,10 @@ class _ChannelWithAllowFrom(BaseChannel):
def __init__(self, config, bus, allow_from):
super().__init__(config, bus)
self.config.allow_from = allow_from
if isinstance(self.config, dict):
self.config["allow_from"] = allow_from
else:
self.config.allow_from = allow_from
async def start(self) -> None:
pass
@@ -714,6 +717,25 @@ async def test_validate_allow_from_passes_with_asterisk():
mgr._validate_allow_from()
@pytest.mark.asyncio
async def test_validate_allow_from_raises_on_empty_dict_allow_from():
"""_validate_allow_from should reject empty dict-backed allow_from lists."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.channels = {"test": _ChannelWithAllowFrom({"enabled": True}, None, [])}
mgr._dispatch_task = None
with pytest.raises(SystemExit) as exc_info:
mgr._validate_allow_from()
assert "empty allowFrom" in str(exc_info.value)
@pytest.mark.asyncio
async def test_get_channel_returns_channel_if_exists():
"""get_channel should return the channel if it exists."""
tests/channels/test_feishu_streaming.py
+9 -56
View File
@@ -205,53 +205,22 @@ class TestSendDelta:
ch._client.im.v1.message.create.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_resuming_keeps_buffer(self):
"""_resuming=True flushes text to card but keeps the buffer for the next segment."""
async def test_stream_end_fallback_when_final_update_fails(self):
"""If streaming mode was closed (e.g. Feishu timeout), fall back to a regular card."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Partial answer", card_id="card_1", sequence=2, last_edit=0.0,
text="Lost content", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(success=False)
ch._client.im.v1.message.create.return_value = _mock_send_response("om_fb")
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True, "_resuming": True})
assert "oc_chat1" in ch._stream_bufs
buf = ch._stream_bufs["oc_chat1"]
assert buf.card_id == "card_1"
assert buf.sequence == 3
ch._client.cardkit.v1.card_element.content.assert_called_once()
ch._client.cardkit.v1.card.settings.assert_not_called()
@pytest.mark.asyncio
async def test_stream_end_resuming_then_final_end(self):
"""Full multi-segment flow: resuming mid-turn, then final end closes the card."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Seg1", card_id="card_1", sequence=1, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True, "_resuming": True})
assert "oc_chat1" in ch._stream_bufs
ch._stream_bufs["oc_chat1"].text += " Seg2"
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True})
assert "oc_chat1" not in ch._stream_bufs
ch._client.cardkit.v1.card.settings.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_resuming_no_card_is_noop(self):
"""_resuming with no card_id (card creation failed) is a safe no-op."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="text", card_id=None, sequence=0, last_edit=0.0,
)
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True, "_resuming": True})
assert "oc_chat1" in ch._stream_bufs
ch._client.cardkit.v1.card_element.content.assert_not_called()
# Should NOT attempt to close streaming mode since update failed
ch._client.cardkit.v1.card.settings.assert_not_called()
# Should fall back to sending a regular interactive card
ch._client.im.v1.message.create.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_without_buf_is_noop(self):
@@ -375,22 +344,6 @@ class TestToolHintInlineStreaming:
assert "🔧 $ cd /project" in buf.text
assert "🔧 $ git status" in buf.text
@pytest.mark.asyncio
async def test_tool_hint_preserved_on_resuming_flush(self):
"""When _resuming flushes the buffer, tool hint is kept as permanent content."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Partial answer\n\n🔧 $ cd /project\n\n",
card_id="card_1", sequence=2, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True, "_resuming": True})
buf = ch._stream_bufs["oc_chat1"]
assert "Partial answer" in buf.text
assert "🔧 $ cd /project" in buf.text
@pytest.mark.asyncio
async def test_tool_hint_preserved_on_final_stream_end(self):
"""When final _stream_end closes the card, tool hint is kept in the final text."""
tests/channels/test_feishu_tool_hint_code_block.py
+40 -47
View File
@@ -1,6 +1,7 @@
"""Tests for FeishuChannel tool hint code block formatting."""
"""Tests for FeishuChannel tool hint formatting."""
import json
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
@@ -28,15 +29,24 @@ def mock_feishu_channel():
config.app_secret = "test_app_secret"
config.encrypt_key = None
config.verification_token = None
config.tool_hint_prefix = "\U0001f527" # 🔧
bus = MagicMock()
channel = FeishuChannel(config, bus)
channel._client = MagicMock() # Simulate initialized client
channel._client = MagicMock()
return channel
def _get_tool_hint_card(mock_send):
"""Extract the interactive card from _send_message_sync calls."""
call_args = mock_send.call_args[0]
_, _, msg_type, content = call_args
assert msg_type == "interactive"
return json.loads(content)
@mark.asyncio
async def test_tool_hint_sends_code_message(mock_feishu_channel):
"""Tool hint messages should be sent as interactive cards with code blocks."""
async def test_tool_hint_sends_interactive_card(mock_feishu_channel):
"""Tool hint without active buffer sends an interactive card with 🔧 style."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
@@ -47,23 +57,12 @@ async def test_tool_hint_sends_code_message(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
# Verify interactive message with card was sent
assert mock_send.call_count == 1
call_args = mock_send.call_args[0]
receive_id_type, receive_id, msg_type, content = call_args
assert receive_id_type == "chat_id"
assert receive_id == "oc_123456"
assert msg_type == "interactive"
# Parse content to verify card structure
card = json.loads(content)
card = _get_tool_hint_card(mock_send)
assert card["config"]["wide_screen_mode"] is True
assert len(card["elements"]) == 1
assert card["elements"][0]["tag"] == "markdown"
# Check that code block is properly formatted with language hint
expected_md = "**Tool Calls**\n\n```text\nweb_search(\"test query\")\n```"
assert card["elements"][0]["content"] == expected_md
md = card["elements"][0]["content"]
assert "\U0001f527" in md
assert "web_search" in md
@mark.asyncio
@@ -78,8 +77,6 @@ async def test_tool_hint_empty_content_does_not_send(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
# Should not send any message
mock_send.assert_not_called()
@@ -96,7 +93,6 @@ async def test_tool_hint_without_metadata_sends_as_normal(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
# Should send as text message (detected format)
assert mock_send.call_count == 1
call_args = mock_send.call_args[0]
_, _, msg_type, content = call_args
@@ -106,7 +102,7 @@ async def test_tool_hint_without_metadata_sends_as_normal(mock_feishu_channel):
@mark.asyncio
async def test_tool_hint_multiple_tools_in_one_message(mock_feishu_channel):
"""Multiple tool calls should be displayed each on its own line in a code block."""
"""Multiple tool calls should each get the 🔧 prefix."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
@@ -117,13 +113,11 @@ async def test_tool_hint_multiple_tools_in_one_message(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
call_args = mock_send.call_args[0]
msg_type = call_args[2]
content = json.loads(call_args[3])
assert msg_type == "interactive"
# Each tool call should be on its own line
expected_md = "**Tool Calls**\n\n```text\nweb_search(\"query\"),\nread_file(\"/path/to/file\")\n```"
assert content["elements"][0]["content"] == expected_md
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "web_search" in md
assert "read_file" in md
assert "\U0001f527" in md
@mark.asyncio
@@ -139,8 +133,8 @@ async def test_tool_hint_new_format_basic(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
content = json.loads(mock_send.call_args[0][3])
md = content["elements"][0]["content"]
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "read src/main.py" in md
assert 'grep "TODO"' in md
@@ -158,16 +152,15 @@ async def test_tool_hint_new_format_with_comma_in_quotes(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
content = json.loads(mock_send.call_args[0][3])
md = content["elements"][0]["content"]
# The comma inside quotes should NOT cause a line break
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert 'grep "hello, world"' in md
assert "$ echo test" in md
@mark.asyncio
async def test_tool_hint_new_format_with_folding(mock_feishu_channel):
"""Folded calls (× N) should display on separate lines."""
"""Folded calls (× N) should display correctly."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
@@ -178,8 +171,8 @@ async def test_tool_hint_new_format_with_folding(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
content = json.loads(mock_send.call_args[0][3])
md = content["elements"][0]["content"]
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "\u00d7 3" in md
assert 'grep "pattern"' in md
@@ -197,9 +190,12 @@ async def test_tool_hint_new_format_mcp(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
content = json.loads(mock_send.call_args[0][3])
md = content["elements"][0]["content"]
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "4_5v::analyze_image" in md
@mark.asyncio
async def test_tool_hint_keeps_commas_inside_arguments(mock_feishu_channel):
"""Commas inside a single tool argument must not be split onto a new line."""
msg = OutboundMessage(
@@ -212,10 +208,7 @@ async def test_tool_hint_keeps_commas_inside_arguments(mock_feishu_channel):
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
content = json.loads(mock_send.call_args[0][3])
expected_md = (
"**Tool Calls**\n\n```text\n"
"web_search(\"foo, bar\"),\n"
"read_file(\"/path/to/file\")\n```"
)
assert content["elements"][0]["content"] == expected_md
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert 'web_search("foo, bar")' in md
assert 'read_file("/path/to/file")' in md
tests/channels/test_slack_channel.py
+167 -2
View File
@@ -10,8 +10,7 @@ except ImportError:
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.slack import SlackChannel
from nanobot.channels.slack import SlackConfig
from nanobot.channels.slack import SlackChannel, SlackConfig
class _FakeAsyncWebClient:
@@ -20,6 +19,12 @@ class _FakeAsyncWebClient:
self.file_upload_calls: list[dict[str, object | None]] = []
self.reactions_add_calls: list[dict[str, object | None]] = []
self.reactions_remove_calls: list[dict[str, object | None]] = []
self.conversations_list_calls: list[dict[str, object | None]] = []
self.users_list_calls: list[dict[str, object | None]] = []
self.conversations_open_calls: list[dict[str, object | None]] = []
self._conversations_pages: list[dict[str, object]] = []
self._users_pages: list[dict[str, object]] = []
self._open_dm_response: dict[str, object] = {"channel": {"id": "D_OPENED"}}
async def chat_postMessage(
self,
@@ -81,6 +86,22 @@ class _FakeAsyncWebClient:
}
)
async def conversations_list(self, **kwargs):
self.conversations_list_calls.append(kwargs)
if self._conversations_pages:
return self._conversations_pages.pop(0)
return {"channels": [], "response_metadata": {"next_cursor": ""}}
async def users_list(self, **kwargs):
self.users_list_calls.append(kwargs)
if self._users_pages:
return self._users_pages.pop(0)
return {"members": [], "response_metadata": {"next_cursor": ""}}
async def conversations_open(self, **kwargs):
self.conversations_open_calls.append(kwargs)
return self._open_dm_response
@pytest.mark.asyncio
async def test_send_uses_thread_for_channel_messages() -> None:
@@ -151,3 +172,147 @@ async def test_send_updates_reaction_when_final_response_sent() -> None:
assert fake_web.reactions_add_calls == [
{"channel": "C123", "name": "white_check_mark", "timestamp": "1700000000.000100"}
]
@pytest.mark.asyncio
async def test_send_resolves_channel_name_to_channel_id() -> None:
channel = SlackChannel(SlackConfig(enabled=True), MessageBus())
fake_web = _FakeAsyncWebClient()
fake_web._conversations_pages = [
{
"channels": [{"id": "C999", "name": "channel_x"}],
"response_metadata": {"next_cursor": ""},
}
]
channel._web_client = fake_web
await channel.send(
OutboundMessage(
channel="slack",
chat_id="#channel_x",
content="hello",
)
)
assert fake_web.chat_post_calls == [
{"channel": "C999", "text": "hello\n", "thread_ts": None}
]
assert len(fake_web.conversations_list_calls) == 1
@pytest.mark.asyncio
async def test_send_resolves_user_handle_to_dm_channel() -> None:
channel = SlackChannel(SlackConfig(enabled=True), MessageBus())
fake_web = _FakeAsyncWebClient()
fake_web._users_pages = [
{
"members": [
{
"id": "U234",
"name": "alice",
"profile": {"display_name": "Alice"},
}
],
"response_metadata": {"next_cursor": ""},
}
]
fake_web._open_dm_response = {"channel": {"id": "D234"}}
channel._web_client = fake_web
await channel.send(
OutboundMessage(
channel="slack",
chat_id="@alice",
content="hello",
)
)
assert fake_web.conversations_open_calls == [{"users": "U234"}]
assert fake_web.chat_post_calls == [
{"channel": "D234", "text": "hello\n", "thread_ts": None}
]
@pytest.mark.asyncio
async def test_send_updates_reaction_on_origin_channel_for_cross_channel_send() -> None:
channel = SlackChannel(SlackConfig(enabled=True, react_emoji="eyes"), MessageBus())
fake_web = _FakeAsyncWebClient()
fake_web._conversations_pages = [
{
"channels": [{"id": "C999", "name": "channel_x"}],
"response_metadata": {"next_cursor": ""},
}
]
channel._web_client = fake_web
await channel.send(
OutboundMessage(
channel="slack",
chat_id="channel_x",
content="done",
metadata={
"slack": {
"event": {"ts": "1700000000.000100", "channel": "D_ORIGIN"},
"channel_type": "im",
},
},
)
)
assert fake_web.chat_post_calls == [
{"channel": "C999", "text": "done\n", "thread_ts": None}
]
assert fake_web.reactions_remove_calls == [
{"channel": "D_ORIGIN", "name": "eyes", "timestamp": "1700000000.000100"}
]
assert fake_web.reactions_add_calls == [
{"channel": "D_ORIGIN", "name": "white_check_mark", "timestamp": "1700000000.000100"}
]
@pytest.mark.asyncio
async def test_send_does_not_reuse_origin_thread_ts_for_cross_channel_send() -> None:
channel = SlackChannel(SlackConfig(enabled=True), MessageBus())
fake_web = _FakeAsyncWebClient()
fake_web._conversations_pages = [
{
"channels": [{"id": "C999", "name": "channel_x"}],
"response_metadata": {"next_cursor": ""},
}
]
channel._web_client = fake_web
await channel.send(
OutboundMessage(
channel="slack",
chat_id="channel_x",
content="done",
metadata={
"slack": {
"event": {"ts": "1700000000.000100", "channel": "C_ORIGIN"},
"thread_ts": "1700000000.000200",
"channel_type": "channel",
},
},
)
)
assert fake_web.chat_post_calls == [
{"channel": "C999", "text": "done\n", "thread_ts": None}
]
@pytest.mark.asyncio
async def test_send_raises_when_named_target_cannot_be_resolved() -> None:
channel = SlackChannel(SlackConfig(enabled=True), MessageBus())
fake_web = _FakeAsyncWebClient()
channel._web_client = fake_web
with pytest.raises(ValueError, match="was not found"):
await channel.send(
OutboundMessage(
channel="slack",
chat_id="#missing-channel",
content="hello",
)
)
tests/channels/test_wecom_channel.py
+44
View File
@@ -541,6 +541,50 @@ async def test_process_voice_message() -> None:
assert "[voice]" in msg.content
@pytest.mark.asyncio
async def test_process_mixed_message() -> None:
"""Mixed message: contains picture and text message types."""
channel = WecomChannel(WecomConfig(bot_id="b", secret="s", allow_from=["user1"]), MessageBus())
client = _FakeWeComClient()
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
f.write(b"\x89PNG\r\n")
saved = f.name
client.download_file.return_value = (b"\x89PNG\r\n", "photo.png")
channel._client = client
try:
with patch("nanobot.channels.wecom.get_media_dir", return_value=Path(os.path.dirname(saved))):
frame = _FakeFrame(body={
"msgid": "msg_mixed_1",
"chatid": "chat1",
"msgtype": "mixed",
"from": {"userid": "user1"},
"mixed": {
"msg_item": [
{"msgtype": "text", "text": {"content": "hello wecom"}},
{"msgtype": "image", "image": {"url": "https://example.com/img.png", "aeskey": "key123"}}
]
}
})
await channel._process_message(frame, "mixed")
msg = await channel.bus.consume_inbound()
assert msg.sender_id == "user1"
assert msg.chat_id == "chat1"
assert msg.content.startswith("hello wecom")
assert msg.metadata["msg_type"] == "mixed"
assert len(msg.media) == 1
assert msg.media[0].endswith("photo.png")
assert "[image:" in msg.content
finally:
# Clean up any photo.png in tempdir
p = os.path.join(os.path.dirname(saved), "photo.png")
if os.path.exists(p):
os.unlink(p)
@pytest.mark.asyncio
async def test_process_message_deduplication() -> None:
"""Same msg_id is not processed twice."""
tests/cli/test_commands.py
+147
View File
@@ -1126,6 +1126,153 @@ def test_gateway_cli_port_overrides_configured_port(monkeypatch, tmp_path: Path)
assert "port 18792" in result.stdout
def test_gateway_health_endpoint_binds_and_serves_expected_responses(
monkeypatch, tmp_path: Path
) -> None:
config_file = _write_instance_config(tmp_path)
config = Config()
config.gateway.port = 18791
captured: dict[str, object] = {}
class _FakeDream:
model = None
max_batch_size = 0
max_iterations = 0
async def run(self) -> None:
return None
class _FakeAgentLoop:
def __init__(self, **_kwargs) -> None:
self.model = "test-model"
self.dream = _FakeDream()
async def run(self) -> None:
await asyncio.Event().wait()
async def close_mcp(self) -> None:
return None
def stop(self) -> None:
return None
class _FakeChannelManager:
def __init__(self, _config, _bus) -> None:
self.enabled_channels = ["telegram", "discord"]
async def start_all(self) -> None:
await asyncio.Event().wait()
async def stop_all(self) -> None:
return None
class _FakeCronService:
def __init__(self, _store_path: Path) -> None:
self.on_job = None
async def start(self) -> None:
return None
def stop(self) -> None:
return None
def status(self) -> dict[str, int]:
return {"jobs": 0}
def register_system_job(self, _job) -> None:
return None
class _FakeHeartbeatService:
def __init__(self, **_kwargs) -> None:
return None
async def start(self) -> None:
return None
def stop(self) -> None:
return None
class _FakeServer:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb) -> bool:
return False
async def serve_forever(self) -> None:
raise _StopGatewayError("stop")
async def _fake_start_server(handler, host: str, port: int):
captured["handler"] = handler
captured["host"] = host
captured["port"] = port
return _FakeServer()
class _FakeReader:
def __init__(self, payload: bytes) -> None:
self.payload = payload
async def read(self, _size: int) -> bytes:
return self.payload
class _FakeWriter:
def __init__(self) -> None:
self.output = b""
self.closed = False
def write(self, data: bytes) -> None:
self.output += data
async def drain(self) -> None:
return None
def close(self) -> None:
self.closed = True
_patch_cli_command_runtime(
monkeypatch,
config,
message_bus=lambda: object(),
session_manager=lambda _workspace: object(),
)
monkeypatch.setattr("nanobot.agent.loop.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCronService)
monkeypatch.setattr("nanobot.heartbeat.service.HeartbeatService", _FakeHeartbeatService)
monkeypatch.setattr("asyncio.start_server", _fake_start_server)
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
assert result.exit_code == 0
assert captured["host"] == "127.0.0.1"
assert captured["port"] == 18791
assert "Health endpoint: http://127.0.0.1:18791/health" in result.stdout
def _call_handler(path: str) -> tuple[str, _FakeWriter]:
request = f"GET {path} HTTP/1.1\r\nHost: localhost\r\n\r\n".encode()
writer = _FakeWriter()
handler = captured["handler"]
assert callable(handler)
asyncio.run(handler(_FakeReader(request), writer))
return writer.output.decode(), writer
root_response, root_writer = _call_handler("/")
assert root_writer.closed is True
assert "HTTP/1.0 404 Not Found" in root_response
assert root_response.endswith("\r\n\r\nNot Found")
health_response, health_writer = _call_handler("/health")
assert health_writer.closed is True
assert "HTTP/1.0 200 OK" in health_response
health_body = json.loads(health_response.split("\r\n\r\n", 1)[1])
assert health_body == {"status": "ok"}
missing_response, missing_writer = _call_handler("/missing")
assert missing_writer.closed is True
assert "HTTP/1.0 404 Not Found" in missing_response
assert missing_response.endswith("\r\n\r\nNot Found")
def test_serve_uses_api_config_defaults_and_workspace_override(
monkeypatch, tmp_path: Path
) -> None:
tests/cli/test_restart_command.py
+29
View File
@@ -140,6 +140,7 @@ class TestRestartCommand:
loop.consolidator.estimate_session_prompt_tokens = MagicMock(
return_value=(20500, "tiktoken")
)
loop.subagents.get_running_count_by_session.return_value = 0
msg = InboundMessage(channel="telegram", sender_id="u1", chat_id="c1", content="/status")
@@ -151,8 +152,33 @@ class TestRestartCommand:
assert "Context: 20k/65k (31%)" in response.content
assert "Session: 3 messages" in response.content
assert "Uptime: 2m 5s" in response.content
assert "Tasks: 0 active" in response.content
assert response.metadata == {"render_as": "text"}
@pytest.mark.asyncio
async def test_status_counts_running_dispatch_and_subagent_tasks(self):
loop, _bus = _make_loop()
session = MagicMock()
session.get_history.return_value = [{"role": "user"}]
loop.sessions.get_or_create.return_value = session
loop.consolidator.estimate_session_prompt_tokens = MagicMock(
return_value=(1000, "tiktoken")
)
running_task = MagicMock()
running_task.done.return_value = False
finished_task = MagicMock()
finished_task.done.return_value = True
msg = InboundMessage(channel="telegram", sender_id="u1", chat_id="c1", content="/status")
loop._active_tasks[msg.session_key] = [running_task, finished_task]
loop.subagents.get_running_count_by_session.return_value = 2
response = await loop._process_message(msg)
assert response is not None
assert "Tasks: 3 active" in response.content
@pytest.mark.asyncio
async def test_run_agent_loop_resets_usage_when_provider_omits_it(self):
loop, _bus = _make_loop()
@@ -179,6 +205,7 @@ class TestRestartCommand:
loop.consolidator.estimate_session_prompt_tokens = MagicMock(
return_value=(0, "none")
)
loop.subagents.get_running_count_by_session.return_value = 0
response = await loop._process_message(
InboundMessage(channel="telegram", sender_id="u1", chat_id="c1", content="/status")
@@ -187,6 +214,7 @@ class TestRestartCommand:
assert response is not None
assert "Tokens: 1200 in / 34 out" in response.content
assert "Context: 1k/65k (1%)" in response.content
assert "Tasks: 0 active" in response.content
@pytest.mark.asyncio
async def test_process_direct_preserves_render_metadata(self):
@@ -195,6 +223,7 @@ class TestRestartCommand:
session.get_history.return_value = []
loop.sessions.get_or_create.return_value = session
loop.subagents.get_running_count.return_value = 0
loop.subagents.get_running_count_by_session.return_value = 0
response = await loop.process_direct("/status", session_key="cli:test")
tests/providers/test_litellm_kwargs.py
+119
View File
@@ -584,6 +584,78 @@ def test_openai_compat_keeps_tool_calls_after_consecutive_assistant_messages() -
assert sanitized[2]["tool_call_id"] == "3ec83c30d"
def test_openai_compat_stringifies_dict_tool_arguments() -> None:
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
provider = OpenAICompatProvider()
sanitized = provider._sanitize_messages([
{"role": "user", "content": "hi"},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "exec", "arguments": {"cmd": "ls -la"}},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "name": "exec", "content": "ok"},
{"role": "user", "content": "done"},
])
assert sanitized[1]["tool_calls"][0]["function"]["arguments"] == '{"cmd": "ls -la"}'
def test_openai_compat_repairs_non_json_tool_arguments_string() -> None:
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
provider = OpenAICompatProvider()
sanitized = provider._sanitize_messages([
{"role": "user", "content": "hi"},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "exec", "arguments": "{'cmd': 'pwd'}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "name": "exec", "content": "ok"},
{"role": "user", "content": "done"},
])
assert sanitized[1]["tool_calls"][0]["function"]["arguments"] == '{"cmd": "pwd"}'
def test_openai_compat_defaults_missing_tool_arguments_to_empty_object() -> None:
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
provider = OpenAICompatProvider()
sanitized = provider._sanitize_messages([
{"role": "user", "content": "hi"},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "exec"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "name": "exec", "content": "ok"},
{"role": "user", "content": "done"},
])
assert sanitized[1]["tool_calls"][0]["function"]["arguments"] == "{}"
@pytest.mark.asyncio
async def test_openai_compat_stream_watchdog_returns_error_on_stall(monkeypatch) -> None:
monkeypatch.setenv("NANOBOT_STREAM_IDLE_TIMEOUT_S", "0")
@@ -658,3 +730,50 @@ def test_openai_no_thinking_extra_body() -> None:
"""Non-thinking providers should never get extra_body for thinking."""
kw = _build_kwargs_for("openai", "gpt-4o", reasoning_effort="medium")
assert "extra_body" not in kw
def test_kimi_k25_thinking_enabled() -> None:
"""kimi-k2.5 with reasoning_effort set should opt in to thinking."""
kw = _build_kwargs_for("moonshot", "kimi-k2.5", reasoning_effort="medium")
assert kw.get("extra_body") == {"thinking": {"type": "enabled"}}
def test_kimi_k25_thinking_disabled_for_minimal() -> None:
"""reasoning_effort='minimal' maps to thinking disabled for kimi-k2.5."""
kw = _build_kwargs_for("moonshot", "kimi-k2.5", reasoning_effort="minimal")
assert kw.get("extra_body") == {"thinking": {"type": "disabled"}}
def test_kimi_k25_no_extra_body_when_reasoning_effort_none() -> None:
"""Without reasoning_effort the thinking param must not be injected."""
kw = _build_kwargs_for("moonshot", "kimi-k2.5", reasoning_effort=None)
assert "extra_body" not in kw
def test_kimi_k25_thinking_enabled_with_openrouter_prefix() -> None:
"""OpenRouter-style model names like moonshotai/kimi-k2.5 must trigger thinking."""
kw = _build_kwargs_for("openrouter", "moonshotai/kimi-k2.5", reasoning_effort="medium")
assert kw.get("extra_body") == {"thinking": {"type": "enabled"}}
def test_kimi_k25_thinking_disabled_with_openrouter_prefix() -> None:
"""OpenRouter names must NOT trigger thinking without reasoning_effort."""
kw = _build_kwargs_for("openrouter", "moonshotai/kimi-k2.5", reasoning_effort=None)
assert "extra_body" not in kw
def test_kimi_k26_code_preview_thinking_enabled() -> None:
"""k2.6-code-preview also supports thinking; should behave like k2.5."""
kw = _build_kwargs_for("moonshot", "k2.6-code-preview", reasoning_effort="high")
assert kw.get("extra_body") == {"thinking": {"type": "enabled"}}
def test_kimi_k2_series_no_thinking_injection() -> None:
"""kimi-k2 (non-thinking) models must NOT receive extra_body.thinking."""
kw = _build_kwargs_for("moonshot", "kimi-k2", reasoning_effort="high")
assert "extra_body" not in kw
def test_kimi_k2_thinking_series_no_thinking_injection() -> None:
"""kimi-k2-thinking series models must NOT receive extra_body.thinking."""
kw = _build_kwargs_for("moonshot", "kimi-k2-thinking", reasoning_effort="high")
assert "extra_body" not in kw
tests/providers/test_provider_retry.py
+52
View File
@@ -87,6 +87,33 @@ async def test_chat_with_retry_returns_final_error_after_retries(monkeypatch) ->
assert delays == [1, 2, 4]
@pytest.mark.asyncio
async def test_chat_with_retry_emits_terminal_progress_when_standard_retries_exhaust(monkeypatch) -> None:
provider = ScriptedProvider([
LLMResponse(content="429 rate limit a", finish_reason="error"),
LLMResponse(content="429 rate limit b", finish_reason="error"),
LLMResponse(content="429 rate limit c", finish_reason="error"),
LLMResponse(content="503 final server error", finish_reason="error"),
])
progress: list[str] = []
async def _fake_sleep(delay: int) -> None:
return None
async def _progress(msg: str) -> None:
progress.append(msg)
monkeypatch.setattr("nanobot.providers.base.asyncio.sleep", _fake_sleep)
response = await provider.chat_with_retry(
messages=[{"role": "user", "content": "hello"}],
on_retry_wait=_progress,
)
assert response.content == "503 final server error"
assert progress[-1] == "Model request failed after 4 retries, giving up."
@pytest.mark.asyncio
async def test_chat_with_retry_preserves_cancelled_error() -> None:
provider = ScriptedProvider([asyncio.CancelledError()])
@@ -469,3 +496,28 @@ async def test_persistent_retry_aborts_after_ten_identical_transient_errors(monk
assert response.content == "429 rate limit"
assert provider.calls == 10
assert delays == [1, 2, 4, 4, 4, 4, 4, 4, 4]
@pytest.mark.asyncio
async def test_persistent_retry_emits_terminal_progress_on_identical_error_limit(monkeypatch) -> None:
provider = ScriptedProvider([
*[LLMResponse(content="429 rate limit", finish_reason="error") for _ in range(10)],
])
progress: list[str] = []
async def _fake_sleep(delay: float) -> None:
return None
async def _progress(msg: str) -> None:
progress.append(msg)
monkeypatch.setattr("nanobot.providers.base.asyncio.sleep", _fake_sleep)
response = await provider.chat_with_retry(
messages=[{"role": "user", "content": "hello"}],
retry_mode="persistent",
on_retry_wait=_progress,
)
assert response.finish_reason == "error"
assert progress[-1] == "Persistent retry stopped after 10 identical errors."
tests/test_api_attachment.py
+496
View File
@@ -0,0 +1,496 @@
"""Tests for API file upload functionality (JSON base64 + multipart)."""
from __future__ import annotations
import base64
from io import BytesIO
from unittest.mock import AsyncMock, MagicMock
import pytest
import pytest_asyncio
from nanobot.api.server import (
_FileSizeExceeded,
_parse_json_content,
_save_base64_data_url,
create_app,
)
from nanobot.utils.document import extract_documents
try:
from aiohttp.test_utils import TestClient, TestServer
HAS_AIOHTTP = True
except ImportError:
HAS_AIOHTTP = False
pytest_plugins = ("pytest_asyncio",)
def _make_mock_agent(response_text: str = "mock response") -> MagicMock:
agent = MagicMock()
agent.process_direct = AsyncMock(return_value=response_text)
agent._connect_mcp = AsyncMock()
agent.close_mcp = AsyncMock()
return agent
@pytest.fixture
def mock_agent():
return _make_mock_agent()
@pytest.fixture
def app(mock_agent):
return create_app(mock_agent, model_name="test-model", request_timeout=10.0)
@pytest_asyncio.fixture
async def aiohttp_client():
clients: list[TestClient] = []
async def _make_client(app):
client = TestClient(TestServer(app))
await client.start_server()
clients.append(client)
return client
try:
yield _make_client
finally:
for client in clients:
await client.close()
# ---------------------------------------------------------------------------
# Helper function tests
# ---------------------------------------------------------------------------
def test_save_base64_data_url_saves_png(tmp_path) -> None:
"""Saving a base64 data URL creates a file with correct extension."""
b64_data = base64.b64encode(b"fake png data").decode()
data_url = f"data:image/png;base64,{b64_data}"
result = _save_base64_data_url(data_url, tmp_path)
assert result is not None
assert result.endswith(".png")
assert (tmp_path / result.replace(str(tmp_path) + "/", "")).read_bytes() == b"fake png data"
def test_save_base64_data_url_handles_invalid_b64(tmp_path) -> None:
"""Invalid base64 returns None."""
result = _save_base64_data_url("data:image/png;base64,not-valid-base64!!!", tmp_path)
assert result is None
def test_save_base64_data_url_handles_unknown_mime(tmp_path) -> None:
"""Unknown MIME type defaults to .bin."""
b64_data = base64.b64encode(b"some data").decode()
data_url = f"data:unknown/type;base64,{b64_data}"
result = _save_base64_data_url(data_url, tmp_path)
assert result is not None
assert result.endswith(".bin")
def test_save_base64_data_url_rejects_oversized_payload(tmp_path) -> None:
"""Base64 uploads should respect the same per-file limit as multipart."""
large_payload = base64.b64encode(b"x" * (11 * 1024 * 1024)).decode()
data_url = f"data:image/png;base64,{large_payload}"
with pytest.raises(_FileSizeExceeded, match="10MB limit"):
_save_base64_data_url(data_url, tmp_path)
def test_parse_json_content_extracts_text_and_media(tmp_path) -> None:
"""Parse JSON with text + base64 image saves image and returns paths."""
b64_data = base64.b64encode(b"img").decode()
body = {
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "describe this"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64_data}"}},
],
}
]
}
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
text, media_paths = _parse_json_content(body)
assert text == "describe this"
assert len(media_paths) == 1
finally:
os.chdir(original_cwd)
def test_parse_json_content_plain_text_only() -> None:
"""Plain text string content returns no media."""
body = {"messages": [{"role": "user", "content": "hello"}]}
text, media_paths = _parse_json_content(body)
assert text == "hello"
assert media_paths == []
def test_parse_json_content_validates_single_message() -> None:
"""Multiple messages raise ValueError."""
body = {
"messages": [
{"role": "user", "content": "first"},
{"role": "user", "content": "second"},
]
}
with pytest.raises(ValueError, match="single user message"):
_parse_json_content(body)
def test_parse_json_content_validates_user_role() -> None:
"""Non-user role raises ValueError."""
body = {"messages": [{"role": "system", "content": "you are a bot"}]}
with pytest.raises(ValueError, match="single user message"):
_parse_json_content(body)
def test_parse_json_content_rejects_oversized_base64_file(tmp_path) -> None:
"""Oversized JSON data URLs should fail before writing to disk."""
large_payload = base64.b64encode(b"x" * (11 * 1024 * 1024)).decode()
body = {
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "describe"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{large_payload}"}},
],
}
]
}
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
with pytest.raises(_FileSizeExceeded, match="10MB limit"):
_parse_json_content(body)
finally:
os.chdir(original_cwd)
# ---------------------------------------------------------------------------
# Multipart upload tests
# ---------------------------------------------------------------------------
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_multipart_upload_saves_file(aiohttp_client, mock_agent, tmp_path) -> None:
"""Multipart upload saves file to media dir and passes path to process_direct."""
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
app = create_app(mock_agent, model_name="m")
client = await aiohttp_client(app)
file_data = b"test file content"
data = BytesIO(file_data)
resp = await client.post(
"/v1/chat/completions",
data={"message": "analyze this", "files": data},
)
assert resp.status == 200
call_kwargs = mock_agent.process_direct.call_args.kwargs
assert call_kwargs["content"] == "analyze this"
assert len(call_kwargs.get("media") or []) == 1
finally:
os.chdir(original_cwd)
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_multipart_multiple_files(aiohttp_client, mock_agent, tmp_path) -> None:
"""Multipart upload with multiple files saves all and passes paths."""
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
app = create_app(mock_agent, model_name="m")
client = await aiohttp_client(app)
# Note: aiohttp test client has limited multipart support
# This test verifies the basic flow
file_data = b"test content"
data = BytesIO(file_data)
resp = await client.post(
"/v1/chat/completions",
data={"message": "analyze", "files": data},
)
assert resp.status == 200
finally:
os.chdir(original_cwd)
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_multipart_file_size_limit(aiohttp_client, mock_agent, tmp_path) -> None:
"""File exceeding MAX_FILE_SIZE returns 413."""
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
app = create_app(mock_agent, model_name="m")
client = await aiohttp_client(app)
# Create a file larger than 10MB
large_data = b"x" * (11 * 1024 * 1024)
data = BytesIO(large_data)
resp = await client.post(
"/v1/chat/completions",
data={"message": "analyze", "files": data},
)
assert resp.status == 413
finally:
os.chdir(original_cwd)
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_multipart_defaults_text_when_missing(aiohttp_client, mock_agent, tmp_path) -> None:
"""Multipart without message field uses default text."""
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
app = create_app(mock_agent, model_name="m")
client = await aiohttp_client(app)
file_data = b"content"
data = BytesIO(file_data)
resp = await client.post(
"/v1/chat/completions",
data={"files": data},
)
assert resp.status == 200
call_kwargs = mock_agent.process_direct.call_args.kwargs
assert call_kwargs["content"] == "请分析上传的文件"
finally:
os.chdir(original_cwd)
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_multipart_with_session_id(aiohttp_client, mock_agent, tmp_path) -> None:
"""Multipart upload with session_id uses custom session key."""
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
app = create_app(mock_agent, model_name="m")
client = await aiohttp_client(app)
file_data = b"content"
data = BytesIO(file_data)
resp = await client.post(
"/v1/chat/completions",
data={"message": "hello", "session_id": "my-session", "files": data},
)
assert resp.status == 200
call_kwargs = mock_agent.process_direct.call_args.kwargs
assert call_kwargs["session_key"] == "api:my-session"
finally:
os.chdir(original_cwd)
# ---------------------------------------------------------------------------
# Backward compatibility tests
# ---------------------------------------------------------------------------
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_plain_text_backward_compat(aiohttp_client, mock_agent) -> None:
"""Plain text JSON request (no media) works as before."""
app = create_app(mock_agent, model_name="m")
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "hello world"}]},
)
assert resp.status == 200
body = await resp.json()
assert body["choices"][0]["message"]["content"] == "mock response"
call_kwargs = mock_agent.process_direct.call_args.kwargs
assert call_kwargs["content"] == "hello world"
assert call_kwargs.get("media") is None
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_json_base64_image_upload(aiohttp_client, mock_agent, tmp_path) -> None:
"""JSON request with base64 data URL saves file and passes path."""
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
app = create_app(mock_agent, model_name="m")
client = await aiohttp_client(app)
# Use valid base64 for a tiny PNG (1x1 transparent pixel)
tiny_png_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
resp = await client.post(
"/v1/chat/completions",
json={
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "what is this"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{tiny_png_b64}"}},
],
}
]
},
)
assert resp.status == 200
call_kwargs = mock_agent.process_direct.call_args.kwargs
assert call_kwargs["content"] == "what is this"
assert len(call_kwargs.get("media", [])) == 1
finally:
os.chdir(original_cwd)
# ---------------------------------------------------------------------------
# extract_documents tests (now in nanobot.utils.document)
# ---------------------------------------------------------------------------
def test_extract_documents_separates_images_from_docs(tmp_path) -> None:
"""Images stay in media; document text is appended to content."""
from docx import Document
png = tmp_path / "chart.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
doc = Document()
doc.add_paragraph("Quarterly revenue is $5M")
docx_path = tmp_path / "report.docx"
doc.save(docx_path)
text, image_paths = extract_documents("summarize", [str(png), str(docx_path)])
assert len(image_paths) == 1
assert image_paths[0] == str(png)
assert "Quarterly revenue" in text
assert "summarize" in text
def test_extract_documents_skips_extraction_errors(tmp_path, monkeypatch) -> None:
"""Document extraction errors should not leak into user text."""
bad_file = tmp_path / "broken.docx"
bad_file.write_text("not a docx", encoding="utf-8")
import nanobot.utils.document as _doc
monkeypatch.setattr(
_doc, "extract_text",
lambda _path: "[error: failed to extract DOCX: boom]",
)
text, image_paths = extract_documents("hello", [str(bad_file)])
assert text == "hello"
assert image_paths == []
def test_extract_documents_images_only(tmp_path) -> None:
"""When all files are images, text is unchanged and all paths kept."""
png = tmp_path / "a.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
text, image_paths = extract_documents("describe", [str(png)])
assert text == "describe"
assert len(image_paths) == 1
def test_extract_documents_skips_oversized_files(tmp_path) -> None:
"""Files exceeding the size limit should be silently skipped."""
big = tmp_path / "huge.txt"
big.write_bytes(b"x" * 200)
text, image_paths = extract_documents("hello", [str(big)], max_file_size=100)
assert text == "hello"
assert image_paths == []
def test_extract_documents_does_not_read_full_file_for_mime(tmp_path) -> None:
"""MIME detection should only read header bytes, not the entire file."""
from pathlib import Path as _Path
big_txt = tmp_path / "big.txt"
big_txt.write_bytes(b"hello world " * 100_000) # ~1.2 MB
original_read_bytes = _Path.read_bytes
read_sizes: list[int] = []
def _tracking_read_bytes(self):
data = original_read_bytes(self)
read_sizes.append(len(data))
return data
import unittest.mock
with unittest.mock.patch.object(_Path, "read_bytes", _tracking_read_bytes):
extract_documents("test", [str(big_txt)])
# If the full file was read for MIME detection, read_sizes would
# contain a >1MB entry. After the fix, only a small header is read.
assert all(size <= 4096 for size in read_sizes), (
f"extract_documents read full file for MIME detection: sizes={read_sizes}"
)
# ---------------------------------------------------------------------------
# DOCX upload test — API saves file, loop layer extracts text
# ---------------------------------------------------------------------------
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_docx_upload_passes_media_path(aiohttp_client, tmp_path) -> None:
"""Uploaded DOCX is saved to disk and its path passed as media.
(Text extraction happens later in AgentLoop._process_message.)"""
agent = _make_mock_agent("report summary")
import os
original_cwd = os.getcwd()
os.chdir(tmp_path)
try:
app = create_app(agent, model_name="m")
client = await aiohttp_client(app)
from docx import Document
doc = Document()
doc.add_paragraph("Total revenue: $5,000,000")
buf = BytesIO()
doc.save(buf)
import aiohttp
data = aiohttp.FormData()
data.add_field("message", "summarize the report")
data.add_field("files", buf.getvalue(), filename="report.docx",
content_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document")
resp = await client.post("/v1/chat/completions", data=data)
assert resp.status == 200
call_kwargs = agent.process_direct.call_args.kwargs
assert call_kwargs["content"] == "summarize the report"
media = call_kwargs.get("media", [])
assert len(media) == 1
assert "report.docx" in media[0]
finally:
os.chdir(original_cwd)
tests/test_build_status.py
+2
View File
@@ -15,6 +15,7 @@ def test_status_shows_cache_hit_rate():
)
assert "60% cached" in content
assert "2000 in / 300 out" in content
assert "Tasks: 0 active" in content
def test_status_no_cache_info():
@@ -30,6 +31,7 @@ def test_status_no_cache_info():
)
assert "cached" not in content.lower()
assert "2000 in / 300 out" in content
assert "Tasks: 0 active" in content
def test_status_zero_cached_tokens():
tests/test_context_documents.py
+113
View File
@@ -0,0 +1,113 @@
"""Tests for context builder media handling.
The ContextBuilder._build_user_content method should ONLY handle images.
Document text extraction is the responsibility of the processing layer
(AgentLoop._process_message and _drain_pending).
"""
from __future__ import annotations
from pathlib import Path
from nanobot.agent.context import ContextBuilder
from nanobot.utils.document import extract_documents
def _make_builder(tmp_path: Path) -> ContextBuilder:
"""Create a minimal ContextBuilder for testing."""
return ContextBuilder(workspace=tmp_path, timezone="UTC")
def test_build_user_content_with_no_media_returns_string(tmp_path: Path) -> None:
builder = _make_builder(tmp_path)
result = builder._build_user_content("hello", None)
assert result == "hello"
def test_build_user_content_with_image_returns_list(tmp_path: Path) -> None:
"""Image files should produce base64 content blocks."""
builder = _make_builder(tmp_path)
png = tmp_path / "test.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
result = builder._build_user_content("describe this", [str(png)])
assert isinstance(result, list)
types = [b["type"] for b in result]
assert "image_url" in types
assert "text" in types
def test_build_user_content_ignores_non_image_files(tmp_path: Path) -> None:
"""Non-image files should be silently skipped — extraction is not context builder's job."""
builder = _make_builder(tmp_path)
txt = tmp_path / "notes.txt"
txt.write_text("some text", encoding="utf-8")
result = builder._build_user_content("summarize", [str(txt)])
assert result == "summarize"
def test_build_user_content_mixed_image_and_non_image(tmp_path: Path) -> None:
"""Only images should be included; non-image files are skipped."""
builder = _make_builder(tmp_path)
png = tmp_path / "chart.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
txt = tmp_path / "report.txt"
txt.write_text("report text", encoding="utf-8")
result = builder._build_user_content("analyze", [str(png), str(txt)])
assert isinstance(result, list)
assert any(b["type"] == "image_url" for b in result)
text_parts = [b.get("text", "") for b in result if b.get("type") == "text"]
assert all("report text" not in t for t in text_parts)
# ---------------------------------------------------------------------------
# Bug detection: extract_documents must be called BEFORE _build_user_content
# to prevent document media from being silently dropped.
# This simulates the _drain_pending code path.
# ---------------------------------------------------------------------------
def test_drain_pending_path_preserves_document_text(tmp_path: Path) -> None:
"""Simulates the _drain_pending path: a pending follow-up message
with a document attachment must have its text extracted before being
passed to _build_user_content. Without extract_documents, the
document is silently dropped."""
from docx import Document
doc = Document()
doc.add_paragraph("Quarterly revenue is $5M")
docx_path = tmp_path / "report.docx"
doc.save(docx_path)
content = "summarize"
media = [str(docx_path)]
# Step 1: extract_documents separates docs from images
new_content, image_only = extract_documents(content, media)
# Step 2: _build_user_content handles only images (none left here)
builder = _make_builder(tmp_path)
result = builder._build_user_content(new_content, image_only if image_only else None)
# The document text should be present in the final content
assert "Quarterly revenue" in result
assert "summarize" in result
def test_drain_pending_path_without_extract_loses_document(tmp_path: Path) -> None:
"""Demonstrates the BUG: if _drain_pending calls _build_user_content
directly without extract_documents, document content is lost."""
from docx import Document
doc = Document()
doc.add_paragraph("Secret data in document")
docx_path = tmp_path / "report.docx"
doc.save(docx_path)
builder = _make_builder(tmp_path)
# Bug path: call _build_user_content directly with document media
result = builder._build_user_content("summarize", [str(docx_path)])
# The document text is LOST — _build_user_content ignores non-images
assert result == "summarize" # only the original text, no doc content
assert "Secret data" not in result
tests/test_document_parsing.py
+273
View File
@@ -0,0 +1,273 @@
"""Tests for document text extraction utilities."""
from pathlib import Path
from nanobot.utils.document import (
SUPPORTED_EXTENSIONS,
_is_text_extension,
extract_text,
)
class TestSupportedExtensions:
"""Test the SUPPORTED_EXTENSIONS constant."""
def test_supported_extensions_include_common_formats(self):
"""Test that common document formats are included."""
# Document formats
assert ".pdf" in SUPPORTED_EXTENSIONS
assert ".docx" in SUPPORTED_EXTENSIONS
assert ".xlsx" in SUPPORTED_EXTENSIONS
assert ".pptx" in SUPPORTED_EXTENSIONS
# Text formats
assert ".txt" in SUPPORTED_EXTENSIONS
assert ".md" in SUPPORTED_EXTENSIONS
assert ".csv" in SUPPORTED_EXTENSIONS
assert ".json" in SUPPORTED_EXTENSIONS
assert ".yaml" in SUPPORTED_EXTENSIONS
assert ".yml" in SUPPORTED_EXTENSIONS
# Image formats
assert ".png" in SUPPORTED_EXTENSIONS
assert ".jpg" in SUPPORTED_EXTENSIONS
assert ".jpeg" in SUPPORTED_EXTENSIONS
class TestExtractText:
"""Test the extract_text function."""
def test_extract_text_unsupported_returns_none(self, tmp_path: Path):
"""Test that unsupported file types return None."""
unsupported_file = tmp_path / "file.xyz"
unsupported_file.write_text("content")
result = extract_text(unsupported_file)
assert result is None
def test_extract_text_file_not_found(self, tmp_path: Path):
"""Test that non-existent files return error string."""
missing_file = tmp_path / "nonexistent.txt"
result = extract_text(missing_file)
assert result is not None
assert "[error: file not found:" in result
def test_extract_text_txt_file(self, tmp_path: Path):
"""Test extracting text from a .txt file."""
txt_file = tmp_path / "test.txt"
content = "Hello, world!\nThis is a test."
txt_file.write_text(content, encoding="utf-8")
result = extract_text(txt_file)
assert result == content
def test_extract_text_txt_file_with_truncation(self, tmp_path: Path):
"""Test that large text files are truncated."""
txt_file = tmp_path / "large.txt"
# Create content larger than _MAX_TEXT_LENGTH
content = "x" * 300_000
txt_file.write_text(content, encoding="utf-8")
result = extract_text(txt_file)
assert len(result) < 300_000
assert "(truncated," in result
assert "chars total)" in result
def test_extract_text_md_file(self, tmp_path: Path):
"""Test extracting text from a .md file."""
md_file = tmp_path / "test.md"
content = "# Header\n\nSome markdown content."
md_file.write_text(content, encoding="utf-8")
result = extract_text(md_file)
assert result == content
def test_extract_text_csv_file(self, tmp_path: Path):
"""Test extracting text from a .csv file."""
csv_file = tmp_path / "test.csv"
content = "name,age\nAlice,30\nBob,25"
csv_file.write_text(content, encoding="utf-8")
result = extract_text(csv_file)
assert result == content
def test_extract_text_json_file(self, tmp_path: Path):
"""Test extracting text from a .json file."""
json_file = tmp_path / "test.json"
content = '{"key": "value", "number": 42}'
json_file.write_text(content, encoding="utf-8")
result = extract_text(json_file)
assert result == content
def test_extract_text_xlsx(self, tmp_path: Path):
"""Test extracting text from an .xlsx file."""
from openpyxl import Workbook
xlsx_file = tmp_path / "test.xlsx"
wb = Workbook()
ws = wb.active
ws.title = "Sheet1"
ws["A1"] = "Name"
ws["B1"] = "Age"
ws["A2"] = "Alice"
ws["B2"] = 30
ws["A3"] = "Bob"
ws["B3"] = 25
# Add a second sheet
ws2 = wb.create_sheet("Sheet2")
ws2["A1"] = "Product"
ws2["B1"] = "Price"
ws2["A2"] = "Widget"
ws2["B2"] = 9.99
wb.save(xlsx_file)
wb.close()
result = extract_text(xlsx_file)
assert result is not None
assert "--- Sheet: Sheet1 ---" in result
assert "--- Sheet: Sheet2 ---" in result
assert "Alice" in result
assert "Bob" in result
assert "Widget" in result
assert "9.99" in result
def test_extract_text_xlsx_empty_sheet(self, tmp_path: Path):
"""Test extracting text from an .xlsx file with empty sheets."""
from openpyxl import Workbook
xlsx_file = tmp_path / "empty.xlsx"
wb = Workbook()
# Clear the default sheet
wb.remove(wb.active)
# Add an empty sheet
wb.create_sheet("EmptySheet")
wb.save(xlsx_file)
wb.close()
result = extract_text(xlsx_file)
# Empty sheets should return empty string or header only
assert result == "--- Sheet: EmptySheet ---" or result == ""
def test_extract_text_docx(self, tmp_path: Path):
"""Test extracting text from a .docx file."""
from docx import Document
docx_file = tmp_path / "test.docx"
doc = Document()
doc.add_heading("Test Document", 0)
doc.add_paragraph("This is paragraph one.")
doc.add_paragraph("This is paragraph two.")
doc.save(docx_file)
result = extract_text(docx_file)
assert result is not None
assert "Test Document" in result
assert "This is paragraph one." in result
assert "This is paragraph two." in result
def test_extract_text_docx_empty(self, tmp_path: Path):
"""Test extracting text from an empty .docx file."""
from docx import Document
docx_file = tmp_path / "empty.docx"
doc = Document()
doc.save(docx_file)
result = extract_text(docx_file)
assert result == ""
def test_extract_text_pptx(self, tmp_path: Path):
"""Test extracting text from a .pptx file."""
from pptx import Presentation
pptx_file = tmp_path / "test.pptx"
prs = Presentation()
# Slide 1
slide1 = prs.slides.add_slide(prs.slide_layouts[0])
for shape in slide1.shapes:
if hasattr(shape, "text"):
shape.text = "First Slide Title"
# Slide 2
slide2 = prs.slides.add_slide(prs.slide_layouts[5])
left = top = width = height = 1000000
textbox = slide2.shapes.add_textbox(left, top, width, height)
text_frame = textbox.text_frame
text_frame.text = "Bullet point content"
prs.save(pptx_file)
result = extract_text(pptx_file)
assert result is not None
assert "--- Slide 1 ---" in result
assert "--- Slide 2 ---" in result
# Text content may vary depending on PowerPoint layout defaults
assert len(result) > 0
def test_extract_text_pdf_not_found(self, tmp_path: Path):
"""Test that missing PDF files return error string."""
missing_pdf = tmp_path / "nonexistent.pdf"
result = extract_text(missing_pdf)
assert result is not None
assert "[error: file not found:" in result
def test_extract_text_image_files(self, tmp_path: Path):
"""Test that image files return placeholder text."""
# Create a minimal PNG file (1x1 pixel)
png_file = tmp_path / "test.png"
# Minimal valid PNG: 8-byte signature + IHDR + IDAT + IEND
png_data = (
b"\x89PNG\r\n\x1a\n"
b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
b"\x08\x02\x00\x00\x00\x90wS\xde"
b"\x00\x00\x00\x0cIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01"
b"\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
)
png_file.write_bytes(png_data)
result = extract_text(png_file)
assert result is not None
assert "[image:" in result
assert "test.png" in result
class TestIsTextExtension:
"""Test the _is_text_extension helper."""
def test_text_extensions_return_true(self):
"""Test that known text extensions return True."""
assert _is_text_extension(".txt") is True
assert _is_text_extension(".md") is True
assert _is_text_extension(".csv") is True
assert _is_text_extension(".json") is True
assert _is_text_extension(".yaml") is True
assert _is_text_extension(".yml") is True
assert _is_text_extension(".xml") is True
assert _is_text_extension(".html") is True
assert _is_text_extension(".htm") is True
def test_non_text_extensions_return_false(self):
"""Test that non-text extensions return False."""
assert _is_text_extension(".pdf") is False
assert _is_text_extension(".docx") is False
assert _is_text_extension(".xlsx") is False
assert _is_text_extension(".pptx") is False
assert _is_text_extension(".png") is False
assert _is_text_extension(".xyz") is False
def test_case_sensitivity(self):
"""Test that _is_text_extension requires lowercase extension.
Note: The main extract_text function handles case-insensitivity by
converting extensions to lowercase before calling _is_text_extension.
"""
# _is_text_extension itself is case-sensitive (lowercase only)
assert _is_text_extension(".txt") is True
assert _is_text_extension(".TXT") is False
assert _is_text_extension(".pdf") is False
tests/test_openai_api.py
+39 -10
View File
@@ -194,6 +194,7 @@ async def test_successful_request_uses_fixed_api_session(aiohttp_client, mock_ag
assert body["model"] == "test-model"
mock_agent.process_direct.assert_called_once_with(
content="hello",
media=None,
session_key=API_SESSION_KEY,
channel="api",
chat_id=API_CHAT_ID,
@@ -205,7 +206,7 @@ async def test_successful_request_uses_fixed_api_session(aiohttp_client, mock_ag
async def test_followup_requests_share_same_session_key(aiohttp_client) -> None:
call_log: list[str] = []
async def fake_process(content, session_key="", channel="", chat_id=""):
async def fake_process(content, session_key="", channel="", chat_id="", **kwargs):
call_log.append(session_key)
return f"reply to {content}"
@@ -236,7 +237,7 @@ async def test_followup_requests_share_same_session_key(aiohttp_client) -> None:
async def test_fixed_session_requests_are_serialized(aiohttp_client) -> None:
order: list[str] = []
async def slow_process(content, session_key="", channel="", chat_id=""):
async def slow_process(content, session_key="", channel="", chat_id="", **kwargs):
order.append(f"start:{content}")
await asyncio.sleep(0.1)
order.append(f"end:{content}")
@@ -307,12 +308,12 @@ async def test_multimodal_content_extracts_text(aiohttp_client, mock_agent) -> N
},
)
assert resp.status == 200
mock_agent.process_direct.assert_called_once_with(
content="describe this",
session_key=API_SESSION_KEY,
channel="api",
chat_id=API_CHAT_ID,
)
call_kwargs = mock_agent.process_direct.call_args.kwargs
assert call_kwargs["content"] == "describe this"
assert call_kwargs["session_key"] == API_SESSION_KEY
assert call_kwargs["channel"] == "api"
assert call_kwargs["chat_id"] == API_CHAT_ID
assert len(call_kwargs.get("media") or []) >= 0 # base64 images saved to disk
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@@ -320,7 +321,7 @@ async def test_multimodal_content_extracts_text(aiohttp_client, mock_agent) -> N
async def test_empty_response_retry_then_success(aiohttp_client) -> None:
call_count = 0
async def sometimes_empty(content, session_key="", channel="", chat_id=""):
async def sometimes_empty(content, session_key="", channel="", chat_id="", **kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
@@ -351,7 +352,7 @@ async def test_empty_response_falls_back(aiohttp_client) -> None:
call_count = 0
async def always_empty(content, session_key="", channel="", chat_id=""):
async def always_empty(content, session_key="", channel="", chat_id="", **kwargs):
nonlocal call_count
call_count += 1
return ""
@@ -371,3 +372,31 @@ async def test_empty_response_falls_back(aiohttp_client) -> None:
body = await resp.json()
assert body["choices"][0]["message"]["content"] == EMPTY_FINAL_RESPONSE_MESSAGE
assert call_count == 2
@pytest.mark.asyncio
async def test_process_direct_accepts_media() -> None:
"""process_direct should forward media paths to _process_message."""
from nanobot.agent.loop import AgentLoop
loop = AgentLoop.__new__(AgentLoop)
loop._connect_mcp = AsyncMock()
captured_msg = None
async def fake_process(msg, *, session_key="", on_progress=None, on_stream=None, on_stream_end=None):
nonlocal captured_msg
captured_msg = msg
return None
loop._process_message = fake_process
await loop.process_direct(
content="analyze this",
media=["/tmp/image.png", "/tmp/report.pdf"],
session_key="test:1",
)
assert captured_msg is not None
assert captured_msg.media == ["/tmp/image.png", "/tmp/report.pdf"]
assert captured_msg.content == "analyze this"
tests/tools/test_web_search_tool.py
+19 -5
View File
@@ -1,7 +1,5 @@
"""Tests for multi-provider web search."""
import asyncio
import httpx
import pytest
@@ -20,6 +18,25 @@ def _response(status: int = 200, json: dict | None = None) -> httpx.Response:
return r
def test_duckduckgo_search_is_exclusive():
tool = _tool(provider="duckduckgo")
assert tool.exclusive is True
assert tool.concurrency_safe is False
def test_brave_with_api_key_remains_concurrency_safe():
tool = _tool(provider="brave", api_key="brave-key")
assert tool.exclusive is False
assert tool.concurrency_safe is True
def test_brave_without_api_key_is_treated_as_duckduckgo_for_concurrency(monkeypatch):
monkeypatch.delenv("BRAVE_API_KEY", raising=False)
tool = _tool(provider="brave", api_key="")
assert tool.exclusive is True
assert tool.concurrency_safe is False
@pytest.mark.asyncio
async def test_brave_search(monkeypatch):
async def mock_get(self, url, **kw):
@@ -79,7 +96,6 @@ async def test_duckduckgo_search(monkeypatch):
import nanobot.agent.tools.web as web_mod
monkeypatch.setattr(web_mod, "DDGS", MockDDGS, raising=False)
from ddgs import DDGS
monkeypatch.setattr("ddgs.DDGS", MockDDGS)
tool = _tool(provider="duckduckgo")
@@ -265,5 +281,3 @@ async def test_duckduckgo_timeout_returns_error(monkeypatch):
result = await tool.execute(query="test")
gate.set()
assert "Error" in result