mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
289 lines
10 KiB
Python
289 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from plyngent.agent import ReasoningDeltaEvent, TextDeltaEvent, ToolCallEvent, ToolResultEvent
|
|
from plyngent.cli.display import (
|
|
get_markdown_enabled,
|
|
markdown_render_available,
|
|
print_markdown,
|
|
render_events,
|
|
set_markdown_enabled,
|
|
set_verbose_tool_results,
|
|
)
|
|
from plyngent.lmproto.openai_compatible.model import (
|
|
AssistantFunctionTool,
|
|
AssistantFunctionToolCall,
|
|
ToolChatMessage,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator
|
|
|
|
import pytest
|
|
|
|
from plyngent.agent import AgentEvent
|
|
|
|
|
|
async def _aiter(events: list[AgentEvent]) -> AsyncIterator[AgentEvent]:
|
|
for event in events:
|
|
yield event
|
|
|
|
|
|
async def test_render_reasoning_and_text(capsys: pytest.CaptureFixture[str]) -> None:
|
|
set_markdown_enabled(False)
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
ReasoningDeltaEvent(content="think"),
|
|
TextDeltaEvent(content="hello"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "reasoning:" in out
|
|
assert "think" in out
|
|
assert "assistant:" in out
|
|
assert "hello" in out
|
|
# Labels on their own lines (content begins after newline).
|
|
assert "assistant:\nhello" in out or "assistant:\r\nhello" in out
|
|
set_markdown_enabled(True)
|
|
|
|
|
|
async def test_flush_markdown_on_source_change(
|
|
capsys: pytest.CaptureFixture[str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Assistant segment before tools is flushed so later text is a new segment."""
|
|
monkeypatch.setattr("plyngent.cli.display.markdown_render_available", lambda: True)
|
|
set_markdown_enabled(True)
|
|
from plyngent.agent import ToolCallEvent
|
|
from plyngent.lmproto.openai_compatible.model import (
|
|
AssistantFunctionTool,
|
|
AssistantFunctionToolCall,
|
|
)
|
|
|
|
call = AssistantFunctionToolCall(
|
|
id="1",
|
|
function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["echo", "x"]}'),
|
|
)
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
TextDeltaEvent(content="before **tool**"),
|
|
ToolCallEvent(tool_call=call),
|
|
TextDeltaEvent(content="after"),
|
|
]
|
|
),
|
|
markdown=True,
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "[tool]" in out
|
|
assert "after" in out
|
|
|
|
|
|
def _read_call(args_json: str) -> ToolCallEvent:
|
|
return ToolCallEvent(
|
|
tool_call=AssistantFunctionToolCall(
|
|
id="1",
|
|
function=AssistantFunctionTool(name="read_file", arguments=args_json),
|
|
)
|
|
)
|
|
|
|
|
|
def _result(content: str) -> ToolResultEvent:
|
|
return ToolResultEvent(message=ToolChatMessage(content=content, tool_call_id="1"))
|
|
|
|
|
|
async def test_pretty_read_file_done(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\ntwo\n")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Read 'a.txt' L1-4 (done)" in out
|
|
assert "[tool]" not in out
|
|
assert "[tool ok]" not in out
|
|
|
|
|
|
async def test_pretty_read_file_statuses(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cases = [
|
|
("error: file not found: a.txt", "(file not found)"),
|
|
("error: not a file: a.txt", "(not a file)"),
|
|
("error: something else", "(error)"),
|
|
("no header content", "(done)"),
|
|
]
|
|
for content, status in cases:
|
|
await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result(content)]))
|
|
out = capsys.readouterr().out
|
|
assert f"* Read 'a.txt' {status}" in out, f"status {status} not rendered for {content!r}"
|
|
|
|
|
|
def _tree_call(args_json: str) -> ToolCallEvent:
|
|
return ToolCallEvent(
|
|
tool_call=AssistantFunctionToolCall(
|
|
id="1",
|
|
function=AssistantFunctionTool(name="tree", arguments=args_json),
|
|
)
|
|
)
|
|
|
|
|
|
async def test_pretty_tree_markdown(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "- src/\n - main.py\n - nested/\n - deep.txt\n- a.txt\n"
|
|
await render_events(_aiter([_tree_call('{"path": "."}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree '.' (2 dirs, 3 files, depth 3)" in out
|
|
assert "[tool]" not in out
|
|
|
|
|
|
async def test_pretty_tree_flat_omits_depth(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "src/\nsrc/main.py\na.txt\n… (5 more paths not shown)"
|
|
await render_events(_aiter([_tree_call('{"path": ".", "format": "flat"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree '.' (1 dirs, 2 files)" in out
|
|
assert "depth" not in out
|
|
|
|
|
|
async def test_pretty_tree_decorated(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "src/\n├── main.py\n└── nested/\n └── deep.txt\n"
|
|
await render_events(_aiter([_tree_call('{"path": "src", "format": "decorated"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree 'src' (1 dirs, 2 files, depth 2)" in out
|
|
|
|
|
|
async def test_pretty_tree_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_tree_call('{"path": "nope"}'), _result("error: not a directory: nope")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree 'nope' (error)" in out
|
|
|
|
|
|
async def test_pretty_todo_push(capsys: pytest.CaptureFixture[str]) -> None:
|
|
call = ToolCallEvent(
|
|
tool_call=AssistantFunctionToolCall(
|
|
id="2",
|
|
function=AssistantFunctionTool(name="todo_push", arguments='{"titles": ["T1"]}'),
|
|
)
|
|
)
|
|
result = (
|
|
"pushed group (depth=1) items=[a1]\n"
|
|
"(LIFO of groups: depth=1; TOP group = current breakdown level)\n"
|
|
"group d=0 TOP:\n"
|
|
" [ ] a1: T1"
|
|
)
|
|
await render_events(_aiter([call, _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Todo Push:" in out
|
|
assert " [ ] a1: T1" in out
|
|
assert "[tool]" not in out
|
|
|
|
|
|
async def test_non_pretty_tool_keeps_old_style(capsys: pytest.CaptureFixture[str]) -> None:
|
|
call = ToolCallEvent(
|
|
tool_call=AssistantFunctionToolCall(
|
|
id="3",
|
|
function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["ls"]}'),
|
|
)
|
|
)
|
|
await render_events(_aiter([call, _result("exit_code=0")]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool] run_argv(" in out
|
|
assert "[tool ok]" in out
|
|
|
|
|
|
async def test_tool_result_preview_vs_verbose(capsys: pytest.CaptureFixture[str]) -> None:
|
|
long = "x" * 200
|
|
msg = ToolChatMessage(content=long, tool_call_id="1")
|
|
set_verbose_tool_results(False)
|
|
set_markdown_enabled(False)
|
|
await render_events(_aiter([ToolResultEvent(message=msg)]))
|
|
out = capsys.readouterr().out
|
|
assert "…" in out
|
|
assert long not in out
|
|
|
|
set_verbose_tool_results(True)
|
|
await render_events(_aiter([ToolResultEvent(message=msg)]), verbose=True)
|
|
out2 = capsys.readouterr().out
|
|
assert long in out2
|
|
set_verbose_tool_results(False)
|
|
set_markdown_enabled(True)
|
|
|
|
|
|
async def test_markdown_off_keeps_plain_stream(
|
|
capsys: pytest.CaptureFixture[str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr("plyngent.cli.display.markdown_render_available", lambda: True)
|
|
set_markdown_enabled(False)
|
|
await render_events(_aiter([TextDeltaEvent(content="**bold**")]))
|
|
out = capsys.readouterr().out
|
|
assert "**bold**" in out
|
|
set_markdown_enabled(True)
|
|
|
|
|
|
async def test_markdown_on_replaces_with_rich(
|
|
capsys: pytest.CaptureFixture[str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr("plyngent.cli.display.markdown_render_available", lambda: True)
|
|
set_markdown_enabled(True)
|
|
await render_events(_aiter([TextDeltaEvent(content="hello **world**")]), markdown=True)
|
|
out = capsys.readouterr().out
|
|
# Rich markdown renders emphasis; raw ** markers should not remain as the sole form.
|
|
assert "assistant:" in out or "assistant: " in out
|
|
assert "world" in out
|
|
|
|
|
|
def test_print_markdown_renders(capsys: pytest.CaptureFixture[str]) -> None:
|
|
print_markdown("# Title\n\n`code`", label="assistant: ")
|
|
out = capsys.readouterr().out
|
|
assert "Title" in out
|
|
assert "code" in out
|
|
|
|
|
|
def test_markdown_flags_roundtrip() -> None:
|
|
set_markdown_enabled(False)
|
|
assert get_markdown_enabled() is False
|
|
set_markdown_enabled(True)
|
|
assert get_markdown_enabled() is True
|
|
|
|
|
|
def test_markdown_render_available_respects_plain_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("PLYNGENT_PLAIN", "1")
|
|
assert markdown_render_available() is False
|
|
monkeypatch.delenv("PLYNGENT_PLAIN", raising=False)
|
|
|
|
|
|
async def test_pretty_verbose_shows_full_result(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Verbose mode prints the full tool result even for pretty tools."""
|
|
await render_events(
|
|
_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\ntwo\n")]),
|
|
verbose=True,
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok]" in out
|
|
assert "L1-4\none\ntwo" in out
|
|
assert "* Read" not in out
|
|
|
|
|
|
async def test_pretty_plain_output_has_no_ansi(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Pretty summaries are plain text when stdout is not a TTY."""
|
|
await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\n")]))
|
|
out = capsys.readouterr().out
|
|
assert "\x1b[" not in out
|
|
|
|
|
|
async def test_tool_result_preview_first_line_and_count(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Non-verbose preview shows the first line plus a line-count tail."""
|
|
set_markdown_enabled(False)
|
|
content = "first line\nsecond\nthird"
|
|
await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content=content, tool_call_id="1"))]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok] first line (…2 more lines)" in out
|
|
|
|
await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content="only line", tool_call_id="2"))]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok] only line" in out
|
|
assert "more" not in out
|
|
|
|
await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content="", tool_call_id="3"))]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok] (no output)" in out
|
|
set_markdown_enabled(True)
|