Merge origin/main into fix/structured-retry-classification-main

Made-with: Cursor
This commit is contained in:
Xubin Ren
2026-04-06 08:28:20 +00:00
108 changed files with 7719 additions and 1495 deletions
+10 -10
View File
@@ -506,7 +506,7 @@ class TestNewCommandArchival:
@pytest.mark.asyncio
async def test_new_clears_session_immediately_even_if_archive_fails(self, tmp_path: Path) -> None:
"""/new clears session immediately; archive_messages retries until raw dump."""
"""/new clears session immediately; archive is fire-and-forget."""
from nanobot.bus.events import InboundMessage
loop = self._make_loop(tmp_path)
@@ -518,12 +518,12 @@ class TestNewCommandArchival:
call_count = 0
async def _failing_consolidate(_messages) -> bool:
async def _failing_summarize(_messages) -> bool:
nonlocal call_count
call_count += 1
return False
loop.memory_consolidator.consolidate_messages = _failing_consolidate # type: ignore[method-assign]
loop.consolidator.archive = _failing_summarize # type: ignore[method-assign]
new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
response = await loop._process_message(new_msg)
@@ -535,7 +535,7 @@ class TestNewCommandArchival:
assert len(session_after.messages) == 0
await loop.close_mcp()
assert call_count == 3 # retried up to raw-archive threshold
assert call_count == 1
@pytest.mark.asyncio
async def test_new_archives_only_unconsolidated_messages(self, tmp_path: Path) -> None:
@@ -551,12 +551,12 @@ class TestNewCommandArchival:
archived_count = -1
async def _fake_consolidate(messages) -> bool:
async def _fake_summarize(messages) -> bool:
nonlocal archived_count
archived_count = len(messages)
return True
loop.memory_consolidator.consolidate_messages = _fake_consolidate # type: ignore[method-assign]
loop.consolidator.archive = _fake_summarize # type: ignore[method-assign]
new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
response = await loop._process_message(new_msg)
@@ -578,10 +578,10 @@ class TestNewCommandArchival:
session.add_message("assistant", f"resp{i}")
loop.sessions.save(session)
async def _ok_consolidate(_messages) -> bool:
async def _ok_summarize(_messages) -> bool:
return True
loop.memory_consolidator.consolidate_messages = _ok_consolidate # type: ignore[method-assign]
loop.consolidator.archive = _ok_summarize # type: ignore[method-assign]
new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
response = await loop._process_message(new_msg)
@@ -604,12 +604,12 @@ class TestNewCommandArchival:
archived = asyncio.Event()
async def _slow_consolidate(_messages) -> bool:
async def _slow_summarize(_messages) -> bool:
await asyncio.sleep(0.1)
archived.set()
return True
loop.memory_consolidator.consolidate_messages = _slow_consolidate # type: ignore[method-assign]
loop.consolidator.archive = _slow_summarize # type: ignore[method-assign]
new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
await loop._process_message(new_msg)
+78
View File
@@ -0,0 +1,78 @@
"""Tests for the lightweight Consolidator — append-only to HISTORY.md."""
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from nanobot.agent.memory import Consolidator, MemoryStore
@pytest.fixture
def store(tmp_path):
return MemoryStore(tmp_path)
@pytest.fixture
def mock_provider():
p = MagicMock()
p.chat_with_retry = AsyncMock()
return p
@pytest.fixture
def consolidator(store, mock_provider):
sessions = MagicMock()
sessions.save = MagicMock()
return Consolidator(
store=store,
provider=mock_provider,
model="test-model",
sessions=sessions,
context_window_tokens=1000,
build_messages=MagicMock(return_value=[]),
get_tool_definitions=MagicMock(return_value=[]),
max_completion_tokens=100,
)
class TestConsolidatorSummarize:
async def test_summarize_appends_to_history(self, consolidator, mock_provider, store):
"""Consolidator should call LLM to summarize, then append to HISTORY.md."""
mock_provider.chat_with_retry.return_value = MagicMock(
content="User fixed a bug in the auth module."
)
messages = [
{"role": "user", "content": "fix the auth bug"},
{"role": "assistant", "content": "Done, fixed the race condition."},
]
result = await consolidator.archive(messages)
assert result is True
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 1
async def test_summarize_raw_dumps_on_llm_failure(self, consolidator, mock_provider, store):
"""On LLM failure, raw-dump messages to HISTORY.md."""
mock_provider.chat_with_retry.side_effect = Exception("API error")
messages = [{"role": "user", "content": "hello"}]
result = await consolidator.archive(messages)
assert result is True # always succeeds
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 1
assert "[RAW]" in entries[0]["content"]
async def test_summarize_skips_empty_messages(self, consolidator):
result = await consolidator.archive([])
assert result is False
class TestConsolidatorTokenBudget:
async def test_prompt_below_threshold_does_not_consolidate(self, consolidator):
"""No consolidation when tokens are within budget."""
session = MagicMock()
session.last_consolidated = 0
session.messages = [{"role": "user", "content": "hi"}]
session.key = "test:key"
consolidator.estimate_session_prompt_tokens = MagicMock(return_value=(100, "tiktoken"))
consolidator.archive = AsyncMock(return_value=True)
await consolidator.maybe_consolidate_by_tokens(session)
consolidator.archive.assert_not_called()
+13
View File
@@ -47,6 +47,19 @@ def test_system_prompt_stays_stable_when_clock_changes(tmp_path, monkeypatch) ->
assert prompt1 == prompt2
def test_system_prompt_reflects_current_dream_memory_contract(tmp_path) -> None:
workspace = _make_workspace(tmp_path)
builder = ContextBuilder(workspace)
prompt = builder.build_system_prompt()
assert "memory/history.jsonl" in prompt
assert "automatically managed by Dream" in prompt
assert "do not edit directly" in prompt
assert "memory/HISTORY.md" not in prompt
assert "write important facts here" not in prompt
def test_runtime_context_is_separate_untrusted_user_message(tmp_path) -> None:
"""Runtime metadata should be merged with the user message."""
workspace = _make_workspace(tmp_path)
+97
View File
@@ -0,0 +1,97 @@
"""Tests for the Dream class — two-phase memory consolidation via AgentRunner."""
import pytest
from unittest.mock import AsyncMock, MagicMock
from nanobot.agent.memory import Dream, MemoryStore
from nanobot.agent.runner import AgentRunResult
@pytest.fixture
def store(tmp_path):
s = MemoryStore(tmp_path)
s.write_soul("# Soul\n- Helpful")
s.write_user("# User\n- Developer")
s.write_memory("# Memory\n- Project X active")
return s
@pytest.fixture
def mock_provider():
p = MagicMock()
p.chat_with_retry = AsyncMock()
return p
@pytest.fixture
def mock_runner():
return MagicMock()
@pytest.fixture
def dream(store, mock_provider, mock_runner):
d = Dream(store=store, provider=mock_provider, model="test-model", max_batch_size=5)
d._runner = mock_runner
return d
def _make_run_result(
stop_reason="completed",
final_content=None,
tool_events=None,
usage=None,
):
return AgentRunResult(
final_content=final_content or stop_reason,
stop_reason=stop_reason,
messages=[],
tools_used=[],
usage={},
tool_events=tool_events or [],
)
class TestDreamRun:
async def test_noop_when_no_unprocessed_history(self, dream, mock_provider, mock_runner, store):
"""Dream should not call LLM when there's nothing to process."""
result = await dream.run()
assert result is False
mock_provider.chat_with_retry.assert_not_called()
mock_runner.run.assert_not_called()
async def test_calls_runner_for_unprocessed_entries(self, dream, mock_provider, mock_runner, store):
"""Dream should call AgentRunner when there are unprocessed history entries."""
store.append_history("User prefers dark mode")
mock_provider.chat_with_retry.return_value = MagicMock(content="New fact")
mock_runner.run = AsyncMock(return_value=_make_run_result(
tool_events=[{"name": "edit_file", "status": "ok", "detail": "memory/MEMORY.md"}],
))
result = await dream.run()
assert result is True
mock_runner.run.assert_called_once()
spec = mock_runner.run.call_args[0][0]
assert spec.max_iterations == 10
assert spec.fail_on_tool_error is False
async def test_advances_dream_cursor(self, dream, mock_provider, mock_runner, store):
"""Dream should advance the cursor after processing."""
store.append_history("event 1")
store.append_history("event 2")
mock_provider.chat_with_retry.return_value = MagicMock(content="Nothing new")
mock_runner.run = AsyncMock(return_value=_make_run_result())
await dream.run()
assert store.get_last_dream_cursor() == 2
async def test_compacts_processed_history(self, dream, mock_provider, mock_runner, store):
"""Dream should compact history after processing."""
store.append_history("event 1")
store.append_history("event 2")
store.append_history("event 3")
mock_provider.chat_with_retry.return_value = MagicMock(content="Nothing new")
mock_runner.run = AsyncMock(return_value=_make_run_result())
await dream.run()
# After Dream, cursor is advanced and 3, compact keeps last max_history_entries
entries = store.read_unprocessed_history(since_cursor=0)
assert all(e["cursor"] > 0 for e in entries)
+234
View File
@@ -0,0 +1,234 @@
"""Tests for GitStore — git-backed version control for memory files."""
import pytest
from pathlib import Path
from nanobot.utils.gitstore import GitStore, CommitInfo
TRACKED = ["SOUL.md", "USER.md", "memory/MEMORY.md"]
@pytest.fixture
def git(tmp_path):
"""Uninitialized GitStore."""
return GitStore(tmp_path, tracked_files=TRACKED)
@pytest.fixture
def git_ready(git):
"""Initialized GitStore with one initial commit."""
git.init()
return git
class TestInit:
def test_not_initialized_by_default(self, git, tmp_path):
assert not git.is_initialized()
assert not (tmp_path / ".git").is_dir()
def test_init_creates_git_dir(self, git, tmp_path):
assert git.init()
assert (tmp_path / ".git").is_dir()
def test_init_idempotent(self, git_ready):
assert not git_ready.init()
def test_init_creates_gitignore(self, git_ready):
gi = git_ready._workspace / ".gitignore"
assert gi.exists()
content = gi.read_text(encoding="utf-8")
for f in TRACKED:
assert f"!{f}" in content
def test_init_touches_tracked_files(self, git_ready):
for f in TRACKED:
assert (git_ready._workspace / f).exists()
def test_init_makes_initial_commit(self, git_ready):
commits = git_ready.log()
assert len(commits) == 1
assert "init" in commits[0].message
class TestBuildGitignore:
def test_subdirectory_dirs(self, git):
content = git._build_gitignore()
assert "!memory/\n" in content
for f in TRACKED:
assert f"!{f}\n" in content
assert content.startswith("/*\n")
def test_root_level_files_no_dir_entries(self, tmp_path):
gs = GitStore(tmp_path, tracked_files=["a.md", "b.md"])
content = gs._build_gitignore()
assert "!a.md\n" in content
assert "!b.md\n" in content
dir_lines = [l for l in content.split("\n") if l.startswith("!") and l.endswith("/")]
assert dir_lines == []
class TestAutoCommit:
def test_returns_none_when_not_initialized(self, git):
assert git.auto_commit("test") is None
def test_commits_file_change(self, git_ready):
(git_ready._workspace / "SOUL.md").write_text("updated", encoding="utf-8")
sha = git_ready.auto_commit("update soul")
assert sha is not None
assert len(sha) == 8
def test_returns_none_when_no_changes(self, git_ready):
assert git_ready.auto_commit("no change") is None
def test_commit_appears_in_log(self, git_ready):
ws = git_ready._workspace
(ws / "SOUL.md").write_text("v2", encoding="utf-8")
sha = git_ready.auto_commit("update soul")
commits = git_ready.log()
assert len(commits) == 2
assert commits[0].sha == sha
def test_does_not_create_empty_commits(self, git_ready):
git_ready.auto_commit("nothing 1")
git_ready.auto_commit("nothing 2")
assert len(git_ready.log()) == 1 # only init commit
class TestLog:
def test_empty_when_not_initialized(self, git):
assert git.log() == []
def test_newest_first(self, git_ready):
ws = git_ready._workspace
for i in range(3):
(ws / "SOUL.md").write_text(f"v{i}", encoding="utf-8")
git_ready.auto_commit(f"commit {i}")
commits = git_ready.log()
assert len(commits) == 4 # init + 3
assert "commit 2" in commits[0].message
assert "init" in commits[-1].message
def test_max_entries(self, git_ready):
ws = git_ready._workspace
for i in range(10):
(ws / "SOUL.md").write_text(f"v{i}", encoding="utf-8")
git_ready.auto_commit(f"c{i}")
assert len(git_ready.log(max_entries=3)) == 3
def test_commit_info_fields(self, git_ready):
c = git_ready.log()[0]
assert isinstance(c, CommitInfo)
assert len(c.sha) == 8
assert c.timestamp
assert c.message
class TestDiffCommits:
def test_empty_when_not_initialized(self, git):
assert git.diff_commits("a", "b") == ""
def test_diff_between_two_commits(self, git_ready):
ws = git_ready._workspace
(ws / "SOUL.md").write_text("original", encoding="utf-8")
git_ready.auto_commit("v1")
(ws / "SOUL.md").write_text("modified", encoding="utf-8")
git_ready.auto_commit("v2")
commits = git_ready.log()
diff = git_ready.diff_commits(commits[1].sha, commits[0].sha)
assert "modified" in diff
def test_invalid_sha_returns_empty(self, git_ready):
assert git_ready.diff_commits("deadbeef", "cafebabe") == ""
class TestFindCommit:
def test_finds_by_prefix(self, git_ready):
ws = git_ready._workspace
(ws / "SOUL.md").write_text("v2", encoding="utf-8")
sha = git_ready.auto_commit("v2")
found = git_ready.find_commit(sha[:4])
assert found is not None
assert found.sha == sha
def test_returns_none_for_unknown(self, git_ready):
assert git_ready.find_commit("deadbeef") is None
class TestShowCommitDiff:
def test_returns_commit_with_diff(self, git_ready):
ws = git_ready._workspace
(ws / "SOUL.md").write_text("content", encoding="utf-8")
sha = git_ready.auto_commit("add content")
result = git_ready.show_commit_diff(sha)
assert result is not None
commit, diff = result
assert commit.sha == sha
assert "content" in diff
def test_first_commit_has_empty_diff(self, git_ready):
init_sha = git_ready.log()[-1].sha
result = git_ready.show_commit_diff(init_sha)
assert result is not None
_, diff = result
assert diff == ""
def test_returns_none_for_unknown(self, git_ready):
assert git_ready.show_commit_diff("deadbeef") is None
class TestCommitInfoFormat:
def test_format_with_diff(self):
from nanobot.utils.gitstore import CommitInfo
c = CommitInfo(sha="abcd1234", message="test commit\nsecond line", timestamp="2026-04-02 12:00")
result = c.format(diff="some diff")
assert "test commit" in result
assert "`abcd1234`" in result
assert "some diff" in result
def test_format_without_diff(self):
from nanobot.utils.gitstore import CommitInfo
c = CommitInfo(sha="abcd1234", message="test", timestamp="2026-04-02 12:00")
result = c.format()
assert "(no file changes)" in result
class TestRevert:
def test_returns_none_when_not_initialized(self, git):
assert git.revert("abc") is None
def test_undoes_commit_changes(self, git_ready):
"""revert(sha) should undo the given commit by restoring to its parent."""
ws = git_ready._workspace
(ws / "SOUL.md").write_text("v2 content", encoding="utf-8")
git_ready.auto_commit("v2")
commits = git_ready.log()
# commits[0] = v2 (HEAD), commits[1] = init
# Revert v2 → restore to init's state (empty SOUL.md)
new_sha = git_ready.revert(commits[0].sha)
assert new_sha is not None
assert (ws / "SOUL.md").read_text(encoding="utf-8") == ""
def test_root_commit_returns_none(self, git_ready):
"""Cannot revert the root commit (no parent to restore to)."""
commits = git_ready.log()
assert len(commits) == 1
assert git_ready.revert(commits[0].sha) is None
def test_invalid_sha_returns_none(self, git_ready):
assert git_ready.revert("deadbeef") is None
class TestMemoryStoreGitProperty:
def test_git_property_exposes_gitstore(self, tmp_path):
from nanobot.agent.memory import MemoryStore
store = MemoryStore(tmp_path)
assert isinstance(store.git, GitStore)
def test_git_property_is_same_object(self, tmp_path):
from nanobot.agent.memory import MemoryStore
store = MemoryStore(tmp_path)
assert store.git is store._git
+2 -1
View File
@@ -249,7 +249,8 @@ def _make_loop(tmp_path, hooks=None):
with patch("nanobot.agent.loop.ContextBuilder"), \
patch("nanobot.agent.loop.SessionManager"), \
patch("nanobot.agent.loop.SubagentManager") as mock_sub_mgr, \
patch("nanobot.agent.loop.MemoryConsolidator"):
patch("nanobot.agent.loop.Consolidator"), \
patch("nanobot.agent.loop.Dream"):
mock_sub_mgr.return_value.cancel_by_session = AsyncMock(return_value=0)
loop = AgentLoop(
bus=bus, provider=provider, workspace=tmp_path, hooks=hooks,
+18 -18
View File
@@ -26,24 +26,24 @@ def _make_loop(tmp_path, *, estimated_tokens: int, context_window_tokens: int) -
context_window_tokens=context_window_tokens,
)
loop.tools.get_definitions = MagicMock(return_value=[])
loop.memory_consolidator._SAFETY_BUFFER = 0
loop.consolidator._SAFETY_BUFFER = 0
return loop
@pytest.mark.asyncio
async def test_prompt_below_threshold_does_not_consolidate(tmp_path) -> None:
loop = _make_loop(tmp_path, estimated_tokens=100, context_window_tokens=200)
loop.memory_consolidator.consolidate_messages = AsyncMock(return_value=True) # type: ignore[method-assign]
loop.consolidator.archive = AsyncMock(return_value=True) # type: ignore[method-assign]
await loop.process_direct("hello", session_key="cli:test")
loop.memory_consolidator.consolidate_messages.assert_not_awaited()
loop.consolidator.archive.assert_not_awaited()
@pytest.mark.asyncio
async def test_prompt_above_threshold_triggers_consolidation(tmp_path, monkeypatch) -> None:
loop = _make_loop(tmp_path, estimated_tokens=1000, context_window_tokens=200)
loop.memory_consolidator.consolidate_messages = AsyncMock(return_value=True) # type: ignore[method-assign]
loop.consolidator.archive = AsyncMock(return_value=True) # type: ignore[method-assign]
session = loop.sessions.get_or_create("cli:test")
session.messages = [
{"role": "user", "content": "u1", "timestamp": "2026-01-01T00:00:00"},
@@ -55,13 +55,13 @@ async def test_prompt_above_threshold_triggers_consolidation(tmp_path, monkeypat
await loop.process_direct("hello", session_key="cli:test")
assert loop.memory_consolidator.consolidate_messages.await_count >= 1
assert loop.consolidator.archive.await_count >= 1
@pytest.mark.asyncio
async def test_prompt_above_threshold_archives_until_next_user_boundary(tmp_path, monkeypatch) -> None:
loop = _make_loop(tmp_path, estimated_tokens=1000, context_window_tokens=200)
loop.memory_consolidator.consolidate_messages = AsyncMock(return_value=True) # type: ignore[method-assign]
loop.consolidator.archive = AsyncMock(return_value=True) # type: ignore[method-assign]
session = loop.sessions.get_or_create("cli:test")
session.messages = [
@@ -76,9 +76,9 @@ async def test_prompt_above_threshold_archives_until_next_user_boundary(tmp_path
token_map = {"u1": 120, "a1": 120, "u2": 120, "a2": 120, "u3": 120}
monkeypatch.setattr(memory_module, "estimate_message_tokens", lambda message: token_map[message["content"]])
await loop.memory_consolidator.maybe_consolidate_by_tokens(session)
await loop.consolidator.maybe_consolidate_by_tokens(session)
archived_chunk = loop.memory_consolidator.consolidate_messages.await_args.args[0]
archived_chunk = loop.consolidator.archive.await_args.args[0]
assert [message["content"] for message in archived_chunk] == ["u1", "a1", "u2", "a2"]
assert session.last_consolidated == 4
@@ -87,7 +87,7 @@ async def test_prompt_above_threshold_archives_until_next_user_boundary(tmp_path
async def test_consolidation_loops_until_target_met(tmp_path, monkeypatch) -> None:
"""Verify maybe_consolidate_by_tokens keeps looping until under threshold."""
loop = _make_loop(tmp_path, estimated_tokens=0, context_window_tokens=200)
loop.memory_consolidator.consolidate_messages = AsyncMock(return_value=True) # type: ignore[method-assign]
loop.consolidator.archive = AsyncMock(return_value=True) # type: ignore[method-assign]
session = loop.sessions.get_or_create("cli:test")
session.messages = [
@@ -110,12 +110,12 @@ async def test_consolidation_loops_until_target_met(tmp_path, monkeypatch) -> No
return (300, "test")
return (80, "test")
loop.memory_consolidator.estimate_session_prompt_tokens = mock_estimate # type: ignore[method-assign]
loop.consolidator.estimate_session_prompt_tokens = mock_estimate # type: ignore[method-assign]
monkeypatch.setattr(memory_module, "estimate_message_tokens", lambda _m: 100)
await loop.memory_consolidator.maybe_consolidate_by_tokens(session)
await loop.consolidator.maybe_consolidate_by_tokens(session)
assert loop.memory_consolidator.consolidate_messages.await_count == 2
assert loop.consolidator.archive.await_count == 2
assert session.last_consolidated == 6
@@ -123,7 +123,7 @@ async def test_consolidation_loops_until_target_met(tmp_path, monkeypatch) -> No
async def test_consolidation_continues_below_trigger_until_half_target(tmp_path, monkeypatch) -> None:
"""Once triggered, consolidation should continue until it drops below half threshold."""
loop = _make_loop(tmp_path, estimated_tokens=0, context_window_tokens=200)
loop.memory_consolidator.consolidate_messages = AsyncMock(return_value=True) # type: ignore[method-assign]
loop.consolidator.archive = AsyncMock(return_value=True) # type: ignore[method-assign]
session = loop.sessions.get_or_create("cli:test")
session.messages = [
@@ -147,12 +147,12 @@ async def test_consolidation_continues_below_trigger_until_half_target(tmp_path,
return (150, "test")
return (80, "test")
loop.memory_consolidator.estimate_session_prompt_tokens = mock_estimate # type: ignore[method-assign]
loop.consolidator.estimate_session_prompt_tokens = mock_estimate # type: ignore[method-assign]
monkeypatch.setattr(memory_module, "estimate_message_tokens", lambda _m: 100)
await loop.memory_consolidator.maybe_consolidate_by_tokens(session)
await loop.consolidator.maybe_consolidate_by_tokens(session)
assert loop.memory_consolidator.consolidate_messages.await_count == 2
assert loop.consolidator.archive.await_count == 2
assert session.last_consolidated == 6
@@ -166,7 +166,7 @@ async def test_preflight_consolidation_before_llm_call(tmp_path, monkeypatch) ->
async def track_consolidate(messages):
order.append("consolidate")
return True
loop.memory_consolidator.consolidate_messages = track_consolidate # type: ignore[method-assign]
loop.consolidator.archive = track_consolidate # type: ignore[method-assign]
async def track_llm(*args, **kwargs):
order.append("llm")
@@ -187,7 +187,7 @@ async def test_preflight_consolidation_before_llm_call(tmp_path, monkeypatch) ->
def mock_estimate(_session):
call_count[0] += 1
return (1000 if call_count[0] <= 1 else 80, "test")
loop.memory_consolidator.estimate_session_prompt_tokens = mock_estimate # type: ignore[method-assign]
loop.consolidator.estimate_session_prompt_tokens = mock_estimate # type: ignore[method-assign]
await loop.process_direct("hello", session_key="cli:test")
@@ -1,478 +0,0 @@
"""Test MemoryStore.consolidate() handles non-string tool call arguments.
Regression test for https://github.com/HKUDS/nanobot/issues/1042
When memory consolidation receives dict values instead of strings from the LLM
tool call response, it should serialize them to JSON instead of raising TypeError.
"""
import json
from pathlib import Path
from unittest.mock import AsyncMock
import pytest
from nanobot.agent.memory import MemoryStore
from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
def _make_messages(message_count: int = 30):
"""Create a list of mock messages."""
return [
{"role": "user", "content": f"msg{i}", "timestamp": "2026-01-01 00:00"}
for i in range(message_count)
]
def _make_tool_response(history_entry, memory_update):
"""Create an LLMResponse with a save_memory tool call."""
return LLMResponse(
content=None,
tool_calls=[
ToolCallRequest(
id="call_1",
name="save_memory",
arguments={
"history_entry": history_entry,
"memory_update": memory_update,
},
)
],
)
class ScriptedProvider(LLMProvider):
def __init__(self, responses: list[LLMResponse]):
super().__init__()
self._responses = list(responses)
self.calls = 0
async def chat(self, *args, **kwargs) -> LLMResponse:
self.calls += 1
if self._responses:
return self._responses.pop(0)
return LLMResponse(content="", tool_calls=[])
def get_default_model(self) -> str:
return "test-model"
class TestMemoryConsolidationTypeHandling:
"""Test that consolidation handles various argument types correctly."""
@pytest.mark.asyncio
async def test_string_arguments_work(self, tmp_path: Path) -> None:
"""Normal case: LLM returns string arguments."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat = AsyncMock(
return_value=_make_tool_response(
history_entry="[2026-01-01] User discussed testing.",
memory_update="# Memory\nUser likes testing.",
)
)
provider.chat_with_retry = provider.chat
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is True
assert store.history_file.exists()
assert "[2026-01-01] User discussed testing." in store.history_file.read_text()
assert "User likes testing." in store.memory_file.read_text()
@pytest.mark.asyncio
async def test_dict_arguments_serialized_to_json(self, tmp_path: Path) -> None:
"""Issue #1042: LLM returns dict instead of string — must not raise TypeError."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat = AsyncMock(
return_value=_make_tool_response(
history_entry={"timestamp": "2026-01-01", "summary": "User discussed testing."},
memory_update={"facts": ["User likes testing"], "topics": ["testing"]},
)
)
provider.chat_with_retry = provider.chat
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is True
assert store.history_file.exists()
history_content = store.history_file.read_text()
parsed = json.loads(history_content.strip())
assert parsed["summary"] == "User discussed testing."
memory_content = store.memory_file.read_text()
parsed_mem = json.loads(memory_content)
assert "User likes testing" in parsed_mem["facts"]
@pytest.mark.asyncio
async def test_string_arguments_as_raw_json(self, tmp_path: Path) -> None:
"""Some providers return arguments as a JSON string instead of parsed dict."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
response = LLMResponse(
content=None,
tool_calls=[
ToolCallRequest(
id="call_1",
name="save_memory",
arguments=json.dumps({
"history_entry": "[2026-01-01] User discussed testing.",
"memory_update": "# Memory\nUser likes testing.",
}),
)
],
)
provider.chat = AsyncMock(return_value=response)
provider.chat_with_retry = provider.chat
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is True
assert "User discussed testing." in store.history_file.read_text()
@pytest.mark.asyncio
async def test_no_tool_call_returns_false(self, tmp_path: Path) -> None:
"""When LLM doesn't use the save_memory tool, return False."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat = AsyncMock(
return_value=LLMResponse(content="I summarized the conversation.", tool_calls=[])
)
provider.chat_with_retry = provider.chat
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
assert not store.history_file.exists()
@pytest.mark.asyncio
async def test_skips_when_message_chunk_is_empty(self, tmp_path: Path) -> None:
"""Consolidation should be a no-op when the selected chunk is empty."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat_with_retry = provider.chat
messages: list[dict] = []
result = await store.consolidate(messages, provider, "test-model")
assert result is True
provider.chat.assert_not_called()
@pytest.mark.asyncio
async def test_list_arguments_extracts_first_dict(self, tmp_path: Path) -> None:
"""Some providers return arguments as a list - extract first element if it's a dict."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
response = LLMResponse(
content=None,
tool_calls=[
ToolCallRequest(
id="call_1",
name="save_memory",
arguments=[{
"history_entry": "[2026-01-01] User discussed testing.",
"memory_update": "# Memory\nUser likes testing.",
}],
)
],
)
provider.chat = AsyncMock(return_value=response)
provider.chat_with_retry = provider.chat
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is True
assert "User discussed testing." in store.history_file.read_text()
assert "User likes testing." in store.memory_file.read_text()
@pytest.mark.asyncio
async def test_list_arguments_empty_list_returns_false(self, tmp_path: Path) -> None:
"""Empty list arguments should return False."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
response = LLMResponse(
content=None,
tool_calls=[
ToolCallRequest(
id="call_1",
name="save_memory",
arguments=[],
)
],
)
provider.chat = AsyncMock(return_value=response)
provider.chat_with_retry = provider.chat
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
@pytest.mark.asyncio
async def test_list_arguments_non_dict_content_returns_false(self, tmp_path: Path) -> None:
"""List with non-dict content should return False."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
response = LLMResponse(
content=None,
tool_calls=[
ToolCallRequest(
id="call_1",
name="save_memory",
arguments=["string", "content"],
)
],
)
provider.chat = AsyncMock(return_value=response)
provider.chat_with_retry = provider.chat
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
@pytest.mark.asyncio
async def test_missing_history_entry_returns_false_without_writing(self, tmp_path: Path) -> None:
"""Do not persist partial results when required fields are missing."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(
return_value=LLMResponse(
content=None,
tool_calls=[
ToolCallRequest(
id="call_1",
name="save_memory",
arguments={"memory_update": "# Memory\nOnly memory update"},
)
],
)
)
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
assert not store.history_file.exists()
assert not store.memory_file.exists()
@pytest.mark.asyncio
async def test_missing_memory_update_returns_false_without_writing(self, tmp_path: Path) -> None:
"""Do not append history if memory_update is missing."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(
return_value=LLMResponse(
content=None,
tool_calls=[
ToolCallRequest(
id="call_1",
name="save_memory",
arguments={"history_entry": "[2026-01-01] Partial output."},
)
],
)
)
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
assert not store.history_file.exists()
assert not store.memory_file.exists()
@pytest.mark.asyncio
async def test_null_required_field_returns_false_without_writing(self, tmp_path: Path) -> None:
"""Null required fields should be rejected before persistence."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(
return_value=_make_tool_response(
history_entry=None,
memory_update="# Memory\nUser likes testing.",
)
)
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
assert not store.history_file.exists()
assert not store.memory_file.exists()
@pytest.mark.asyncio
async def test_empty_history_entry_returns_false_without_writing(self, tmp_path: Path) -> None:
"""Empty history entries should be rejected to avoid blank archival records."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(
return_value=_make_tool_response(
history_entry=" ",
memory_update="# Memory\nUser likes testing.",
)
)
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
assert not store.history_file.exists()
assert not store.memory_file.exists()
@pytest.mark.asyncio
async def test_retries_transient_error_then_succeeds(self, tmp_path: Path, monkeypatch) -> None:
store = MemoryStore(tmp_path)
provider = ScriptedProvider([
LLMResponse(content="503 server error", finish_reason="error"),
_make_tool_response(
history_entry="[2026-01-01] User discussed testing.",
memory_update="# Memory\nUser likes testing.",
),
])
messages = _make_messages(message_count=60)
delays: list[int] = []
async def _fake_sleep(delay: int) -> None:
delays.append(delay)
monkeypatch.setattr("nanobot.providers.base.asyncio.sleep", _fake_sleep)
result = await store.consolidate(messages, provider, "test-model")
assert result is True
assert provider.calls == 2
assert delays == [1]
@pytest.mark.asyncio
async def test_consolidation_delegates_to_provider_defaults(self, tmp_path: Path) -> None:
"""Consolidation no longer passes generation params — the provider owns them."""
store = MemoryStore(tmp_path)
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(
return_value=_make_tool_response(
history_entry="[2026-01-01] User discussed testing.",
memory_update="# Memory\nUser likes testing.",
)
)
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is True
provider.chat_with_retry.assert_awaited_once()
_, kwargs = provider.chat_with_retry.await_args
assert kwargs["model"] == "test-model"
assert "temperature" not in kwargs
assert "max_tokens" not in kwargs
assert "reasoning_effort" not in kwargs
@pytest.mark.asyncio
async def test_tool_choice_fallback_on_unsupported_error(self, tmp_path: Path) -> None:
"""Forced tool_choice rejected by provider -> retry with auto and succeed."""
store = MemoryStore(tmp_path)
error_resp = LLMResponse(
content="Error calling LLM: BadRequestError: "
"The tool_choice parameter does not support being set to required or object",
finish_reason="error",
tool_calls=[],
)
ok_resp = _make_tool_response(
history_entry="[2026-01-01] Fallback worked.",
memory_update="# Memory\nFallback OK.",
)
call_log: list[dict] = []
async def _tracking_chat(**kwargs):
call_log.append(kwargs)
return error_resp if len(call_log) == 1 else ok_resp
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(side_effect=_tracking_chat)
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is True
assert len(call_log) == 2
assert isinstance(call_log[0]["tool_choice"], dict)
assert call_log[1]["tool_choice"] == "auto"
assert "Fallback worked." in store.history_file.read_text()
@pytest.mark.asyncio
async def test_tool_choice_fallback_auto_no_tool_call(self, tmp_path: Path) -> None:
"""Forced rejected, auto retry also produces no tool call -> return False."""
store = MemoryStore(tmp_path)
error_resp = LLMResponse(
content="Error: tool_choice must be none or auto",
finish_reason="error",
tool_calls=[],
)
no_tool_resp = LLMResponse(
content="Here is a summary.",
finish_reason="stop",
tool_calls=[],
)
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(side_effect=[error_resp, no_tool_resp])
messages = _make_messages(message_count=60)
result = await store.consolidate(messages, provider, "test-model")
assert result is False
assert not store.history_file.exists()
@pytest.mark.asyncio
async def test_raw_archive_after_consecutive_failures(self, tmp_path: Path) -> None:
"""After 3 consecutive failures, raw-archive messages and return True."""
store = MemoryStore(tmp_path)
no_tool = LLMResponse(content="No tool call.", finish_reason="stop", tool_calls=[])
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(return_value=no_tool)
messages = _make_messages(message_count=10)
assert await store.consolidate(messages, provider, "m") is False
assert await store.consolidate(messages, provider, "m") is False
assert await store.consolidate(messages, provider, "m") is True
assert store.history_file.exists()
content = store.history_file.read_text()
assert "[RAW]" in content
assert "10 messages" in content
assert "msg0" in content
assert not store.memory_file.exists()
@pytest.mark.asyncio
async def test_raw_archive_counter_resets_on_success(self, tmp_path: Path) -> None:
"""A successful consolidation resets the failure counter."""
store = MemoryStore(tmp_path)
no_tool = LLMResponse(content="Nope.", finish_reason="stop", tool_calls=[])
ok_resp = _make_tool_response(
history_entry="[2026-01-01] OK.",
memory_update="# Memory\nOK.",
)
messages = _make_messages(message_count=10)
provider = AsyncMock()
provider.chat_with_retry = AsyncMock(return_value=no_tool)
assert await store.consolidate(messages, provider, "m") is False
assert await store.consolidate(messages, provider, "m") is False
assert store._consecutive_failures == 2
provider.chat_with_retry = AsyncMock(return_value=ok_resp)
assert await store.consolidate(messages, provider, "m") is True
assert store._consecutive_failures == 0
provider.chat_with_retry = AsyncMock(return_value=no_tool)
assert await store.consolidate(messages, provider, "m") is False
assert store._consecutive_failures == 1
+267
View File
@@ -0,0 +1,267 @@
"""Tests for the restructured MemoryStore — pure file I/O layer."""
from datetime import datetime
import json
from pathlib import Path
import pytest
from nanobot.agent.memory import MemoryStore
@pytest.fixture
def store(tmp_path):
return MemoryStore(tmp_path)
class TestMemoryStoreBasicIO:
def test_read_memory_returns_empty_when_missing(self, store):
assert store.read_memory() == ""
def test_write_and_read_memory(self, store):
store.write_memory("hello")
assert store.read_memory() == "hello"
def test_read_soul_returns_empty_when_missing(self, store):
assert store.read_soul() == ""
def test_write_and_read_soul(self, store):
store.write_soul("soul content")
assert store.read_soul() == "soul content"
def test_read_user_returns_empty_when_missing(self, store):
assert store.read_user() == ""
def test_write_and_read_user(self, store):
store.write_user("user content")
assert store.read_user() == "user content"
def test_get_memory_context_returns_empty_when_missing(self, store):
assert store.get_memory_context() == ""
def test_get_memory_context_returns_formatted_content(self, store):
store.write_memory("important fact")
ctx = store.get_memory_context()
assert "Long-term Memory" in ctx
assert "important fact" in ctx
class TestHistoryWithCursor:
def test_append_history_returns_cursor(self, store):
cursor = store.append_history("event 1")
assert cursor == 1
cursor2 = store.append_history("event 2")
assert cursor2 == 2
def test_append_history_includes_cursor_in_file(self, store):
store.append_history("event 1")
content = store.read_file(store.history_file)
data = json.loads(content)
assert data["cursor"] == 1
def test_cursor_persists_across_appends(self, store):
store.append_history("event 1")
store.append_history("event 2")
cursor = store.append_history("event 3")
assert cursor == 3
def test_read_unprocessed_history(self, store):
store.append_history("event 1")
store.append_history("event 2")
store.append_history("event 3")
entries = store.read_unprocessed_history(since_cursor=1)
assert len(entries) == 2
assert entries[0]["cursor"] == 2
def test_read_unprocessed_history_returns_all_when_cursor_zero(self, store):
store.append_history("event 1")
store.append_history("event 2")
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 2
def test_compact_history_drops_oldest(self, tmp_path):
store = MemoryStore(tmp_path, max_history_entries=2)
store.append_history("event 1")
store.append_history("event 2")
store.append_history("event 3")
store.append_history("event 4")
store.append_history("event 5")
store.compact_history()
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 2
assert entries[0]["cursor"] in {4, 5}
class TestDreamCursor:
def test_initial_cursor_is_zero(self, store):
assert store.get_last_dream_cursor() == 0
def test_set_and_get_cursor(self, store):
store.set_last_dream_cursor(5)
assert store.get_last_dream_cursor() == 5
def test_cursor_persists(self, store):
store.set_last_dream_cursor(3)
store2 = MemoryStore(store.workspace)
assert store2.get_last_dream_cursor() == 3
class TestLegacyHistoryMigration:
def test_read_unprocessed_history_handles_entries_without_cursor(self, store):
"""JSONL entries with cursor=1 are correctly parsed and returned."""
store.history_file.write_text(
'{"cursor": 1, "timestamp": "2026-03-30 14:30", "content": "Old event"}\n',
encoding="utf-8")
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 1
assert entries[0]["cursor"] == 1
def test_migrates_legacy_history_md_preserving_partial_entries(self, tmp_path):
memory_dir = tmp_path / "memory"
memory_dir.mkdir()
legacy_file = memory_dir / "HISTORY.md"
legacy_content = (
"[2026-04-01 10:00] User prefers dark mode.\n\n"
"[2026-04-01 10:05] [RAW] 2 messages\n"
"[2026-04-01 10:04] USER: hello\n"
"[2026-04-01 10:04] ASSISTANT: hi\n\n"
"Legacy chunk without timestamp.\n"
"Keep whatever content we can recover.\n"
)
legacy_file.write_text(legacy_content, encoding="utf-8")
store = MemoryStore(tmp_path)
fallback_timestamp = datetime.fromtimestamp(
(memory_dir / "HISTORY.md.bak").stat().st_mtime,
).strftime("%Y-%m-%d %H:%M")
entries = store.read_unprocessed_history(since_cursor=0)
assert [entry["cursor"] for entry in entries] == [1, 2, 3]
assert entries[0]["timestamp"] == "2026-04-01 10:00"
assert entries[0]["content"] == "User prefers dark mode."
assert entries[1]["timestamp"] == "2026-04-01 10:05"
assert entries[1]["content"].startswith("[RAW] 2 messages")
assert "USER: hello" in entries[1]["content"]
assert entries[2]["timestamp"] == fallback_timestamp
assert entries[2]["content"].startswith("Legacy chunk without timestamp.")
assert store.read_file(store._cursor_file).strip() == "3"
assert store.read_file(store._dream_cursor_file).strip() == "3"
assert not legacy_file.exists()
assert (memory_dir / "HISTORY.md.bak").read_text(encoding="utf-8") == legacy_content
def test_migrates_consecutive_entries_without_blank_lines(self, tmp_path):
memory_dir = tmp_path / "memory"
memory_dir.mkdir()
legacy_file = memory_dir / "HISTORY.md"
legacy_content = (
"[2026-04-01 10:00] First event.\n"
"[2026-04-01 10:01] Second event.\n"
"[2026-04-01 10:02] Third event.\n"
)
legacy_file.write_text(legacy_content, encoding="utf-8")
store = MemoryStore(tmp_path)
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 3
assert [entry["content"] for entry in entries] == [
"First event.",
"Second event.",
"Third event.",
]
def test_raw_archive_stays_single_entry_while_following_events_split(self, tmp_path):
memory_dir = tmp_path / "memory"
memory_dir.mkdir()
legacy_file = memory_dir / "HISTORY.md"
legacy_content = (
"[2026-04-01 10:05] [RAW] 2 messages\n"
"[2026-04-01 10:04] USER: hello\n"
"[2026-04-01 10:04] ASSISTANT: hi\n"
"[2026-04-01 10:06] Normal event after raw block.\n"
)
legacy_file.write_text(legacy_content, encoding="utf-8")
store = MemoryStore(tmp_path)
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 2
assert entries[0]["content"].startswith("[RAW] 2 messages")
assert "USER: hello" in entries[0]["content"]
assert entries[1]["content"] == "Normal event after raw block."
def test_nonstandard_date_headers_still_start_new_entries(self, tmp_path):
memory_dir = tmp_path / "memory"
memory_dir.mkdir()
legacy_file = memory_dir / "HISTORY.md"
legacy_content = (
"[2026-03-252026-04-02] Multi-day summary.\n"
"[2026-03-26/27] Cross-day summary.\n"
)
legacy_file.write_text(legacy_content, encoding="utf-8")
store = MemoryStore(tmp_path)
fallback_timestamp = datetime.fromtimestamp(
(memory_dir / "HISTORY.md.bak").stat().st_mtime,
).strftime("%Y-%m-%d %H:%M")
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 2
assert entries[0]["timestamp"] == fallback_timestamp
assert entries[0]["content"] == "[2026-03-252026-04-02] Multi-day summary."
assert entries[1]["timestamp"] == fallback_timestamp
assert entries[1]["content"] == "[2026-03-26/27] Cross-day summary."
def test_existing_history_jsonl_skips_legacy_migration(self, tmp_path):
memory_dir = tmp_path / "memory"
memory_dir.mkdir()
history_file = memory_dir / "history.jsonl"
history_file.write_text(
'{"cursor": 7, "timestamp": "2026-04-01 12:00", "content": "existing"}\n',
encoding="utf-8",
)
legacy_file = memory_dir / "HISTORY.md"
legacy_file.write_text("[2026-04-01 10:00] legacy\n\n", encoding="utf-8")
store = MemoryStore(tmp_path)
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 1
assert entries[0]["cursor"] == 7
assert entries[0]["content"] == "existing"
assert legacy_file.exists()
assert not (memory_dir / "HISTORY.md.bak").exists()
def test_empty_history_jsonl_still_allows_legacy_migration(self, tmp_path):
memory_dir = tmp_path / "memory"
memory_dir.mkdir()
history_file = memory_dir / "history.jsonl"
history_file.write_text("", encoding="utf-8")
legacy_file = memory_dir / "HISTORY.md"
legacy_file.write_text("[2026-04-01 10:00] legacy\n\n", encoding="utf-8")
store = MemoryStore(tmp_path)
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 1
assert entries[0]["cursor"] == 1
assert entries[0]["timestamp"] == "2026-04-01 10:00"
assert entries[0]["content"] == "legacy"
assert not legacy_file.exists()
assert (memory_dir / "HISTORY.md.bak").exists()
def test_migrates_legacy_history_with_invalid_utf8_bytes(self, tmp_path):
memory_dir = tmp_path / "memory"
memory_dir.mkdir()
legacy_file = memory_dir / "HISTORY.md"
legacy_file.write_bytes(
b"[2026-04-01 10:00] Broken \xff data still needs migration.\n\n"
)
store = MemoryStore(tmp_path)
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 1
assert entries[0]["timestamp"] == "2026-04-01 10:00"
assert "Broken" in entries[0]["content"]
assert "migration." in entries[0]["content"]
@@ -173,6 +173,27 @@ def test_empty_session_history():
assert history == []
def test_get_history_preserves_reasoning_content():
session = Session(key="test:reasoning")
session.messages.append({"role": "user", "content": "hi"})
session.messages.append({
"role": "assistant",
"content": "done",
"reasoning_content": "hidden chain of thought",
})
history = session.get_history(max_messages=500)
assert history == [
{"role": "user", "content": "hi"},
{
"role": "assistant",
"content": "done",
"reasoning_content": "hidden chain of thought",
},
]
# --- Window cuts mid-group: assistant present but some tool results orphaned ---
def test_window_cuts_mid_tool_group():
+252
View File
@@ -0,0 +1,252 @@
"""Tests for nanobot.agent.skills.SkillsLoader."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from nanobot.agent.skills import SkillsLoader
def _write_skill(
base: Path,
name: str,
*,
metadata_json: dict | None = None,
body: str = "# Skill\n",
) -> Path:
"""Create ``base / name / SKILL.md`` with optional nanobot metadata JSON."""
skill_dir = base / name
skill_dir.mkdir(parents=True)
lines = ["---"]
if metadata_json is not None:
payload = json.dumps({"nanobot": metadata_json}, separators=(",", ":"))
lines.append(f'metadata: {payload}')
lines.extend(["---", "", body])
path = skill_dir / "SKILL.md"
path.write_text("\n".join(lines), encoding="utf-8")
return path
def test_list_skills_empty_when_skills_dir_missing(tmp_path: Path) -> None:
workspace = tmp_path / "ws"
workspace.mkdir()
builtin = tmp_path / "builtin"
builtin.mkdir()
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
assert loader.list_skills(filter_unavailable=False) == []
def test_list_skills_empty_when_skills_dir_exists_but_empty(tmp_path: Path) -> None:
workspace = tmp_path / "ws"
(workspace / "skills").mkdir(parents=True)
builtin = tmp_path / "builtin"
builtin.mkdir()
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
assert loader.list_skills(filter_unavailable=False) == []
def test_list_skills_workspace_entry_shape_and_source(tmp_path: Path) -> None:
workspace = tmp_path / "ws"
skills_root = workspace / "skills"
skills_root.mkdir(parents=True)
skill_path = _write_skill(skills_root, "alpha", body="# Alpha")
builtin = tmp_path / "builtin"
builtin.mkdir()
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
entries = loader.list_skills(filter_unavailable=False)
assert entries == [
{"name": "alpha", "path": str(skill_path), "source": "workspace"},
]
def test_list_skills_skips_non_directories_and_missing_skill_md(tmp_path: Path) -> None:
workspace = tmp_path / "ws"
skills_root = workspace / "skills"
skills_root.mkdir(parents=True)
(skills_root / "not_a_dir.txt").write_text("x", encoding="utf-8")
(skills_root / "no_skill_md").mkdir()
ok_path = _write_skill(skills_root, "ok", body="# Ok")
builtin = tmp_path / "builtin"
builtin.mkdir()
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
entries = loader.list_skills(filter_unavailable=False)
names = {entry["name"] for entry in entries}
assert names == {"ok"}
assert entries[0]["path"] == str(ok_path)
def test_list_skills_workspace_shadows_builtin_same_name(tmp_path: Path) -> None:
workspace = tmp_path / "ws"
ws_skills = workspace / "skills"
ws_skills.mkdir(parents=True)
ws_path = _write_skill(ws_skills, "dup", body="# Workspace wins")
builtin = tmp_path / "builtin"
_write_skill(builtin, "dup", body="# Builtin")
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
entries = loader.list_skills(filter_unavailable=False)
assert len(entries) == 1
assert entries[0]["source"] == "workspace"
assert entries[0]["path"] == str(ws_path)
def test_list_skills_merges_workspace_and_builtin(tmp_path: Path) -> None:
workspace = tmp_path / "ws"
ws_skills = workspace / "skills"
ws_skills.mkdir(parents=True)
ws_path = _write_skill(ws_skills, "ws_only", body="# W")
builtin = tmp_path / "builtin"
bi_path = _write_skill(builtin, "bi_only", body="# B")
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
entries = sorted(loader.list_skills(filter_unavailable=False), key=lambda item: item["name"])
assert entries == [
{"name": "bi_only", "path": str(bi_path), "source": "builtin"},
{"name": "ws_only", "path": str(ws_path), "source": "workspace"},
]
def test_list_skills_builtin_omitted_when_dir_missing(tmp_path: Path) -> None:
workspace = tmp_path / "ws"
ws_skills = workspace / "skills"
ws_skills.mkdir(parents=True)
ws_path = _write_skill(ws_skills, "solo", body="# S")
missing_builtin = tmp_path / "no_such_builtin"
loader = SkillsLoader(workspace, builtin_skills_dir=missing_builtin)
entries = loader.list_skills(filter_unavailable=False)
assert entries == [{"name": "solo", "path": str(ws_path), "source": "workspace"}]
def test_list_skills_filter_unavailable_excludes_unmet_bin_requirement(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
workspace = tmp_path / "ws"
skills_root = workspace / "skills"
skills_root.mkdir(parents=True)
_write_skill(
skills_root,
"needs_bin",
metadata_json={"requires": {"bins": ["nanobot_test_fake_binary"]}},
)
builtin = tmp_path / "builtin"
builtin.mkdir()
def fake_which(cmd: str) -> str | None:
if cmd == "nanobot_test_fake_binary":
return None
return "/usr/bin/true"
monkeypatch.setattr("nanobot.agent.skills.shutil.which", fake_which)
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
assert loader.list_skills(filter_unavailable=True) == []
def test_list_skills_filter_unavailable_includes_when_bin_requirement_met(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
workspace = tmp_path / "ws"
skills_root = workspace / "skills"
skills_root.mkdir(parents=True)
skill_path = _write_skill(
skills_root,
"has_bin",
metadata_json={"requires": {"bins": ["nanobot_test_fake_binary"]}},
)
builtin = tmp_path / "builtin"
builtin.mkdir()
def fake_which(cmd: str) -> str | None:
if cmd == "nanobot_test_fake_binary":
return "/fake/nanobot_test_fake_binary"
return None
monkeypatch.setattr("nanobot.agent.skills.shutil.which", fake_which)
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
entries = loader.list_skills(filter_unavailable=True)
assert entries == [
{"name": "has_bin", "path": str(skill_path), "source": "workspace"},
]
def test_list_skills_filter_unavailable_false_keeps_unmet_requirements(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
workspace = tmp_path / "ws"
skills_root = workspace / "skills"
skills_root.mkdir(parents=True)
skill_path = _write_skill(
skills_root,
"blocked",
metadata_json={"requires": {"bins": ["nanobot_test_fake_binary"]}},
)
builtin = tmp_path / "builtin"
builtin.mkdir()
monkeypatch.setattr("nanobot.agent.skills.shutil.which", lambda _cmd: None)
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
entries = loader.list_skills(filter_unavailable=False)
assert entries == [
{"name": "blocked", "path": str(skill_path), "source": "workspace"},
]
def test_list_skills_filter_unavailable_excludes_unmet_env_requirement(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
workspace = tmp_path / "ws"
skills_root = workspace / "skills"
skills_root.mkdir(parents=True)
_write_skill(
skills_root,
"needs_env",
metadata_json={"requires": {"env": ["NANOBOT_SKILLS_TEST_ENV_VAR"]}},
)
builtin = tmp_path / "builtin"
builtin.mkdir()
monkeypatch.delenv("NANOBOT_SKILLS_TEST_ENV_VAR", raising=False)
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
assert loader.list_skills(filter_unavailable=True) == []
def test_list_skills_openclaw_metadata_parsed_for_requirements(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
workspace = tmp_path / "ws"
skills_root = workspace / "skills"
skills_root.mkdir(parents=True)
skill_dir = skills_root / "openclaw_skill"
skill_dir.mkdir(parents=True)
skill_path = skill_dir / "SKILL.md"
oc_payload = json.dumps({"openclaw": {"requires": {"bins": ["nanobot_oc_bin"]}}}, separators=(",", ":"))
skill_path.write_text(
"\n".join(["---", f"metadata: {oc_payload}", "---", "", "# OC"]),
encoding="utf-8",
)
builtin = tmp_path / "builtin"
builtin.mkdir()
monkeypatch.setattr("nanobot.agent.skills.shutil.which", lambda _cmd: None)
loader = SkillsLoader(workspace, builtin_skills_dir=builtin)
assert loader.list_skills(filter_unavailable=True) == []
monkeypatch.setattr(
"nanobot.agent.skills.shutil.which",
lambda cmd: "/x" if cmd == "nanobot_oc_bin" else None,
)
entries = loader.list_skills(filter_unavailable=True)
assert entries == [
{"name": "openclaw_skill", "path": str(skill_path), "source": "workspace"},
]
+222
View File
@@ -1,5 +1,6 @@
from email.message import EmailMessage
from datetime import date
from pathlib import Path
import imaplib
import pytest
@@ -650,3 +651,224 @@ def test_check_authentication_results_method() -> None:
spf, dkim = EmailChannel._check_authentication_results(parsed)
assert spf is False
assert dkim is True
# ---------------------------------------------------------------------------
# Attachment extraction tests
# ---------------------------------------------------------------------------
def _make_raw_email_with_attachment(
from_addr: str = "alice@example.com",
subject: str = "With attachment",
body: str = "See attached.",
attachment_name: str = "doc.pdf",
attachment_content: bytes = b"%PDF-1.4 fake pdf content",
attachment_mime: str = "application/pdf",
auth_results: str | None = None,
) -> bytes:
msg = EmailMessage()
msg["From"] = from_addr
msg["To"] = "bot@example.com"
msg["Subject"] = subject
msg["Message-ID"] = "<m1@example.com>"
if auth_results:
msg["Authentication-Results"] = auth_results
msg.set_content(body)
maintype, subtype = attachment_mime.split("/", 1)
msg.add_attachment(
attachment_content,
maintype=maintype,
subtype=subtype,
filename=attachment_name,
)
return msg.as_bytes()
def test_extract_attachments_saves_pdf(tmp_path, monkeypatch) -> None:
"""PDF attachment is saved to media dir and path returned in media list."""
monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
raw = _make_raw_email_with_attachment()
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(allowed_attachment_types=["application/pdf"], verify_dkim=False, verify_spf=False)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert len(items[0]["media"]) == 1
saved_path = Path(items[0]["media"][0])
assert saved_path.exists()
assert saved_path.read_bytes() == b"%PDF-1.4 fake pdf content"
assert "500_doc.pdf" in saved_path.name
assert "[attachment:" in items[0]["content"]
def test_extract_attachments_disabled_by_default(monkeypatch) -> None:
"""With no allowed_attachment_types (default), no attachments are extracted."""
raw = _make_raw_email_with_attachment()
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(verify_dkim=False, verify_spf=False)
assert cfg.allowed_attachment_types == []
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert items[0]["media"] == []
assert "[attachment:" not in items[0]["content"]
def test_extract_attachments_mime_type_filter(tmp_path, monkeypatch) -> None:
"""Non-allowed MIME types are skipped."""
monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
raw = _make_raw_email_with_attachment(
attachment_name="image.png",
attachment_content=b"\x89PNG fake",
attachment_mime="image/png",
)
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(
allowed_attachment_types=["application/pdf"],
verify_dkim=False,
verify_spf=False,
)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert items[0]["media"] == []
def test_extract_attachments_empty_allowed_types_rejects_all(tmp_path, monkeypatch) -> None:
"""Empty allowed_attachment_types means no types are accepted."""
monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
raw = _make_raw_email_with_attachment(
attachment_name="image.png",
attachment_content=b"\x89PNG fake",
attachment_mime="image/png",
)
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(
allowed_attachment_types=[],
verify_dkim=False,
verify_spf=False,
)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert items[0]["media"] == []
def test_extract_attachments_wildcard_pattern(tmp_path, monkeypatch) -> None:
"""Glob patterns like 'image/*' match attachment MIME types."""
monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
raw = _make_raw_email_with_attachment(
attachment_name="photo.jpg",
attachment_content=b"\xff\xd8\xff fake jpeg",
attachment_mime="image/jpeg",
)
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(
allowed_attachment_types=["image/*"],
verify_dkim=False,
verify_spf=False,
)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert len(items[0]["media"]) == 1
def test_extract_attachments_size_limit(tmp_path, monkeypatch) -> None:
"""Attachments exceeding max_attachment_size are skipped."""
monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
raw = _make_raw_email_with_attachment(
attachment_content=b"x" * 1000,
)
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(
allowed_attachment_types=["*"],
max_attachment_size=500,
verify_dkim=False,
verify_spf=False,
)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert items[0]["media"] == []
def test_extract_attachments_max_count(tmp_path, monkeypatch) -> None:
"""Only max_attachments_per_email are saved."""
monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
# Build email with 3 attachments
msg = EmailMessage()
msg["From"] = "alice@example.com"
msg["To"] = "bot@example.com"
msg["Subject"] = "Many attachments"
msg["Message-ID"] = "<m1@example.com>"
msg.set_content("See attached.")
for i in range(3):
msg.add_attachment(
f"content {i}".encode(),
maintype="application",
subtype="pdf",
filename=f"doc{i}.pdf",
)
raw = msg.as_bytes()
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(
allowed_attachment_types=["*"],
max_attachments_per_email=2,
verify_dkim=False,
verify_spf=False,
)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert len(items[0]["media"]) == 2
def test_extract_attachments_sanitizes_filename(tmp_path, monkeypatch) -> None:
"""Path traversal in filenames is neutralized."""
monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
raw = _make_raw_email_with_attachment(
attachment_name="../../../etc/passwd",
)
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(allowed_attachment_types=["*"], verify_dkim=False, verify_spf=False)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert len(items[0]["media"]) == 1
saved_path = Path(items[0]["media"][0])
# File must be inside the media dir, not escaped via path traversal
assert saved_path.parent == tmp_path
+62
View File
@@ -0,0 +1,62 @@
"""Tests for Feishu _is_bot_mentioned logic."""
from types import SimpleNamespace
import pytest
from nanobot.channels.feishu import FeishuChannel
def _make_channel(bot_open_id: str | None = None) -> FeishuChannel:
config = SimpleNamespace(
app_id="test_id",
app_secret="test_secret",
verification_token="",
event_encrypt_key="",
group_policy="mention",
)
ch = FeishuChannel.__new__(FeishuChannel)
ch.config = config
ch._bot_open_id = bot_open_id
return ch
def _make_message(mentions=None, content="hello"):
return SimpleNamespace(content=content, mentions=mentions)
def _make_mention(open_id: str, user_id: str | None = None):
mid = SimpleNamespace(open_id=open_id, user_id=user_id)
return SimpleNamespace(id=mid)
class TestIsBotMentioned:
def test_exact_match_with_bot_open_id(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(mentions=[_make_mention("ou_bot123")])
assert ch._is_bot_mentioned(msg) is True
def test_no_match_different_bot(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(mentions=[_make_mention("ou_other_bot")])
assert ch._is_bot_mentioned(msg) is False
def test_at_all_always_matches(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(content="@_all hello")
assert ch._is_bot_mentioned(msg) is True
def test_fallback_heuristic_when_no_bot_open_id(self):
ch = _make_channel(bot_open_id=None)
msg = _make_message(mentions=[_make_mention("ou_some_bot", user_id=None)])
assert ch._is_bot_mentioned(msg) is True
def test_fallback_ignores_user_mentions(self):
ch = _make_channel(bot_open_id=None)
msg = _make_message(mentions=[_make_mention("ou_user", user_id="u_12345")])
assert ch._is_bot_mentioned(msg) is False
def test_no_mentions_returns_false(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(mentions=None)
assert ch._is_bot_mentioned(msg) is False
+238
View File
@@ -0,0 +1,238 @@
"""Tests for Feishu reaction add/remove and auto-cleanup on stream end."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from nanobot.bus.queue import MessageBus
from nanobot.channels.feishu import FeishuChannel, FeishuConfig, _FeishuStreamBuf
def _make_channel() -> FeishuChannel:
config = FeishuConfig(
enabled=True,
app_id="cli_test",
app_secret="secret",
allow_from=["*"],
)
ch = FeishuChannel(config, MessageBus())
ch._client = MagicMock()
ch._loop = None
return ch
def _mock_reaction_create_response(reaction_id: str = "reaction_001", success: bool = True):
resp = MagicMock()
resp.success.return_value = success
resp.code = 0 if success else 99999
resp.msg = "ok" if success else "error"
if success:
resp.data = SimpleNamespace(reaction_id=reaction_id)
else:
resp.data = None
return resp
# ── _add_reaction_sync ──────────────────────────────────────────────────────
class TestAddReactionSync:
def test_returns_reaction_id_on_success(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.create.return_value = _mock_reaction_create_response("rx_42")
result = ch._add_reaction_sync("om_001", "THUMBSUP")
assert result == "rx_42"
def test_returns_none_when_response_fails(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.create.return_value = _mock_reaction_create_response(success=False)
assert ch._add_reaction_sync("om_001", "THUMBSUP") is None
def test_returns_none_when_response_data_is_none(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = True
resp.data = None
ch._client.im.v1.message_reaction.create.return_value = resp
assert ch._add_reaction_sync("om_001", "THUMBSUP") is None
def test_returns_none_on_exception(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.create.side_effect = RuntimeError("network error")
assert ch._add_reaction_sync("om_001", "THUMBSUP") is None
# ── _add_reaction (async) ───────────────────────────────────────────────────
class TestAddReactionAsync:
@pytest.mark.asyncio
async def test_returns_reaction_id(self):
ch = _make_channel()
ch._add_reaction_sync = MagicMock(return_value="rx_99")
result = await ch._add_reaction("om_001", "EYES")
assert result == "rx_99"
@pytest.mark.asyncio
async def test_returns_none_when_no_client(self):
ch = _make_channel()
ch._client = None
result = await ch._add_reaction("om_001", "THUMBSUP")
assert result is None
# ── _remove_reaction_sync ───────────────────────────────────────────────────
class TestRemoveReactionSync:
def test_calls_delete_on_success(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = True
ch._client.im.v1.message_reaction.delete.return_value = resp
ch._remove_reaction_sync("om_001", "rx_42")
ch._client.im.v1.message_reaction.delete.assert_called_once()
def test_handles_failure_gracefully(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "not found"
ch._client.im.v1.message_reaction.delete.return_value = resp
# Should not raise
ch._remove_reaction_sync("om_001", "rx_42")
def test_handles_exception_gracefully(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.delete.side_effect = RuntimeError("network error")
# Should not raise
ch._remove_reaction_sync("om_001", "rx_42")
# ── _remove_reaction (async) ────────────────────────────────────────────────
class TestRemoveReactionAsync:
@pytest.mark.asyncio
async def test_calls_sync_helper(self):
ch = _make_channel()
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", "rx_42")
ch._remove_reaction_sync.assert_called_once_with("om_001", "rx_42")
@pytest.mark.asyncio
async def test_noop_when_no_client(self):
ch = _make_channel()
ch._client = None
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", "rx_42")
ch._remove_reaction_sync.assert_not_called()
@pytest.mark.asyncio
async def test_noop_when_reaction_id_is_empty(self):
ch = _make_channel()
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", "")
ch._remove_reaction_sync.assert_not_called()
@pytest.mark.asyncio
async def test_noop_when_reaction_id_is_none(self):
ch = _make_channel()
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", None)
ch._remove_reaction_sync.assert_not_called()
# ── send_delta stream end: reaction auto-cleanup ────────────────────────────
class TestStreamEndReactionCleanup:
@pytest.mark.asyncio
async def test_removes_reaction_on_stream_end(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "",
metadata={"_stream_end": True, "message_id": "om_001", "reaction_id": "rx_42"},
)
ch._remove_reaction.assert_called_once_with("om_001", "rx_42")
@pytest.mark.asyncio
async def test_no_removal_when_message_id_missing(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "",
metadata={"_stream_end": True, "reaction_id": "rx_42"},
)
ch._remove_reaction.assert_not_called()
@pytest.mark.asyncio
async def test_no_removal_when_reaction_id_missing(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "",
metadata={"_stream_end": True, "message_id": "om_001"},
)
ch._remove_reaction.assert_not_called()
@pytest.mark.asyncio
async def test_no_removal_when_both_ids_missing(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True})
ch._remove_reaction.assert_not_called()
@pytest.mark.asyncio
async def test_no_removal_when_not_stream_end(self):
ch = _make_channel()
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "more text",
metadata={"message_id": "om_001", "reaction_id": "rx_42"},
)
ch._remove_reaction.assert_not_called()
+135
View File
@@ -32,8 +32,10 @@ class _FakeHTTPXRequest:
class _FakeUpdater:
def __init__(self, on_start_polling) -> None:
self._on_start_polling = on_start_polling
self.start_polling_kwargs = None
async def start_polling(self, **kwargs) -> None:
self.start_polling_kwargs = kwargs
self._on_start_polling()
@@ -184,7 +186,11 @@ async def test_start_creates_separate_pools_with_proxy(monkeypatch) -> None:
assert poll_req.kwargs["connection_pool_size"] == 4
assert builder.request_value is api_req
assert builder.get_updates_request_value is poll_req
assert callable(app.updater.start_polling_kwargs["error_callback"])
assert any(cmd.command == "status" for cmd in app.bot.commands)
assert any(cmd.command == "dream" for cmd in app.bot.commands)
assert any(cmd.command == "dream_log" for cmd in app.bot.commands)
assert any(cmd.command == "dream_restore" for cmd in app.bot.commands)
@pytest.mark.asyncio
@@ -304,6 +310,26 @@ async def test_on_error_logs_network_issues_as_warning(monkeypatch) -> None:
assert recorded == [("warning", "Telegram network issue: proxy disconnected")]
@pytest.mark.asyncio
async def test_on_error_summarizes_empty_network_error(monkeypatch) -> None:
from telegram.error import NetworkError
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
recorded: list[tuple[str, str]] = []
monkeypatch.setattr(
"nanobot.channels.telegram.logger.warning",
lambda message, error: recorded.append(("warning", message.format(error))),
)
await channel._on_error(object(), SimpleNamespace(error=NetworkError("")))
assert recorded == [("warning", "Telegram network issue: NetworkError")]
@pytest.mark.asyncio
async def test_on_error_keeps_non_network_exceptions_as_error(monkeypatch) -> None:
channel = TelegramChannel(
@@ -359,6 +385,32 @@ async def test_send_delta_stream_end_treats_not_modified_as_success() -> None:
assert "123" not in channel._stream_bufs
@pytest.mark.asyncio
async def test_send_delta_stream_end_splits_oversized_reply() -> None:
"""Final streamed reply exceeding Telegram limit is split into chunks."""
from nanobot.channels.telegram import TELEGRAM_MAX_MESSAGE_LEN
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
channel._app.bot.edit_message_text = AsyncMock()
channel._app.bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=99))
oversized = "x" * (TELEGRAM_MAX_MESSAGE_LEN + 500)
channel._stream_bufs["123"] = _StreamBuf(text=oversized, message_id=7, last_edit=0.0)
await channel.send_delta("123", "", {"_stream_end": True})
channel._app.bot.edit_message_text.assert_called_once()
edit_text = channel._app.bot.edit_message_text.call_args.kwargs.get("text", "")
assert len(edit_text) <= TELEGRAM_MAX_MESSAGE_LEN
channel._app.bot.send_message.assert_called_once()
assert "123" not in channel._stream_bufs
@pytest.mark.asyncio
async def test_send_delta_new_stream_id_replaces_stale_buffer() -> None:
channel = TelegramChannel(
@@ -398,6 +450,23 @@ async def test_send_delta_incremental_edit_treats_not_modified_as_success() -> N
assert channel._stream_bufs["123"].last_edit > 0.0
@pytest.mark.asyncio
async def test_send_delta_initial_send_keeps_message_in_thread() -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
await channel.send_delta(
"123",
"hello",
{"_stream_delta": True, "_stream_id": "s:0", "message_thread_id": 42},
)
assert channel._app.bot.sent_messages[0]["message_thread_id"] == 42
def test_derive_topic_session_key_uses_thread_id() -> None:
message = SimpleNamespace(
chat=SimpleNamespace(type="supergroup"),
@@ -408,6 +477,27 @@ def test_derive_topic_session_key_uses_thread_id() -> None:
assert TelegramChannel._derive_topic_session_key(message) == "telegram:-100123:topic:42"
def test_derive_topic_session_key_private_dm_thread() -> None:
"""Private DM threads (Telegram Threaded Mode) must get their own session key."""
message = SimpleNamespace(
chat=SimpleNamespace(type="private"),
chat_id=999,
message_thread_id=7,
)
assert TelegramChannel._derive_topic_session_key(message) == "telegram:999:topic:7"
def test_derive_topic_session_key_none_without_thread() -> None:
"""No thread id → no topic session key, regardless of chat type."""
for chat_type in ("private", "supergroup", "group"):
message = SimpleNamespace(
chat=SimpleNamespace(type=chat_type),
chat_id=123,
message_thread_id=None,
)
assert TelegramChannel._derive_topic_session_key(message) is None
def test_get_extension_falls_back_to_original_filename() -> None:
channel = TelegramChannel(TelegramConfig(), MessageBus())
@@ -962,6 +1052,48 @@ async def test_forward_command_does_not_inject_reply_context() -> None:
assert handled[0]["content"] == "/new"
@pytest.mark.asyncio
async def test_forward_command_preserves_dream_log_args_and_strips_bot_suffix() -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], group_policy="open"),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
handled = []
async def capture_handle(**kwargs) -> None:
handled.append(kwargs)
channel._handle_message = capture_handle
update = _make_telegram_update(text="/dream-log@nanobot_test deadbeef", reply_to_message=None)
await channel._forward_command(update, None)
assert len(handled) == 1
assert handled[0]["content"] == "/dream-log deadbeef"
@pytest.mark.asyncio
async def test_forward_command_normalizes_telegram_safe_dream_aliases() -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], group_policy="open"),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
handled = []
async def capture_handle(**kwargs) -> None:
handled.append(kwargs)
channel._handle_message = capture_handle
update = _make_telegram_update(text="/dream_restore@nanobot_test deadbeef", reply_to_message=None)
await channel._forward_command(update, None)
assert len(handled) == 1
assert handled[0]["content"] == "/dream-restore deadbeef"
@pytest.mark.asyncio
async def test_on_help_includes_restart_command() -> None:
channel = TelegramChannel(
@@ -977,3 +1109,6 @@ async def test_on_help_includes_restart_command() -> None:
help_text = update.message.reply_text.await_args.args[0]
assert "/restart" in help_text
assert "/status" in help_text
assert "/dream" in help_text
assert "/dream-log" in help_text
assert "/dream-restore" in help_text
+201 -1
View File
@@ -1,12 +1,18 @@
"""Tests for WhatsApp channel outbound media support."""
import json
import os
import sys
import types
from unittest.mock import AsyncMock, MagicMock
import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.channels.whatsapp import WhatsAppChannel
from nanobot.channels.whatsapp import (
WhatsAppChannel,
_load_or_create_bridge_token,
)
def _make_channel() -> WhatsAppChannel:
@@ -155,3 +161,197 @@ async def test_group_policy_mention_accepts_mentioned_group_message():
kwargs = ch._handle_message.await_args.kwargs
assert kwargs["chat_id"] == "12345@g.us"
assert kwargs["sender_id"] == "user"
@pytest.mark.asyncio
async def test_sender_id_prefers_phone_jid_over_lid():
"""sender_id should resolve to phone number when @s.whatsapp.net JID is present."""
ch = WhatsAppChannel({"enabled": True}, MagicMock())
ch._handle_message = AsyncMock()
await ch._handle_bridge_message(
json.dumps({
"type": "message",
"id": "lid1",
"sender": "ABC123@lid.whatsapp.net",
"pn": "5551234@s.whatsapp.net",
"content": "hi",
"timestamp": 1,
})
)
kwargs = ch._handle_message.await_args.kwargs
assert kwargs["sender_id"] == "5551234"
@pytest.mark.asyncio
async def test_lid_to_phone_cache_resolves_lid_only_messages():
"""When only LID is present, a cached LID→phone mapping should be used."""
ch = WhatsAppChannel({"enabled": True}, MagicMock())
ch._handle_message = AsyncMock()
# First message: both phone and LID → builds cache
await ch._handle_bridge_message(
json.dumps({
"type": "message",
"id": "c1",
"sender": "LID99@lid.whatsapp.net",
"pn": "5559999@s.whatsapp.net",
"content": "first",
"timestamp": 1,
})
)
# Second message: only LID, no phone
await ch._handle_bridge_message(
json.dumps({
"type": "message",
"id": "c2",
"sender": "LID99@lid.whatsapp.net",
"pn": "",
"content": "second",
"timestamp": 2,
})
)
second_kwargs = ch._handle_message.await_args_list[1].kwargs
assert second_kwargs["sender_id"] == "5559999"
@pytest.mark.asyncio
async def test_voice_message_transcription_uses_media_path():
"""Voice messages are transcribed when media path is available."""
ch = WhatsAppChannel({"enabled": True}, MagicMock())
ch.transcription_provider = "openai"
ch.transcription_api_key = "sk-test"
ch._handle_message = AsyncMock()
ch.transcribe_audio = AsyncMock(return_value="Hello world")
await ch._handle_bridge_message(
json.dumps({
"type": "message",
"id": "v1",
"sender": "12345@s.whatsapp.net",
"pn": "",
"content": "[Voice Message]",
"timestamp": 1,
"media": ["/tmp/voice.ogg"],
})
)
ch.transcribe_audio.assert_awaited_once_with("/tmp/voice.ogg")
kwargs = ch._handle_message.await_args.kwargs
assert kwargs["content"].startswith("Hello world")
@pytest.mark.asyncio
async def test_voice_message_no_media_shows_not_available():
"""Voice messages without media produce a fallback placeholder."""
ch = WhatsAppChannel({"enabled": True}, MagicMock())
ch._handle_message = AsyncMock()
await ch._handle_bridge_message(
json.dumps({
"type": "message",
"id": "v2",
"sender": "12345@s.whatsapp.net",
"pn": "",
"content": "[Voice Message]",
"timestamp": 1,
})
)
kwargs = ch._handle_message.await_args.kwargs
assert kwargs["content"] == "[Voice Message: Audio not available]"
def test_load_or_create_bridge_token_persists_generated_secret(tmp_path):
token_path = tmp_path / "whatsapp-auth" / "bridge-token"
first = _load_or_create_bridge_token(token_path)
second = _load_or_create_bridge_token(token_path)
assert first == second
assert token_path.read_text(encoding="utf-8") == first
assert len(first) >= 32
if os.name != "nt":
assert token_path.stat().st_mode & 0o777 == 0o600
def test_configured_bridge_token_skips_local_token_file(monkeypatch, tmp_path):
token_path = tmp_path / "whatsapp-auth" / "bridge-token"
monkeypatch.setattr("nanobot.channels.whatsapp._bridge_token_path", lambda: token_path)
ch = WhatsAppChannel({"enabled": True, "bridgeToken": "manual-secret"}, MagicMock())
assert ch._effective_bridge_token() == "manual-secret"
assert not token_path.exists()
@pytest.mark.asyncio
async def test_login_exports_effective_bridge_token(monkeypatch, tmp_path):
token_path = tmp_path / "whatsapp-auth" / "bridge-token"
bridge_dir = tmp_path / "bridge"
bridge_dir.mkdir()
calls = []
monkeypatch.setattr("nanobot.channels.whatsapp._bridge_token_path", lambda: token_path)
monkeypatch.setattr("nanobot.channels.whatsapp._ensure_bridge_setup", lambda: bridge_dir)
monkeypatch.setattr("nanobot.channels.whatsapp.shutil.which", lambda _: "/usr/bin/npm")
def fake_run(*args, **kwargs):
calls.append((args, kwargs))
return MagicMock()
monkeypatch.setattr("nanobot.channels.whatsapp.subprocess.run", fake_run)
ch = WhatsAppChannel({"enabled": True}, MagicMock())
assert await ch.login() is True
assert len(calls) == 1
_, kwargs = calls[0]
assert kwargs["cwd"] == bridge_dir
assert kwargs["env"]["AUTH_DIR"] == str(token_path.parent)
assert kwargs["env"]["BRIDGE_TOKEN"] == token_path.read_text(encoding="utf-8")
@pytest.mark.asyncio
async def test_start_sends_auth_message_with_generated_token(monkeypatch, tmp_path):
token_path = tmp_path / "whatsapp-auth" / "bridge-token"
sent_messages: list[str] = []
class FakeWS:
def __init__(self) -> None:
self.close = AsyncMock()
async def send(self, message: str) -> None:
sent_messages.append(message)
ch._running = False
def __aiter__(self):
return self
async def __anext__(self):
raise StopAsyncIteration
class FakeConnect:
def __init__(self, ws):
self.ws = ws
async def __aenter__(self):
return self.ws
async def __aexit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr("nanobot.channels.whatsapp._bridge_token_path", lambda: token_path)
monkeypatch.setitem(
sys.modules,
"websockets",
types.SimpleNamespace(connect=lambda url: FakeConnect(FakeWS())),
)
ch = WhatsAppChannel({"enabled": True, "bridgeUrl": "ws://localhost:3001"}, MagicMock())
await ch.start()
assert sent_messages == [
json.dumps({"type": "auth", "token": token_path.read_text(encoding="utf-8")})
]
+26
View File
@@ -145,3 +145,29 @@ def test_response_renderable_without_metadata_keeps_markdown_path():
renderable = commands._response_renderable(help_text, render_markdown=True)
assert renderable.__class__.__name__ == "Markdown"
def test_stream_renderer_stop_for_input_stops_spinner():
"""stop_for_input should stop the active spinner to avoid prompt_toolkit conflicts."""
spinner = MagicMock()
mock_console = MagicMock()
mock_console.status.return_value = spinner
# Create renderer with mocked console
with patch.object(stream_mod, "_make_console", return_value=mock_console):
renderer = stream_mod.StreamRenderer(show_spinner=True)
# Verify spinner started
spinner.start.assert_called_once()
# Stop for input
renderer.stop_for_input()
# Verify spinner stopped
spinner.stop.assert_called_once()
def test_make_console_uses_force_terminal():
"""Console should be created with force_terminal=True for proper ANSI handling."""
console = stream_mod._make_console()
assert console._force_terminal is True
+117 -8
View File
@@ -1,5 +1,7 @@
import asyncio
import json
import re
import shutil
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
@@ -9,6 +11,7 @@ from typer.testing import CliRunner
from nanobot.bus.events import OutboundMessage
from nanobot.cli.commands import _make_provider, app
from nanobot.config.schema import Config
from nanobot.cron.types import CronJob, CronPayload
from nanobot.providers.openai_codex_provider import _strip_model_prefix
from nanobot.providers.registry import find_by_name
@@ -19,11 +22,6 @@ class _StopGatewayError(RuntimeError):
pass
import shutil
import pytest
@pytest.fixture
def mock_paths():
"""Mock config/workspace paths for test isolation."""
@@ -31,7 +29,6 @@ def mock_paths():
patch("nanobot.config.loader.save_config") as mock_sc, \
patch("nanobot.config.loader.load_config") as mock_lc, \
patch("nanobot.cli.commands.get_workspace_path") as mock_ws:
base_dir = Path("./test_onboard_data")
if base_dir.exists():
shutil.rmtree(base_dir)
@@ -425,13 +422,13 @@ def mock_agent_runtime(tmp_path):
config.agents.defaults.workspace = str(tmp_path / "default-workspace")
with patch("nanobot.config.loader.load_config", return_value=config) as mock_load_config, \
patch("nanobot.config.loader.resolve_config_env_vars", side_effect=lambda c: c), \
patch("nanobot.cli.commands.sync_workspace_templates") as mock_sync_templates, \
patch("nanobot.cli.commands._make_provider", return_value=object()), \
patch("nanobot.cli.commands._print_agent_response") as mock_print_response, \
patch("nanobot.bus.queue.MessageBus"), \
patch("nanobot.cron.service.CronService"), \
patch("nanobot.agent.loop.AgentLoop") as mock_agent_loop_cls:
agent_loop = MagicMock()
agent_loop.channels_config = None
agent_loop.process_direct = AsyncMock(
@@ -656,7 +653,9 @@ def test_agent_custom_config_workspace_does_not_migrate_legacy_cron(
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.agent.loop.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
"nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None
)
result = runner.invoke(app, ["agent", "-m", "hello", "-c", str(config_file)])
@@ -739,6 +738,7 @@ def _patch_cli_command_runtime(
set_config_path or (lambda _path: None),
)
monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
monkeypatch.setattr("nanobot.config.loader.resolve_config_env_vars", lambda c: c)
monkeypatch.setattr(
"nanobot.cli.commands.sync_workspace_templates",
sync_templates or (lambda _path: None),
@@ -868,6 +868,115 @@ def test_gateway_uses_workspace_directory_for_cron_store(monkeypatch, tmp_path:
assert seen["cron_store"] == config.workspace_path / "cron" / "jobs.json"
def test_gateway_cron_evaluator_receives_scheduled_reminder_context(
monkeypatch, tmp_path: Path
) -> None:
config_file = tmp_path / "instance" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text("{}")
config = Config()
config.agents.defaults.workspace = str(tmp_path / "config-workspace")
provider = object()
bus = MagicMock()
bus.publish_outbound = AsyncMock()
seen: dict[str, object] = {}
monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.commands._make_provider", lambda _config: provider)
monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: bus)
monkeypatch.setattr("nanobot.session.manager.SessionManager", lambda _workspace: object())
class _FakeCron:
def __init__(self, _store_path: Path) -> None:
self.on_job = None
seen["cron"] = self
class _FakeAgentLoop:
def __init__(self, *args, **kwargs) -> None:
self.model = "test-model"
self.tools = {}
async def process_direct(self, *_args, **_kwargs):
return OutboundMessage(
channel="telegram",
chat_id="user-1",
content="Time to stretch.",
)
async def close_mcp(self) -> None:
return None
async def run(self) -> None:
return None
def stop(self) -> None:
return None
class _StopAfterCronSetup:
def __init__(self, *_args, **_kwargs) -> None:
raise _StopGatewayError("stop")
async def _capture_evaluate_response(
response: str,
task_context: str,
provider_arg: object,
model: str,
) -> bool:
seen["response"] = response
seen["task_context"] = task_context
seen["provider"] = provider_arg
seen["model"] = model
return True
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.agent.loop.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _StopAfterCronSetup)
monkeypatch.setattr(
"nanobot.utils.evaluator.evaluate_response",
_capture_evaluate_response,
)
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
assert isinstance(result.exception, _StopGatewayError)
cron = seen["cron"]
assert isinstance(cron, _FakeCron)
assert cron.on_job is not None
job = CronJob(
id="cron-1",
name="stretch",
payload=CronPayload(
message="Remind me to stretch.",
deliver=True,
channel="telegram",
to="user-1",
),
)
response = asyncio.run(cron.on_job(job))
assert response == "Time to stretch."
assert seen["response"] == "Time to stretch."
assert seen["provider"] is provider
assert seen["model"] == "test-model"
assert seen["task_context"] == (
"[Scheduled Task] Timer finished.\n\n"
"Task 'stretch' has been triggered.\n"
"Scheduled instruction: Remind me to stretch."
)
bus.publish_outbound.assert_awaited_once_with(
OutboundMessage(
channel="telegram",
chat_id="user-1",
content="Time to stretch.",
)
)
def test_gateway_workspace_override_does_not_migrate_legacy_cron(
monkeypatch, tmp_path: Path
) -> None:
+2 -2
View File
@@ -137,7 +137,7 @@ class TestRestartCommand:
loop.sessions.get_or_create.return_value = session
loop._start_time = time.time() - 125
loop._last_usage = {"prompt_tokens": 0, "completion_tokens": 0}
loop.memory_consolidator.estimate_session_prompt_tokens = MagicMock(
loop.consolidator.estimate_session_prompt_tokens = MagicMock(
return_value=(20500, "tiktoken")
)
@@ -176,7 +176,7 @@ class TestRestartCommand:
session.get_history.return_value = [{"role": "user"}]
loop.sessions.get_or_create.return_value = session
loop._last_usage = {"prompt_tokens": 1200, "completion_tokens": 34}
loop.memory_consolidator.estimate_session_prompt_tokens = MagicMock(
loop.consolidator.estimate_session_prompt_tokens = MagicMock(
return_value=(0, "none")
)
+143
View File
@@ -0,0 +1,143 @@
from __future__ import annotations
from types import SimpleNamespace
import pytest
from nanobot.bus.events import InboundMessage
from nanobot.command.builtin import cmd_dream_log, cmd_dream_restore
from nanobot.command.router import CommandContext
from nanobot.utils.gitstore import CommitInfo
class _FakeStore:
def __init__(self, git, last_dream_cursor: int = 1):
self.git = git
self._last_dream_cursor = last_dream_cursor
def get_last_dream_cursor(self) -> int:
return self._last_dream_cursor
class _FakeGit:
def __init__(
self,
*,
initialized: bool = True,
commits: list[CommitInfo] | None = None,
diff_map: dict[str, tuple[CommitInfo, str] | None] | None = None,
revert_result: str | None = None,
):
self._initialized = initialized
self._commits = commits or []
self._diff_map = diff_map or {}
self._revert_result = revert_result
def is_initialized(self) -> bool:
return self._initialized
def log(self, max_entries: int = 20) -> list[CommitInfo]:
return self._commits[:max_entries]
def show_commit_diff(self, sha: str, max_entries: int = 20):
return self._diff_map.get(sha)
def revert(self, sha: str) -> str | None:
return self._revert_result
def _make_ctx(raw: str, git: _FakeGit, *, args: str = "", last_dream_cursor: int = 1) -> CommandContext:
msg = InboundMessage(channel="cli", sender_id="u1", chat_id="direct", content=raw)
store = _FakeStore(git, last_dream_cursor=last_dream_cursor)
loop = SimpleNamespace(consolidator=SimpleNamespace(store=store))
return CommandContext(msg=msg, session=None, key=msg.session_key, raw=raw, args=args, loop=loop)
@pytest.mark.asyncio
async def test_dream_log_latest_is_more_user_friendly() -> None:
commit = CommitInfo(sha="abcd1234", message="dream: 2026-04-04, 2 change(s)", timestamp="2026-04-04 12:00")
diff = (
"diff --git a/SOUL.md b/SOUL.md\n"
"--- a/SOUL.md\n"
"+++ b/SOUL.md\n"
"@@ -1 +1 @@\n"
"-old\n"
"+new\n"
)
git = _FakeGit(commits=[commit], diff_map={commit.sha: (commit, diff)})
out = await cmd_dream_log(_make_ctx("/dream-log", git))
assert "## Dream Update" in out.content
assert "Here is the latest Dream memory change." in out.content
assert "- Commit: `abcd1234`" in out.content
assert "- Changed files: `SOUL.md`" in out.content
assert "Use `/dream-restore abcd1234` to undo this change." in out.content
assert "```diff" in out.content
@pytest.mark.asyncio
async def test_dream_log_missing_commit_guides_user() -> None:
git = _FakeGit(diff_map={})
out = await cmd_dream_log(_make_ctx("/dream-log deadbeef", git, args="deadbeef"))
assert "Couldn't find Dream change `deadbeef`." in out.content
assert "Use `/dream-restore` to list recent versions" in out.content
@pytest.mark.asyncio
async def test_dream_log_before_first_run_is_clear() -> None:
git = _FakeGit(initialized=False)
out = await cmd_dream_log(_make_ctx("/dream-log", git, last_dream_cursor=0))
assert "Dream has not run yet." in out.content
assert "Run `/dream`" in out.content
@pytest.mark.asyncio
async def test_dream_restore_lists_versions_with_next_steps() -> None:
commits = [
CommitInfo(sha="abcd1234", message="dream: latest", timestamp="2026-04-04 12:00"),
CommitInfo(sha="bbbb2222", message="dream: older", timestamp="2026-04-04 08:00"),
]
git = _FakeGit(commits=commits)
out = await cmd_dream_restore(_make_ctx("/dream-restore", git))
assert "## Dream Restore" in out.content
assert "Choose a Dream memory version to restore." in out.content
assert "`abcd1234` 2026-04-04 12:00 - dream: latest" in out.content
assert "Preview a version with `/dream-log <sha>`" in out.content
assert "Restore a version with `/dream-restore <sha>`." in out.content
@pytest.mark.asyncio
async def test_dream_restore_success_mentions_files_and_followup() -> None:
commit = CommitInfo(sha="abcd1234", message="dream: latest", timestamp="2026-04-04 12:00")
diff = (
"diff --git a/SOUL.md b/SOUL.md\n"
"--- a/SOUL.md\n"
"+++ b/SOUL.md\n"
"@@ -1 +1 @@\n"
"-old\n"
"+new\n"
"diff --git a/memory/MEMORY.md b/memory/MEMORY.md\n"
"--- a/memory/MEMORY.md\n"
"+++ b/memory/MEMORY.md\n"
"@@ -1 +1 @@\n"
"-old\n"
"+new\n"
)
git = _FakeGit(
diff_map={commit.sha: (commit, diff)},
revert_result="eeee9999",
)
out = await cmd_dream_restore(_make_ctx("/dream-restore abcd1234", git, args="abcd1234"))
assert "Restored Dream memory to the state before `abcd1234`." in out.content
assert "- New safety commit: `eeee9999`" in out.content
assert "- Restored files: `SOUL.md`, `memory/MEMORY.md`" in out.content
assert "Use `/dream-log eeee9999` to inspect the restore diff." in out.content
+32
View File
@@ -1,6 +1,18 @@
import json
import socket
from unittest.mock import patch
from nanobot.config.loader import load_config, save_config
from nanobot.security.network import validate_url_target
def _fake_resolve(host: str, results: list[str]):
"""Return a getaddrinfo mock that maps the given host to fake IP results."""
def _resolver(hostname, port, family=0, type_=0):
if hostname == host:
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (ip, 0)) for ip in results]
raise socket.gaierror(f"cannot resolve {hostname}")
return _resolver
def test_load_config_keeps_max_tokens_and_ignores_legacy_memory_window(tmp_path) -> None:
@@ -126,3 +138,23 @@ def test_onboard_refresh_backfills_missing_channel_fields(tmp_path, monkeypatch)
assert result.exit_code == 0
saved = json.loads(config_path.read_text(encoding="utf-8"))
assert saved["channels"]["qq"]["msgFormat"] == "plain"
def test_load_config_resets_ssrf_whitelist_when_next_config_is_empty(tmp_path) -> None:
whitelisted = tmp_path / "whitelisted.json"
whitelisted.write_text(
json.dumps({"tools": {"ssrfWhitelist": ["100.64.0.0/10"]}}),
encoding="utf-8",
)
defaulted = tmp_path / "defaulted.json"
defaulted.write_text(json.dumps({}), encoding="utf-8")
load_config(whitelisted)
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
ok, err = validate_url_target("http://ts.local/api")
assert ok, err
load_config(defaulted)
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
ok, _ = validate_url_target("http://ts.local/api")
assert not ok
+48
View File
@@ -0,0 +1,48 @@
from nanobot.config.schema import DreamConfig
def test_dream_config_defaults_to_interval_hours() -> None:
cfg = DreamConfig()
assert cfg.interval_h == 2
assert cfg.cron is None
def test_dream_config_builds_every_schedule_from_interval() -> None:
cfg = DreamConfig(interval_h=3)
schedule = cfg.build_schedule("UTC")
assert schedule.kind == "every"
assert schedule.every_ms == 3 * 3_600_000
assert schedule.expr is None
def test_dream_config_honors_legacy_cron_override() -> None:
cfg = DreamConfig.model_validate({"cron": "0 */4 * * *"})
schedule = cfg.build_schedule("UTC")
assert schedule.kind == "cron"
assert schedule.expr == "0 */4 * * *"
assert schedule.tz == "UTC"
assert cfg.describe_schedule() == "cron 0 */4 * * * (legacy)"
def test_dream_config_dump_uses_interval_h_and_hides_legacy_cron() -> None:
cfg = DreamConfig.model_validate({"intervalH": 5, "cron": "0 */4 * * *"})
dumped = cfg.model_dump(by_alias=True)
assert dumped["intervalH"] == 5
assert "cron" not in dumped
def test_dream_config_uses_model_override_name_and_accepts_legacy_model() -> None:
cfg = DreamConfig.model_validate({"model": "openrouter/sonnet"})
dumped = cfg.model_dump(by_alias=True)
assert cfg.model_override == "openrouter/sonnet"
assert dumped["modelOverride"] == "openrouter/sonnet"
assert "model" not in dumped
+82
View File
@@ -0,0 +1,82 @@
import json
import pytest
from nanobot.config.loader import (
_resolve_env_vars,
load_config,
resolve_config_env_vars,
save_config,
)
class TestResolveEnvVars:
def test_replaces_string_value(self, monkeypatch):
monkeypatch.setenv("MY_SECRET", "hunter2")
assert _resolve_env_vars("${MY_SECRET}") == "hunter2"
def test_partial_replacement(self, monkeypatch):
monkeypatch.setenv("HOST", "example.com")
assert _resolve_env_vars("https://${HOST}/api") == "https://example.com/api"
def test_multiple_vars_in_one_string(self, monkeypatch):
monkeypatch.setenv("USER", "alice")
monkeypatch.setenv("PASS", "secret")
assert _resolve_env_vars("${USER}:${PASS}") == "alice:secret"
def test_nested_dicts(self, monkeypatch):
monkeypatch.setenv("TOKEN", "abc123")
data = {"channels": {"telegram": {"token": "${TOKEN}"}}}
result = _resolve_env_vars(data)
assert result["channels"]["telegram"]["token"] == "abc123"
def test_lists(self, monkeypatch):
monkeypatch.setenv("VAL", "x")
assert _resolve_env_vars(["${VAL}", "plain"]) == ["x", "plain"]
def test_ignores_non_strings(self):
assert _resolve_env_vars(42) == 42
assert _resolve_env_vars(True) is True
assert _resolve_env_vars(None) is None
assert _resolve_env_vars(3.14) == 3.14
def test_plain_strings_unchanged(self):
assert _resolve_env_vars("no vars here") == "no vars here"
def test_missing_var_raises(self):
with pytest.raises(ValueError, match="DOES_NOT_EXIST"):
_resolve_env_vars("${DOES_NOT_EXIST}")
class TestResolveConfig:
def test_resolves_env_vars_in_config(self, tmp_path, monkeypatch):
monkeypatch.setenv("TEST_API_KEY", "resolved-key")
config_path = tmp_path / "config.json"
config_path.write_text(
json.dumps(
{"providers": {"groq": {"apiKey": "${TEST_API_KEY}"}}}
),
encoding="utf-8",
)
raw = load_config(config_path)
assert raw.providers.groq.api_key == "${TEST_API_KEY}"
resolved = resolve_config_env_vars(raw)
assert resolved.providers.groq.api_key == "resolved-key"
def test_save_preserves_templates(self, tmp_path, monkeypatch):
monkeypatch.setenv("MY_TOKEN", "real-token")
config_path = tmp_path / "config.json"
config_path.write_text(
json.dumps(
{"channels": {"telegram": {"token": "${MY_TOKEN}"}}}
),
encoding="utf-8",
)
raw = load_config(config_path)
save_config(raw, config_path)
saved = json.loads(config_path.read_text(encoding="utf-8"))
assert saved["channels"]["telegram"]["token"] == "${MY_TOKEN}"
+16 -1
View File
@@ -4,7 +4,7 @@ import json
import pytest
from nanobot.cron.service import CronService
from nanobot.cron.types import CronSchedule
from nanobot.cron.types import CronJob, CronPayload, CronSchedule
def test_add_job_rejects_unknown_timezone(tmp_path) -> None:
@@ -141,3 +141,18 @@ async def test_running_service_honors_external_disable(tmp_path) -> None:
assert called == []
finally:
service.stop()
def test_remove_job_refuses_system_jobs(tmp_path) -> None:
service = CronService(tmp_path / "cron" / "jobs.json")
service.register_system_job(CronJob(
id="dream",
name="dream",
schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
payload=CronPayload(kind="system_event"),
))
result = service.remove_job("dream")
assert result == "protected"
assert service.get_job("dream") is not None
+34 -1
View File
@@ -4,7 +4,7 @@ from datetime import datetime, timezone
from nanobot.agent.tools.cron import CronTool
from nanobot.cron.service import CronService
from nanobot.cron.types import CronJobState, CronSchedule
from nanobot.cron.types import CronJob, CronJobState, CronPayload, CronSchedule
def _make_tool(tmp_path) -> CronTool:
@@ -262,6 +262,39 @@ def test_list_shows_next_run(tmp_path) -> None:
assert "(UTC)" in result
def test_list_includes_protected_dream_system_job_with_memory_purpose(tmp_path) -> None:
tool = _make_tool(tmp_path)
tool._cron.register_system_job(CronJob(
id="dream",
name="dream",
schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
payload=CronPayload(kind="system_event"),
))
result = tool._list_jobs()
assert "- dream (id: dream, cron: 0 */2 * * * (UTC))" in result
assert "Dream memory consolidation for long-term memory." in result
assert "cannot be removed" in result
def test_remove_protected_dream_job_returns_clear_feedback(tmp_path) -> None:
tool = _make_tool(tmp_path)
tool._cron.register_system_job(CronJob(
id="dream",
name="dream",
schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
payload=CronPayload(kind="system_event"),
))
result = tool._remove_job("dream")
assert "Cannot remove job `dream`." in result
assert "Dream memory consolidation job for long-term memory" in result
assert "cannot be removed" in result
assert tool._cron.get_job("dream") is not None
def test_add_cron_job_defaults_to_tool_timezone(tmp_path) -> None:
tool = _make_tool_with_tz(tmp_path, "Asia/Shanghai")
tool.set_context("telegram", "chat-1")
+86 -3
View File
@@ -226,7 +226,39 @@ def test_openai_model_passthrough() -> None:
assert provider.get_default_model() == "gpt-4o"
def test_openai_compat_strips_message_level_reasoning_fields() -> None:
def test_openai_compat_supports_temperature_matches_reasoning_model_rules() -> None:
assert OpenAICompatProvider._supports_temperature("gpt-4o") is True
assert OpenAICompatProvider._supports_temperature("gpt-5-chat") is False
assert OpenAICompatProvider._supports_temperature("o3-mini") is False
assert OpenAICompatProvider._supports_temperature("gpt-4o", reasoning_effort="medium") is False
def test_openai_compat_build_kwargs_uses_gpt5_safe_parameters() -> None:
spec = find_by_name("openai")
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
provider = OpenAICompatProvider(
api_key="sk-test-key",
default_model="gpt-5-chat",
spec=spec,
)
kwargs = provider._build_kwargs(
messages=[{"role": "user", "content": "hello"}],
tools=None,
model="gpt-5-chat",
max_tokens=4096,
temperature=0.7,
reasoning_effort=None,
tool_choice=None,
)
assert kwargs["model"] == "gpt-5-chat"
assert kwargs["max_completion_tokens"] == 4096
assert "max_tokens" not in kwargs
assert "temperature" not in kwargs
def test_openai_compat_preserves_message_level_reasoning_fields() -> None:
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
provider = OpenAICompatProvider()
@@ -247,8 +279,8 @@ def test_openai_compat_strips_message_level_reasoning_fields() -> None:
}
])
assert "reasoning_content" not in sanitized[0]
assert "extra_content" not in sanitized[0]
assert sanitized[0]["reasoning_content"] == "hidden"
assert sanitized[0]["extra_content"] == {"debug": True}
assert sanitized[0]["tool_calls"][0]["extra_content"] == {"google": {"thought_signature": "sig"}}
@@ -275,3 +307,54 @@ async def test_openai_compat_stream_watchdog_returns_error_on_stall(monkeypatch)
assert result.finish_reason == "error"
assert result.content is not None
assert "stream stalled" in result.content
# ---------------------------------------------------------------------------
# Provider-specific thinking parameters (extra_body)
# ---------------------------------------------------------------------------
def _build_kwargs_for(provider_name: str, model: str, reasoning_effort=None):
spec = find_by_name(provider_name)
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
p = OpenAICompatProvider(api_key="k", default_model=model, spec=spec)
return p._build_kwargs(
messages=[{"role": "user", "content": "hi"}],
tools=None, model=model, max_tokens=1024, temperature=0.7,
reasoning_effort=reasoning_effort, tool_choice=None,
)
def test_dashscope_thinking_enabled_with_reasoning_effort() -> None:
kw = _build_kwargs_for("dashscope", "qwen3-plus", reasoning_effort="medium")
assert kw["extra_body"] == {"enable_thinking": True}
def test_dashscope_thinking_disabled_for_minimal() -> None:
kw = _build_kwargs_for("dashscope", "qwen3-plus", reasoning_effort="minimal")
assert kw["extra_body"] == {"enable_thinking": False}
def test_dashscope_no_extra_body_when_reasoning_effort_none() -> None:
kw = _build_kwargs_for("dashscope", "qwen-turbo", reasoning_effort=None)
assert "extra_body" not in kw
def test_volcengine_thinking_enabled() -> None:
kw = _build_kwargs_for("volcengine", "doubao-seed-2-0-pro", reasoning_effort="high")
assert kw["extra_body"] == {"thinking": {"type": "enabled"}}
def test_byteplus_thinking_disabled_for_minimal() -> None:
kw = _build_kwargs_for("byteplus", "doubao-seed-2-0-pro", reasoning_effort="minimal")
assert kw["extra_body"] == {"thinking": {"type": "disabled"}}
def test_byteplus_no_extra_body_when_reasoning_effort_none() -> None:
kw = _build_kwargs_for("byteplus", "doubao-seed-2-0-pro", reasoning_effort=None)
assert "extra_body" not in kw
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
@@ -0,0 +1,87 @@
from __future__ import annotations
from typing import Any
from nanobot.providers.anthropic_provider import AnthropicProvider
from nanobot.providers.openai_compat_provider import OpenAICompatProvider
def _openai_tools(*names: str) -> list[dict[str, Any]]:
return [
{
"type": "function",
"function": {
"name": name,
"description": f"{name} tool",
"parameters": {"type": "object", "properties": {}},
},
}
for name in names
]
def _anthropic_tools(*names: str) -> list[dict[str, Any]]:
return [
{
"name": name,
"description": f"{name} tool",
"input_schema": {"type": "object", "properties": {}},
}
for name in names
]
def _marked_openai_tool_names(tools: list[dict[str, Any]] | None) -> list[str]:
if not tools:
return []
marked: list[str] = []
for tool in tools:
if "cache_control" in tool:
marked.append((tool.get("function") or {}).get("name", ""))
return marked
def _marked_anthropic_tool_names(tools: list[dict[str, Any]] | None) -> list[str]:
if not tools:
return []
return [tool.get("name", "") for tool in tools if "cache_control" in tool]
def test_openai_compat_marks_builtin_boundary_and_tail_tool() -> None:
messages = [
{"role": "system", "content": "system"},
{"role": "assistant", "content": "assistant"},
{"role": "user", "content": "user"},
]
_, marked_tools = OpenAICompatProvider._apply_cache_control(
messages,
_openai_tools("read_file", "write_file", "mcp_fs_ls", "mcp_git_status"),
)
assert _marked_openai_tool_names(marked_tools) == ["write_file", "mcp_git_status"]
def test_anthropic_marks_builtin_boundary_and_tail_tool() -> None:
messages = [
{"role": "user", "content": "u1"},
{"role": "assistant", "content": "a1"},
{"role": "user", "content": "u2"},
]
_, _, marked_tools = AnthropicProvider._apply_cache_control(
"system",
messages,
_anthropic_tools("read_file", "write_file", "mcp_fs_ls", "mcp_git_status"),
)
assert _marked_anthropic_tool_names(marked_tools) == ["write_file", "mcp_git_status"]
def test_openai_compat_marks_only_tail_without_mcp() -> None:
messages = [
{"role": "system", "content": "system"},
{"role": "assistant", "content": "assistant"},
{"role": "user", "content": "user"},
]
_, marked_tools = OpenAICompatProvider._apply_cache_control(
messages,
_openai_tools("read_file", "write_file"),
)
assert _marked_openai_tool_names(marked_tools) == ["write_file"]
+45 -1
View File
@@ -7,7 +7,7 @@ from unittest.mock import patch
import pytest
from nanobot.security.network import contains_internal_url, validate_url_target
from nanobot.security.network import configure_ssrf_whitelist, contains_internal_url, validate_url_target
def _fake_resolve(host: str, results: list[str]):
@@ -99,3 +99,47 @@ def test_allows_normal_curl():
def test_no_urls_returns_false():
assert not contains_internal_url("echo hello && ls -la")
# ---------------------------------------------------------------------------
# SSRF whitelist — allow specific CIDR ranges (#2669)
# ---------------------------------------------------------------------------
def test_blocks_cgnat_by_default():
"""100.64.0.0/10 (CGNAT / Tailscale) is blocked by default."""
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
ok, _ = validate_url_target("http://ts.local/api")
assert not ok
def test_whitelist_allows_cgnat():
"""Whitelisting 100.64.0.0/10 lets Tailscale addresses through."""
configure_ssrf_whitelist(["100.64.0.0/10"])
try:
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
ok, err = validate_url_target("http://ts.local/api")
assert ok, f"Whitelisted CGNAT should be allowed, got: {err}"
finally:
configure_ssrf_whitelist([])
def test_whitelist_does_not_affect_other_blocked():
"""Whitelisting CGNAT must not unblock other private ranges."""
configure_ssrf_whitelist(["100.64.0.0/10"])
try:
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("evil.com", ["10.0.0.1"])):
ok, _ = validate_url_target("http://evil.com/secret")
assert not ok
finally:
configure_ssrf_whitelist([])
def test_whitelist_invalid_cidr_ignored():
"""Invalid CIDR entries are silently skipped."""
configure_ssrf_whitelist(["not-a-cidr", "100.64.0.0/10"])
try:
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
ok, _ = validate_url_target("http://ts.local/api")
assert ok
finally:
configure_ssrf_whitelist([])
View File
+38
View File
@@ -0,0 +1,38 @@
"""Tests for exec tool environment isolation."""
import pytest
from nanobot.agent.tools.shell import ExecTool
@pytest.mark.asyncio
async def test_exec_does_not_leak_parent_env(monkeypatch):
"""Env vars from the parent process must not be visible to commands."""
monkeypatch.setenv("NANOBOT_SECRET_TOKEN", "super-secret-value")
tool = ExecTool()
result = await tool.execute(command="printenv NANOBOT_SECRET_TOKEN")
assert "super-secret-value" not in result
@pytest.mark.asyncio
async def test_exec_has_working_path():
"""Basic commands should be available via the login shell's PATH."""
tool = ExecTool()
result = await tool.execute(command="echo hello")
assert "hello" in result
@pytest.mark.asyncio
async def test_exec_path_append():
"""The pathAppend config should be available in the command's PATH."""
tool = ExecTool(path_append="/opt/custom/bin")
result = await tool.execute(command="echo $PATH")
assert "/opt/custom/bin" in result
@pytest.mark.asyncio
async def test_exec_path_append_preserves_system_path():
"""pathAppend must not clobber standard system paths."""
tool = ExecTool(path_append="/opt/custom/bin")
result = await tool.execute(command="ls /")
assert "Exit code: 0" in result
+121
View File
@@ -0,0 +1,121 @@
"""Tests for nanobot.agent.tools.sandbox."""
import shlex
import pytest
from nanobot.agent.tools.sandbox import wrap_command
def _parse(cmd: str) -> list[str]:
"""Split a wrapped command back into tokens for assertion."""
return shlex.split(cmd)
class TestBwrapBackend:
def test_basic_structure(self, tmp_path):
ws = str(tmp_path / "project")
result = wrap_command("bwrap", "echo hi", ws, ws)
tokens = _parse(result)
assert tokens[0] == "bwrap"
assert "--new-session" in tokens
assert "--die-with-parent" in tokens
assert "--ro-bind" in tokens
assert "--proc" in tokens
assert "--dev" in tokens
assert "--tmpfs" in tokens
sep = tokens.index("--")
assert tokens[sep + 1:] == ["sh", "-c", "echo hi"]
def test_workspace_bind_mounted_rw(self, tmp_path):
ws = str(tmp_path / "project")
result = wrap_command("bwrap", "ls", ws, ws)
tokens = _parse(result)
bind_idx = [i for i, t in enumerate(tokens) if t == "--bind"]
assert any(tokens[i + 1] == ws and tokens[i + 2] == ws for i in bind_idx)
def test_parent_dir_masked_with_tmpfs(self, tmp_path):
ws = tmp_path / "project"
result = wrap_command("bwrap", "ls", str(ws), str(ws))
tokens = _parse(result)
tmpfs_indices = [i for i, t in enumerate(tokens) if t == "--tmpfs"]
tmpfs_targets = {tokens[i + 1] for i in tmpfs_indices}
assert str(ws.parent) in tmpfs_targets
def test_cwd_inside_workspace(self, tmp_path):
ws = tmp_path / "project"
sub = ws / "src" / "lib"
result = wrap_command("bwrap", "pwd", str(ws), str(sub))
tokens = _parse(result)
chdir_idx = tokens.index("--chdir")
assert tokens[chdir_idx + 1] == str(sub)
def test_cwd_outside_workspace_falls_back(self, tmp_path):
ws = tmp_path / "project"
outside = tmp_path / "other"
result = wrap_command("bwrap", "pwd", str(ws), str(outside))
tokens = _parse(result)
chdir_idx = tokens.index("--chdir")
assert tokens[chdir_idx + 1] == str(ws.resolve())
def test_command_with_special_characters(self, tmp_path):
ws = str(tmp_path / "project")
cmd = "echo 'hello world' && cat \"file with spaces.txt\""
result = wrap_command("bwrap", cmd, ws, ws)
tokens = _parse(result)
sep = tokens.index("--")
assert tokens[sep + 1:] == ["sh", "-c", cmd]
def test_system_dirs_ro_bound(self, tmp_path):
ws = str(tmp_path / "project")
result = wrap_command("bwrap", "ls", ws, ws)
tokens = _parse(result)
ro_bind_indices = [i for i, t in enumerate(tokens) if t == "--ro-bind"]
ro_targets = {tokens[i + 1] for i in ro_bind_indices}
assert "/usr" in ro_targets
def test_optional_dirs_use_ro_bind_try(self, tmp_path):
ws = str(tmp_path / "project")
result = wrap_command("bwrap", "ls", ws, ws)
tokens = _parse(result)
try_indices = [i for i, t in enumerate(tokens) if t == "--ro-bind-try"]
try_targets = {tokens[i + 1] for i in try_indices}
assert "/bin" in try_targets
assert "/etc/ssl/certs" in try_targets
def test_media_dir_ro_bind(self, tmp_path, monkeypatch):
"""Media directory should be read-only mounted inside the sandbox."""
fake_media = tmp_path / "media"
fake_media.mkdir()
monkeypatch.setattr(
"nanobot.agent.tools.sandbox.get_media_dir",
lambda: fake_media,
)
ws = str(tmp_path / "project")
result = wrap_command("bwrap", "ls", ws, ws)
tokens = _parse(result)
try_indices = [i for i, t in enumerate(tokens) if t == "--ro-bind-try"]
try_pairs = {(tokens[i + 1], tokens[i + 2]) for i in try_indices}
assert (str(fake_media), str(fake_media)) in try_pairs
class TestUnknownBackend:
def test_raises_value_error(self, tmp_path):
ws = str(tmp_path / "project")
with pytest.raises(ValueError, match="Unknown sandbox backend"):
wrap_command("nonexistent", "ls", ws, ws)
def test_empty_string_raises(self, tmp_path):
ws = str(tmp_path / "project")
with pytest.raises(ValueError):
wrap_command("", "ls", ws, ws)
+325
View File
@@ -0,0 +1,325 @@
"""Tests for grep/glob search tools."""
from __future__ import annotations
import os
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from nanobot.agent.loop import AgentLoop
from nanobot.agent.subagent import SubagentManager
from nanobot.agent.tools.search import GlobTool, GrepTool
from nanobot.bus.queue import MessageBus
@pytest.mark.asyncio
async def test_glob_matches_recursively_and_skips_noise_dirs(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
(tmp_path / "nested").mkdir()
(tmp_path / "node_modules").mkdir()
(tmp_path / "src" / "app.py").write_text("print('ok')\n", encoding="utf-8")
(tmp_path / "nested" / "util.py").write_text("print('ok')\n", encoding="utf-8")
(tmp_path / "node_modules" / "skip.py").write_text("print('skip')\n", encoding="utf-8")
tool = GlobTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(pattern="*.py", path=".")
assert "src/app.py" in result
assert "nested/util.py" in result
assert "node_modules/skip.py" not in result
@pytest.mark.asyncio
async def test_glob_can_return_directories_only(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
(tmp_path / "src" / "api").mkdir(parents=True)
(tmp_path / "src" / "api" / "handlers.py").write_text("ok\n", encoding="utf-8")
tool = GlobTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="api",
path="src",
entry_type="dirs",
)
assert result.splitlines() == ["src/api/"]
@pytest.mark.asyncio
async def test_grep_respects_glob_filter_and_context(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
(tmp_path / "src" / "main.py").write_text(
"alpha\nbeta\nmatch_here\ngamma\n",
encoding="utf-8",
)
(tmp_path / "README.md").write_text("match_here\n", encoding="utf-8")
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="match_here",
path=".",
glob="*.py",
output_mode="content",
context_before=1,
context_after=1,
)
assert "src/main.py:3" in result
assert " 2| beta" in result
assert "> 3| match_here" in result
assert " 4| gamma" in result
assert "README.md" not in result
@pytest.mark.asyncio
async def test_grep_defaults_to_files_with_matches(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
(tmp_path / "src" / "main.py").write_text("match_here\n", encoding="utf-8")
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="match_here",
path="src",
)
assert result.splitlines() == ["src/main.py"]
assert "1|" not in result
@pytest.mark.asyncio
async def test_grep_supports_case_insensitive_search(tmp_path: Path) -> None:
(tmp_path / "memory").mkdir()
(tmp_path / "memory" / "HISTORY.md").write_text(
"[2026-04-02 10:00] OAuth token rotated\n",
encoding="utf-8",
)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="oauth",
path="memory/HISTORY.md",
case_insensitive=True,
output_mode="content",
)
assert "memory/HISTORY.md:1" in result
assert "OAuth token rotated" in result
@pytest.mark.asyncio
async def test_grep_type_filter_limits_files(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
(tmp_path / "src" / "a.py").write_text("needle\n", encoding="utf-8")
(tmp_path / "src" / "b.md").write_text("needle\n", encoding="utf-8")
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="needle",
path="src",
type="py",
)
assert result.splitlines() == ["src/a.py"]
@pytest.mark.asyncio
async def test_grep_fixed_strings_treats_regex_chars_literally(tmp_path: Path) -> None:
(tmp_path / "memory").mkdir()
(tmp_path / "memory" / "HISTORY.md").write_text(
"[2026-04-02 10:00] OAuth token rotated\n",
encoding="utf-8",
)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="[2026-04-02 10:00]",
path="memory/HISTORY.md",
fixed_strings=True,
output_mode="content",
)
assert "memory/HISTORY.md:1" in result
assert "[2026-04-02 10:00] OAuth token rotated" in result
@pytest.mark.asyncio
async def test_grep_files_with_matches_mode_returns_unique_paths(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
a = tmp_path / "src" / "a.py"
b = tmp_path / "src" / "b.py"
a.write_text("needle\nneedle\n", encoding="utf-8")
b.write_text("needle\n", encoding="utf-8")
os.utime(a, (1, 1))
os.utime(b, (2, 2))
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="needle",
path="src",
output_mode="files_with_matches",
)
assert result.splitlines() == ["src/b.py", "src/a.py"]
@pytest.mark.asyncio
async def test_grep_files_with_matches_supports_head_limit_and_offset(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
for name in ("a.py", "b.py", "c.py"):
(tmp_path / "src" / name).write_text("needle\n", encoding="utf-8")
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="needle",
path="src",
head_limit=1,
offset=1,
)
lines = result.splitlines()
assert lines[0] == "src/b.py"
assert "pagination: limit=1, offset=1" in result
@pytest.mark.asyncio
async def test_grep_count_mode_reports_counts_per_file(tmp_path: Path) -> None:
(tmp_path / "logs").mkdir()
(tmp_path / "logs" / "one.log").write_text("warn\nok\nwarn\n", encoding="utf-8")
(tmp_path / "logs" / "two.log").write_text("warn\n", encoding="utf-8")
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="warn",
path="logs",
output_mode="count",
)
assert "logs/one.log: 2" in result
assert "logs/two.log: 1" in result
assert "total matches: 3 in 2 files" in result
@pytest.mark.asyncio
async def test_grep_files_with_matches_mode_respects_max_results(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
files = []
for idx, name in enumerate(("a.py", "b.py", "c.py"), start=1):
file_path = tmp_path / "src" / name
file_path.write_text("needle\n", encoding="utf-8")
os.utime(file_path, (idx, idx))
files.append(file_path)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="needle",
path="src",
output_mode="files_with_matches",
max_results=2,
)
assert result.splitlines()[:2] == ["src/c.py", "src/b.py"]
assert "pagination: limit=2, offset=0" in result
@pytest.mark.asyncio
async def test_glob_supports_head_limit_offset_and_recent_first(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
a = tmp_path / "src" / "a.py"
b = tmp_path / "src" / "b.py"
c = tmp_path / "src" / "c.py"
a.write_text("a\n", encoding="utf-8")
b.write_text("b\n", encoding="utf-8")
c.write_text("c\n", encoding="utf-8")
os.utime(a, (1, 1))
os.utime(b, (2, 2))
os.utime(c, (3, 3))
tool = GlobTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(
pattern="*.py",
path="src",
head_limit=1,
offset=1,
)
lines = result.splitlines()
assert lines[0] == "src/b.py"
assert "pagination: limit=1, offset=1" in result
@pytest.mark.asyncio
async def test_grep_reports_skipped_binary_and_large_files(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
(tmp_path / "binary.bin").write_bytes(b"\x00\x01\x02")
(tmp_path / "large.txt").write_text("x" * 20, encoding="utf-8")
monkeypatch.setattr(GrepTool, "_MAX_FILE_BYTES", 10)
tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
result = await tool.execute(pattern="needle", path=".")
assert "No matches found" in result
assert "skipped 1 binary/unreadable files" in result
assert "skipped 1 large files" in result
@pytest.mark.asyncio
async def test_search_tools_reject_paths_outside_workspace(tmp_path: Path) -> None:
outside = tmp_path.parent / "outside-search.txt"
outside.write_text("secret\n", encoding="utf-8")
grep_tool = GrepTool(workspace=tmp_path, allowed_dir=tmp_path)
glob_tool = GlobTool(workspace=tmp_path, allowed_dir=tmp_path)
grep_result = await grep_tool.execute(pattern="secret", path=str(outside))
glob_result = await glob_tool.execute(pattern="*.txt", path=str(outside.parent))
assert grep_result.startswith("Error:")
assert glob_result.startswith("Error:")
def test_agent_loop_registers_grep_and_glob(tmp_path: Path) -> None:
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
assert "grep" in loop.tools.tool_names
assert "glob" in loop.tools.tool_names
@pytest.mark.asyncio
async def test_subagent_registers_grep_and_glob(tmp_path: Path) -> None:
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
mgr = SubagentManager(
provider=provider,
workspace=tmp_path,
bus=bus,
max_tool_result_chars=4096,
)
captured: dict[str, list[str]] = {}
async def fake_run(spec):
captured["tool_names"] = spec.tools.tool_names
return SimpleNamespace(
stop_reason="ok",
final_content="done",
tool_events=[],
error=None,
)
mgr.runner.run = fake_run
mgr._announce_result = AsyncMock()
await mgr._run_subagent("sub-1", "search task", "label", {"channel": "cli", "chat_id": "direct"})
assert "grep" in captured["tool_names"]
assert "glob" in captured["tool_names"]
+49
View File
@@ -0,0 +1,49 @@
from __future__ import annotations
from typing import Any
from nanobot.agent.tools.base import Tool
from nanobot.agent.tools.registry import ToolRegistry
class _FakeTool(Tool):
def __init__(self, name: str):
self._name = name
@property
def name(self) -> str:
return self._name
@property
def description(self) -> str:
return f"{self._name} tool"
@property
def parameters(self) -> dict[str, Any]:
return {"type": "object", "properties": {}}
async def execute(self, **kwargs: Any) -> Any:
return kwargs
def _tool_names(definitions: list[dict[str, Any]]) -> list[str]:
names: list[str] = []
for definition in definitions:
fn = definition.get("function", {})
names.append(fn.get("name", ""))
return names
def test_get_definitions_orders_builtins_then_mcp_tools() -> None:
registry = ToolRegistry()
registry.register(_FakeTool("mcp_git_status"))
registry.register(_FakeTool("write_file"))
registry.register(_FakeTool("mcp_fs_list"))
registry.register(_FakeTool("read_file"))
assert _tool_names(registry.get_definitions()) == [
"read_file",
"write_file",
"mcp_fs_list",
"mcp_git_status",
]
+118 -4
View File
@@ -1,5 +1,17 @@
import shlex
import subprocess
import sys
from typing import Any
from nanobot.agent.tools import (
ArraySchema,
IntegerSchema,
ObjectSchema,
Schema,
StringSchema,
tool_parameters,
tool_parameters_schema,
)
from nanobot.agent.tools.base import Tool
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.agent.tools.shell import ExecTool
@@ -41,6 +53,103 @@ class SampleTool(Tool):
return "ok"
@tool_parameters(
tool_parameters_schema(
query=StringSchema(min_length=2),
count=IntegerSchema(2, minimum=1, maximum=10),
required=["query", "count"],
)
)
class DecoratedSampleTool(Tool):
@property
def name(self) -> str:
return "decorated_sample"
@property
def description(self) -> str:
return "decorated sample tool"
async def execute(self, **kwargs: Any) -> str:
return f"ok:{kwargs['count']}"
def test_schema_validate_value_matches_tool_validate_params() -> None:
"""ObjectSchema.validate_value 与 validate_json_schema_value、Tool.validate_params 一致。"""
root = tool_parameters_schema(
query=StringSchema(min_length=2),
count=IntegerSchema(2, minimum=1, maximum=10),
required=["query", "count"],
)
obj = ObjectSchema(
query=StringSchema(min_length=2),
count=IntegerSchema(2, minimum=1, maximum=10),
required=["query", "count"],
)
params = {"query": "h", "count": 2}
class _Mini(Tool):
@property
def name(self) -> str:
return "m"
@property
def description(self) -> str:
return ""
@property
def parameters(self) -> dict[str, Any]:
return root
async def execute(self, **kwargs: Any) -> str:
return ""
expected = _Mini().validate_params(params)
assert Schema.validate_json_schema_value(params, root, "") == expected
assert obj.validate_value(params, "") == expected
assert IntegerSchema(0, minimum=1).validate_value(0, "n") == ["n must be >= 1"]
def test_schema_classes_equivalent_to_sample_tool_parameters() -> None:
"""Schema 类生成的 JSON Schema 应与手写 dict 一致,便于校验行为一致。"""
built = tool_parameters_schema(
query=StringSchema(min_length=2),
count=IntegerSchema(2, minimum=1, maximum=10),
mode=StringSchema("", enum=["fast", "full"]),
meta=ObjectSchema(
tag=StringSchema(""),
flags=ArraySchema(StringSchema("")),
required=["tag"],
),
required=["query", "count"],
)
assert built == SampleTool().parameters
def test_tool_parameters_returns_fresh_copy_per_access() -> None:
tool = DecoratedSampleTool()
first = tool.parameters
second = tool.parameters
assert first == second
assert first is not second
assert first["properties"] is not second["properties"]
first["properties"]["query"]["minLength"] = 99
assert tool.parameters["properties"]["query"]["minLength"] == 2
async def test_registry_executes_decorated_tool_end_to_end() -> None:
reg = ToolRegistry()
reg.register(DecoratedSampleTool())
ok = await reg.execute("decorated_sample", {"query": "hello", "count": "3"})
assert ok == "ok:3"
err = await reg.execute("decorated_sample", {"query": "h", "count": 3})
assert "Invalid parameters" in err
def test_validate_params_missing_required() -> None:
tool = SampleTool()
errors = tool.validate_params({"query": "hi"})
@@ -440,10 +549,15 @@ async def test_exec_head_tail_truncation() -> None:
"""Long output should preserve both head and tail."""
tool = ExecTool()
# Generate output that exceeds _MAX_OUTPUT (10_000 chars)
# Use python to generate output to avoid command line length limits
result = await tool.execute(
command="python -c \"print('A' * 6000 + '\\n' + 'B' * 6000)\""
)
# Use current interpreter (PATH may not have `python`). ExecTool uses
# create_subprocess_shell: POSIX needs shlex.quote; Windows uses cmd.exe
# rules, so list2cmdline is appropriate there.
script = "print('A' * 6000 + '\\n' + 'B' * 6000)"
if sys.platform == "win32":
command = subprocess.list2cmdline([sys.executable, "-c", script])
else:
command = f"{shlex.quote(sys.executable)} -c {shlex.quote(script)}"
result = await tool.execute(command=command)
assert "chars truncated" in result
# Head portion should start with As
assert result.startswith("A")
+69
View File
@@ -1,5 +1,7 @@
"""Tests for multi-provider web search."""
import asyncio
import httpx
import pytest
@@ -160,3 +162,70 @@ async def test_searxng_invalid_url():
tool = _tool(provider="searxng", base_url="not-a-url")
result = await tool.execute(query="test")
assert "Error" in result
@pytest.mark.asyncio
async def test_jina_422_falls_back_to_duckduckgo(monkeypatch):
class MockDDGS:
def __init__(self, **kw):
pass
def text(self, query, max_results=5):
return [{"title": "Fallback", "href": "https://ddg.example", "body": "DuckDuckGo fallback"}]
async def mock_get(self, url, **kw):
assert "s.jina.ai" in str(url)
raise httpx.HTTPStatusError(
"422 Unprocessable Entity",
request=httpx.Request("GET", str(url)),
response=httpx.Response(422, request=httpx.Request("GET", str(url))),
)
monkeypatch.setattr(httpx.AsyncClient, "get", mock_get)
monkeypatch.setattr("ddgs.DDGS", MockDDGS)
tool = _tool(provider="jina", api_key="jina-key")
result = await tool.execute(query="test")
assert "DuckDuckGo fallback" in result
@pytest.mark.asyncio
async def test_jina_search_uses_path_encoded_query(monkeypatch):
calls = {}
async def mock_get(self, url, **kw):
calls["url"] = str(url)
calls["params"] = kw.get("params")
return _response(json={
"data": [{"title": "Jina Result", "url": "https://jina.ai", "content": "AI search"}]
})
monkeypatch.setattr(httpx.AsyncClient, "get", mock_get)
tool = _tool(provider="jina", api_key="jina-key")
await tool.execute(query="hello world")
assert calls["url"].rstrip("/") == "https://s.jina.ai/hello%20world"
assert calls["params"] in (None, {})
@pytest.mark.asyncio
async def test_duckduckgo_timeout_returns_error(monkeypatch):
"""asyncio.wait_for guard should fire when DDG search hangs."""
import threading
gate = threading.Event()
class HangingDDGS:
def __init__(self, **kw):
pass
def text(self, query, max_results=5):
gate.wait(timeout=10)
return []
monkeypatch.setattr("ddgs.DDGS", HangingDDGS)
tool = _tool(provider="duckduckgo")
tool.config.timeout = 0.2
result = await tool.execute(query="test")
gate.set()
assert "Error" in result
+303
View File
@@ -0,0 +1,303 @@
"""Tests for web search provider usage fetching and /status integration."""
from __future__ import annotations
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from nanobot.utils.searchusage import (
SearchUsageInfo,
_parse_tavily_usage,
fetch_search_usage,
)
from nanobot.utils.helpers import build_status_content
# ---------------------------------------------------------------------------
# SearchUsageInfo.format() tests
# ---------------------------------------------------------------------------
class TestSearchUsageInfoFormat:
def test_unsupported_provider_shows_no_tracking(self):
info = SearchUsageInfo(provider="duckduckgo", supported=False)
text = info.format()
assert "duckduckgo" in text
assert "not available" in text
def test_supported_with_error(self):
info = SearchUsageInfo(provider="tavily", supported=True, error="HTTP 401")
text = info.format()
assert "tavily" in text
assert "HTTP 401" in text
assert "unavailable" in text
def test_full_tavily_usage(self):
info = SearchUsageInfo(
provider="tavily",
supported=True,
used=142,
limit=1000,
remaining=858,
reset_date="2026-05-01",
search_used=120,
extract_used=15,
crawl_used=7,
)
text = info.format()
assert "tavily" in text
assert "142 / 1000" in text
assert "858" in text
assert "2026-05-01" in text
assert "Search: 120" in text
assert "Extract: 15" in text
assert "Crawl: 7" in text
def test_usage_without_limit(self):
info = SearchUsageInfo(provider="tavily", supported=True, used=50)
text = info.format()
assert "50 requests" in text
assert "/" not in text.split("Usage:")[1].split("\n")[0]
def test_no_breakdown_when_none(self):
info = SearchUsageInfo(
provider="tavily", supported=True, used=10, limit=100, remaining=90
)
text = info.format()
assert "Breakdown" not in text
def test_brave_unsupported(self):
info = SearchUsageInfo(provider="brave", supported=False)
text = info.format()
assert "brave" in text
assert "not available" in text
# ---------------------------------------------------------------------------
# _parse_tavily_usage tests
# ---------------------------------------------------------------------------
class TestParseTavilyUsage:
def test_full_response(self):
data = {
"used": 142,
"limit": 1000,
"remaining": 858,
"reset_date": "2026-05-01",
"breakdown": {"search": 120, "extract": 15, "crawl": 7},
}
info = _parse_tavily_usage(data)
assert info.provider == "tavily"
assert info.supported is True
assert info.used == 142
assert info.limit == 1000
assert info.remaining == 858
assert info.reset_date == "2026-05-01"
assert info.search_used == 120
assert info.extract_used == 15
assert info.crawl_used == 7
def test_remaining_computed_when_missing(self):
data = {"used": 300, "limit": 1000}
info = _parse_tavily_usage(data)
assert info.remaining == 700
def test_remaining_not_negative(self):
data = {"used": 1100, "limit": 1000}
info = _parse_tavily_usage(data)
assert info.remaining == 0
def test_camel_case_reset_date(self):
data = {"used": 10, "limit": 100, "resetDate": "2026-06-01"}
info = _parse_tavily_usage(data)
assert info.reset_date == "2026-06-01"
def test_empty_response(self):
info = _parse_tavily_usage({})
assert info.provider == "tavily"
assert info.supported is True
assert info.used is None
assert info.limit is None
def test_no_breakdown_key(self):
data = {"used": 5, "limit": 50}
info = _parse_tavily_usage(data)
assert info.search_used is None
assert info.extract_used is None
assert info.crawl_used is None
# ---------------------------------------------------------------------------
# fetch_search_usage routing tests
# ---------------------------------------------------------------------------
class TestFetchSearchUsageRouting:
@pytest.mark.asyncio
async def test_duckduckgo_returns_unsupported(self):
info = await fetch_search_usage("duckduckgo")
assert info.provider == "duckduckgo"
assert info.supported is False
@pytest.mark.asyncio
async def test_searxng_returns_unsupported(self):
info = await fetch_search_usage("searxng")
assert info.supported is False
@pytest.mark.asyncio
async def test_jina_returns_unsupported(self):
info = await fetch_search_usage("jina")
assert info.supported is False
@pytest.mark.asyncio
async def test_brave_returns_unsupported(self):
info = await fetch_search_usage("brave")
assert info.provider == "brave"
assert info.supported is False
@pytest.mark.asyncio
async def test_unknown_provider_returns_unsupported(self):
info = await fetch_search_usage("some_unknown_provider")
assert info.supported is False
@pytest.mark.asyncio
async def test_tavily_no_api_key_returns_error(self):
with patch.dict("os.environ", {}, clear=True):
# Ensure TAVILY_API_KEY is not set
import os
os.environ.pop("TAVILY_API_KEY", None)
info = await fetch_search_usage("tavily", api_key=None)
assert info.provider == "tavily"
assert info.supported is True
assert info.error is not None
assert "not configured" in info.error
@pytest.mark.asyncio
async def test_tavily_success(self):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"used": 142,
"limit": 1000,
"remaining": 858,
"reset_date": "2026-05-01",
"breakdown": {"search": 120, "extract": 15, "crawl": 7},
}
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.get = AsyncMock(return_value=mock_response)
with patch("httpx.AsyncClient", return_value=mock_client):
info = await fetch_search_usage("tavily", api_key="test-key")
assert info.provider == "tavily"
assert info.supported is True
assert info.error is None
assert info.used == 142
assert info.limit == 1000
assert info.remaining == 858
assert info.reset_date == "2026-05-01"
assert info.search_used == 120
@pytest.mark.asyncio
async def test_tavily_http_error(self):
import httpx
mock_response = MagicMock()
mock_response.status_code = 401
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
"401", request=MagicMock(), response=mock_response
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.get = AsyncMock(return_value=mock_response)
with patch("httpx.AsyncClient", return_value=mock_client):
info = await fetch_search_usage("tavily", api_key="bad-key")
assert info.supported is True
assert info.error == "HTTP 401"
@pytest.mark.asyncio
async def test_tavily_network_error(self):
import httpx
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.get = AsyncMock(side_effect=httpx.ConnectError("timeout"))
with patch("httpx.AsyncClient", return_value=mock_client):
info = await fetch_search_usage("tavily", api_key="test-key")
assert info.supported is True
assert info.error is not None
@pytest.mark.asyncio
async def test_provider_name_case_insensitive(self):
info = await fetch_search_usage("Tavily", api_key=None)
assert info.provider == "tavily"
assert info.supported is True
# ---------------------------------------------------------------------------
# build_status_content integration tests
# ---------------------------------------------------------------------------
class TestBuildStatusContentWithSearchUsage:
_BASE_KWARGS = dict(
version="0.1.0",
model="claude-opus-4-5",
start_time=1_000_000.0,
last_usage={"prompt_tokens": 1000, "completion_tokens": 200},
context_window_tokens=65536,
session_msg_count=5,
context_tokens_estimate=3000,
)
def test_no_search_usage_unchanged(self):
"""Omitting search_usage_text keeps existing behaviour."""
content = build_status_content(**self._BASE_KWARGS)
assert "🔍" not in content
assert "Web Search" not in content
def test_search_usage_none_unchanged(self):
content = build_status_content(**self._BASE_KWARGS, search_usage_text=None)
assert "🔍" not in content
def test_search_usage_appended(self):
usage_text = "🔍 Web Search: tavily\n Usage: 142 / 1000 requests"
content = build_status_content(**self._BASE_KWARGS, search_usage_text=usage_text)
assert "🔍 Web Search: tavily" in content
assert "142 / 1000" in content
def test_existing_fields_still_present(self):
usage_text = "🔍 Web Search: duckduckgo\n Usage tracking: not available"
content = build_status_content(**self._BASE_KWARGS, search_usage_text=usage_text)
# Original fields must still be present
assert "nanobot v0.1.0" in content
assert "claude-opus-4-5" in content
assert "1000 in / 200 out" in content
# New field appended
assert "duckduckgo" in content
def test_full_tavily_in_status(self):
info = SearchUsageInfo(
provider="tavily",
supported=True,
used=142,
limit=1000,
remaining=858,
reset_date="2026-05-01",
search_used=120,
extract_used=15,
crawl_used=7,
)
content = build_status_content(**self._BASE_KWARGS, search_usage_text=info.format())
assert "142 / 1000" in content
assert "858" in content
assert "2026-05-01" in content
assert "Search: 120" in content