Xi Zhang
343 lines
11 KiB
Python
343 lines
11 KiB
Python
"""Tests for EvoScientist/stream/utils.py pure functions."""
|
|
|
|
from EvoScientist.stream.utils import (
|
|
_shorten_path,
|
|
count_lines,
|
|
format_tool_compact,
|
|
format_tool_compact_with_result,
|
|
has_args,
|
|
is_success,
|
|
truncate,
|
|
truncate_with_line_hint,
|
|
)
|
|
|
|
# === is_success ===
|
|
|
|
|
|
class TestIsSuccess:
|
|
def test_ok_prefix(self):
|
|
assert is_success("[OK] all good") is True
|
|
|
|
def test_failed_prefix(self):
|
|
assert is_success("[FAILED] bad") is False
|
|
|
|
def test_traceback(self):
|
|
assert is_success("Traceback (most recent call last)\n File ...") is False
|
|
|
|
def test_exception(self):
|
|
assert is_success("Exception: something went wrong") is False
|
|
|
|
def test_error(self):
|
|
assert is_success("Error: file not found") is False
|
|
|
|
def test_clean_output(self):
|
|
assert is_success("file1.py\nfile2.py") is True
|
|
|
|
def test_whitespace_stripped(self):
|
|
assert is_success(" [OK] with spaces ") is True
|
|
|
|
def test_error_in_code_content_not_false_positive(self):
|
|
# read_file returning code with "Error:" deep inside should be success
|
|
content = '#!/usr/bin/env python3\n"""\nSkill Packager\n"""\n\nprint(f"Error: not found")'
|
|
assert is_success(content) is True
|
|
|
|
def test_error_on_line4_not_false_positive(self):
|
|
content = "line1\nline2\nline3\nError: buried deep\nline5"
|
|
assert is_success(content) is True
|
|
|
|
def test_error_on_first_line(self):
|
|
assert is_success("Error: file not found\nsome detail") is False
|
|
|
|
def test_error_invoking_tool(self):
|
|
assert is_success("Error invoking tool 'write_file'") is False
|
|
|
|
def test_failed_to_uninstall(self):
|
|
assert (
|
|
is_success("Failed to uninstall skill: Skill not found: latex-paper-en")
|
|
is False
|
|
)
|
|
|
|
def test_failed_to_install(self):
|
|
assert is_success("Failed to install skill: git clone failed: ...") is False
|
|
|
|
def test_failed_in_code_content_not_false_positive(self):
|
|
content = '#!/usr/bin/env python3\n"""Helper"""\n\nif x:\n print("Failed to connect")'
|
|
assert is_success(content) is True
|
|
|
|
|
|
# === format_tool_compact ===
|
|
|
|
|
|
class TestFormatToolCompact:
|
|
def test_no_args(self):
|
|
assert format_tool_compact("execute", None) == "execute()"
|
|
assert format_tool_compact("execute", {}) == "execute()"
|
|
|
|
def test_execute(self):
|
|
result = format_tool_compact("execute", {"command": "ls -la"})
|
|
assert result == "execute(ls -la)"
|
|
|
|
def test_execute_long_command(self):
|
|
long_cmd = "x" * 60
|
|
result = format_tool_compact("execute", {"command": long_cmd})
|
|
assert len(result) < 70
|
|
assert result.endswith("\u2026)")
|
|
|
|
def test_read_file(self):
|
|
result = format_tool_compact("read_file", {"path": "src/main.py"})
|
|
assert result == "read_file(src/main.py)"
|
|
|
|
def test_write_file(self):
|
|
result = format_tool_compact("write_file", {"path": "out.txt"})
|
|
assert result == "write_file(out.txt)"
|
|
|
|
def test_edit_file(self):
|
|
result = format_tool_compact("edit_file", {"path": "f.py"})
|
|
assert result == "edit_file(f.py)"
|
|
|
|
# Global profile memory display (/memories/ = global)
|
|
def test_read_file_global_memory(self):
|
|
result = format_tool_compact(
|
|
"read_file", {"path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Reading memory"
|
|
|
|
def test_read_file_global_memory_file_path_alias(self):
|
|
result = format_tool_compact(
|
|
"read_file", {"file_path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Reading memory"
|
|
|
|
def test_read_file_any_global_memory_file(self):
|
|
result = format_tool_compact("read_file", {"path": "/memories/history.md"})
|
|
assert result == "Reading memory"
|
|
|
|
def test_write_file_global_memory(self):
|
|
result = format_tool_compact(
|
|
"write_file", {"path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Updating memory"
|
|
|
|
def test_edit_file_global_memory(self):
|
|
result = format_tool_compact(
|
|
"edit_file", {"path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Updating memory"
|
|
|
|
def test_write_edit_any_global_memory_file(self):
|
|
write_result = format_tool_compact("write_file", {"path": "/memories/soul.md"})
|
|
edit_result = format_tool_compact(
|
|
"edit_file", {"path": "/memories/skills-context.md"}
|
|
)
|
|
assert write_result == "Updating memory"
|
|
assert edit_result == "Updating memory"
|
|
|
|
# Project-local /memory/ files show normal tool display
|
|
def test_read_file_project_memory(self):
|
|
result = format_tool_compact(
|
|
"read_file", {"path": "/memory/ideation-memory.md"}
|
|
)
|
|
assert result == "read_file(/memory/ideation-memory.md)"
|
|
|
|
def test_edit_file_project_memory(self):
|
|
result = format_tool_compact(
|
|
"edit_file", {"path": "/memory/experiment-memory.md"}
|
|
)
|
|
assert result == "edit_file(/memory/experiment-memory.md)"
|
|
|
|
def test_memory_display_inferred_from_result_when_args_sparse(self):
|
|
read_result = format_tool_compact_with_result(
|
|
"read_file",
|
|
{},
|
|
"# User profile\n\nFounder: Zachary",
|
|
)
|
|
assert read_result == "Reading memory"
|
|
|
|
edit_result = format_tool_compact_with_result(
|
|
"edit_file",
|
|
{},
|
|
"Successfully replaced 1 instance(s) of the string in '/memories/profile/USER_PROFILE.md'",
|
|
)
|
|
assert edit_result == "Updating memory"
|
|
|
|
write_result = format_tool_compact_with_result(
|
|
"write_file",
|
|
{},
|
|
"Wrote updated content to '/memories/history.md'",
|
|
)
|
|
assert write_result == "Updating memory"
|
|
|
|
def test_profile_memory_inference_uses_profile_template_headings(self, monkeypatch):
|
|
from EvoScientist.middleware import memory
|
|
from EvoScientist.stream import utils
|
|
|
|
monkeypatch.setitem(
|
|
memory.PROFILE_TEMPLATES,
|
|
"/profile/CUSTOM.md",
|
|
"# Custom profile\n\n## Notes\n",
|
|
)
|
|
utils._profile_memory_headings.cache_clear()
|
|
|
|
try:
|
|
result = format_tool_compact_with_result(
|
|
"read_file",
|
|
{},
|
|
"# Custom profile\n\n- remembered",
|
|
)
|
|
finally:
|
|
utils._profile_memory_headings.cache_clear()
|
|
|
|
assert result == "Reading memory"
|
|
|
|
def test_profile_memory_headings_skip_user_profile_frontmatter(self):
|
|
from EvoScientist.stream import utils
|
|
|
|
utils._profile_memory_headings.cache_clear()
|
|
try:
|
|
headings = utils._profile_memory_headings()
|
|
finally:
|
|
utils._profile_memory_headings.cache_clear()
|
|
|
|
assert "# User profile" in headings
|
|
assert not any(heading.startswith("---") for heading in headings)
|
|
|
|
def test_project_memory_result_not_special(self):
|
|
result = format_tool_compact_with_result(
|
|
"write_file",
|
|
{},
|
|
"Wrote updated content to '/memory/ideation-memory.md'",
|
|
)
|
|
assert result != "Updating memory"
|
|
|
|
def test_glob(self):
|
|
result = format_tool_compact("glob", {"pattern": "*.py"})
|
|
assert result == "glob(*.py)"
|
|
|
|
def test_grep(self):
|
|
result = format_tool_compact("grep", {"pattern": "TODO", "path": "src/"})
|
|
assert result == "grep(TODO, src/)"
|
|
|
|
def test_ls(self):
|
|
assert format_tool_compact("ls", {"path": "/src"}) == "ls(/src)"
|
|
|
|
def test_write_todos_list(self):
|
|
todos = [{"status": "todo", "content": "a"}, {"status": "todo", "content": "b"}]
|
|
result = format_tool_compact("write_todos", {"todos": todos})
|
|
assert result == "write_todos(2 items)"
|
|
|
|
def test_write_todos_non_list(self):
|
|
result = format_tool_compact("write_todos", {"todos": "something"})
|
|
assert result == "write_todos(...)"
|
|
|
|
def test_read_todos(self):
|
|
assert format_tool_compact("read_todos", {}) == "read_todos()"
|
|
|
|
def test_task_with_type_and_desc(self):
|
|
result = format_tool_compact(
|
|
"task", {"subagent_type": "research-agent", "description": "Find papers"}
|
|
)
|
|
assert "Cooking with research-agent" in result
|
|
assert "Find papers" in result
|
|
|
|
def test_task_with_type_only(self):
|
|
result = format_tool_compact("task", {"subagent_type": "code-agent"})
|
|
assert result == "Cooking with code-agent"
|
|
|
|
def test_task_with_desc_only(self):
|
|
result = format_tool_compact("task", {"description": "do stuff"})
|
|
assert result == "task(description=do stuff)"
|
|
|
|
def test_task_no_info(self):
|
|
result = format_tool_compact("task", {"other": "value"})
|
|
assert result == "task(other=value)"
|
|
|
|
def test_tavily_search(self):
|
|
result = format_tool_compact("tavily_search", {"query": "python testing"})
|
|
assert result == "tavily_search(python testing)"
|
|
|
|
def test_think_tool(self):
|
|
result = format_tool_compact("think_tool", {"reflection": "need more data"})
|
|
assert result == "think_tool(need more data)"
|
|
|
|
def test_unknown_tool(self):
|
|
result = format_tool_compact("custom_tool", {"key": "value"})
|
|
assert "custom_tool(" in result
|
|
assert "key=value" in result
|
|
|
|
def test_unknown_tool_long_value(self):
|
|
result = format_tool_compact("custom_tool", {"key": "a" * 30})
|
|
assert "\u2026" in result
|
|
|
|
|
|
# === truncate ===
|
|
|
|
|
|
class TestTruncate:
|
|
def test_within_limit(self):
|
|
assert truncate("hello", 10) == "hello"
|
|
|
|
def test_at_limit(self):
|
|
assert truncate("hello", 5) == "hello"
|
|
|
|
def test_over_limit(self):
|
|
result = truncate("hello world", 5)
|
|
assert result.startswith("hello")
|
|
assert "truncated" in result
|
|
|
|
|
|
# === _shorten_path ===
|
|
|
|
|
|
class TestShortenPath:
|
|
def test_short_path(self):
|
|
assert _shorten_path("src/main.py") == "src/main.py"
|
|
|
|
def test_long_path(self):
|
|
long_path = "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r.py"
|
|
result = _shorten_path(long_path, max_len=20)
|
|
assert result.startswith(".../")
|
|
assert result.endswith("r.py")
|
|
|
|
|
|
# === has_args ===
|
|
|
|
|
|
class TestHasArgs:
|
|
def test_none(self):
|
|
assert has_args(None) is False
|
|
|
|
def test_empty_dict(self):
|
|
assert has_args({}) is False
|
|
|
|
def test_non_empty(self):
|
|
assert has_args({"key": "val"}) is True
|
|
|
|
|
|
# === count_lines ===
|
|
|
|
|
|
class TestCountLines:
|
|
def test_empty(self):
|
|
assert count_lines("") == 0
|
|
|
|
def test_single_line(self):
|
|
assert count_lines("hello") == 1
|
|
|
|
def test_multi_line(self):
|
|
assert count_lines("a\nb\nc") == 3
|
|
|
|
|
|
# === truncate_with_line_hint ===
|
|
|
|
|
|
class TestTruncateWithLineHint:
|
|
def test_within_limit(self):
|
|
text, remaining = truncate_with_line_hint("a\nb\nc", max_lines=5)
|
|
assert remaining == 0
|
|
assert "a" in text
|
|
|
|
def test_over_limit(self):
|
|
text, remaining = truncate_with_line_hint("a\nb\nc\nd\ne\nf", max_lines=3)
|
|
assert remaining == 3
|
|
assert "d" not in text
|