banteg
409 lines
12 KiB
Python
409 lines
12 KiB
Python
from typing import cast
|
|
from types import SimpleNamespace
|
|
from pathlib import Path
|
|
|
|
from takopi.markdown import (
|
|
HARD_BREAK,
|
|
MarkdownFormatter,
|
|
STATUS,
|
|
action_status,
|
|
assemble_markdown_parts,
|
|
format_elapsed,
|
|
format_file_change_title,
|
|
render_event_cli,
|
|
shorten,
|
|
)
|
|
from takopi.model import Action, ActionEvent, ResumeToken, StartedEvent, TakopiEvent
|
|
from takopi.progress import ProgressTracker
|
|
from takopi.telegram.render import render_markdown
|
|
from takopi.utils.paths import reset_run_base_dir, set_run_base_dir
|
|
from tests.factories import (
|
|
action_completed,
|
|
action_started,
|
|
session_started,
|
|
)
|
|
|
|
|
|
def _format_resume(token) -> str:
|
|
return f"`codex resume {token.value}`"
|
|
|
|
|
|
SAMPLE_EVENTS: list[TakopiEvent] = [
|
|
session_started("codex", "0199a213-81c0-7800-8aa1-bbab2a035a53", title="Codex"),
|
|
action_started("a-1", "command", "bash -lc ls"),
|
|
action_completed(
|
|
"a-1",
|
|
"command",
|
|
"bash -lc ls",
|
|
ok=True,
|
|
detail={"exit_code": 0},
|
|
),
|
|
action_completed("a-2", "note", "Checking repository root for README", ok=True),
|
|
]
|
|
|
|
|
|
def test_render_event_cli_sample_events() -> None:
|
|
out: list[str] = []
|
|
for evt in SAMPLE_EVENTS:
|
|
out.extend(render_event_cli(evt))
|
|
|
|
assert out == [
|
|
"codex",
|
|
"▸ `bash -lc ls`",
|
|
"✓ `bash -lc ls`",
|
|
"✓ Checking repository root for README",
|
|
]
|
|
|
|
|
|
def test_render_event_cli_handles_action_kinds() -> None:
|
|
events: list[TakopiEvent] = [
|
|
action_completed(
|
|
"c-1", "command", "pytest -q", ok=False, detail={"exit_code": 1}
|
|
),
|
|
action_completed(
|
|
"s-1",
|
|
"web_search",
|
|
"python jsonlines parser handle unknown fields",
|
|
ok=True,
|
|
),
|
|
action_completed("t-1", "tool", "github.search_issues", ok=True),
|
|
action_completed(
|
|
"f-1",
|
|
"file_change",
|
|
"2 files",
|
|
ok=True,
|
|
detail={
|
|
"changes": [
|
|
{"path": "README.md", "kind": "add"},
|
|
{"path": "src/compute_answer.py", "kind": "update"},
|
|
]
|
|
},
|
|
),
|
|
action_completed("n-1", "note", "stream error", ok=False),
|
|
]
|
|
|
|
out: list[str] = []
|
|
for evt in events:
|
|
out.extend(render_event_cli(evt))
|
|
|
|
assert any(line.startswith("✗ `pytest -q` (exit 1)") for line in out)
|
|
assert any(
|
|
"searched: python jsonlines parser handle unknown fields" in line
|
|
for line in out
|
|
)
|
|
assert any("tool: github.search_issues" in line for line in out)
|
|
assert any(
|
|
"files: add `README.md`, update `src/compute_answer.py`" in line for line in out
|
|
)
|
|
assert any(line.startswith("✗ stream error") for line in out)
|
|
|
|
|
|
def test_file_change_renders_relative_paths_inside_cwd() -> None:
|
|
readme_abs = str(Path.cwd() / "README.md")
|
|
weird_abs = "~" + readme_abs
|
|
out = render_event_cli(
|
|
action_completed(
|
|
"f-abs",
|
|
"file_change",
|
|
"README.md",
|
|
ok=True,
|
|
detail={
|
|
"changes": [
|
|
{"path": readme_abs, "kind": "update"},
|
|
{"path": weird_abs, "kind": "update"},
|
|
]
|
|
},
|
|
)
|
|
)
|
|
assert any(
|
|
f"files: update `README.md`, update `{weird_abs}`" in line for line in out
|
|
)
|
|
|
|
|
|
def test_file_change_renders_change_objects(tmp_path: Path) -> None:
|
|
base = tmp_path / "repo"
|
|
base.mkdir()
|
|
abs_path = str(base / "changelog.md")
|
|
token = set_run_base_dir(base)
|
|
try:
|
|
out = render_event_cli(
|
|
action_completed(
|
|
"f-obj",
|
|
"file_change",
|
|
"ignored",
|
|
ok=True,
|
|
detail={"changes": [SimpleNamespace(path=abs_path, kind="update")]},
|
|
)
|
|
)
|
|
finally:
|
|
reset_run_base_dir(token)
|
|
assert any("files: update `changelog.md`" in line for line in out)
|
|
|
|
|
|
def test_file_change_title_relativizes_absolute_title(tmp_path: Path) -> None:
|
|
base = tmp_path / "repo"
|
|
base.mkdir()
|
|
abs_path = str(base / "changelog.md")
|
|
token = set_run_base_dir(base)
|
|
try:
|
|
out = render_event_cli(
|
|
action_completed("f-abs", "file_change", abs_path, ok=True)
|
|
)
|
|
finally:
|
|
reset_run_base_dir(token)
|
|
assert any("files: `changelog.md`" in line for line in out)
|
|
|
|
|
|
def test_progress_renderer_renders_progress_and_final() -> None:
|
|
tracker = ProgressTracker(engine="codex")
|
|
for evt in SAMPLE_EVENTS:
|
|
tracker.note_event(evt)
|
|
|
|
state = tracker.snapshot(resume_formatter=_format_resume)
|
|
formatter = MarkdownFormatter(max_actions=5)
|
|
progress_parts = formatter.render_progress_parts(state, elapsed_s=3.0)
|
|
progress = assemble_markdown_parts(progress_parts)
|
|
assert progress.startswith("working · codex · 3s · step 2")
|
|
assert "✓ `bash -lc ls`" in progress
|
|
assert "`codex resume 0199a213-81c0-7800-8aa1-bbab2a035a53`" in progress
|
|
|
|
final_parts = formatter.render_final_parts(
|
|
state, elapsed_s=3.0, status="done", answer="answer"
|
|
)
|
|
final = assemble_markdown_parts(final_parts)
|
|
assert final.startswith("done · codex · 3s · step 2")
|
|
assert "✓ `bash -lc ls`" not in final
|
|
assert "Checking repository root for README" not in final
|
|
assert "answer" in final
|
|
assert final.rstrip().endswith(
|
|
"`codex resume 0199a213-81c0-7800-8aa1-bbab2a035a53`"
|
|
)
|
|
|
|
|
|
def test_progress_renderer_footer_includes_ctx_before_resume() -> None:
|
|
tracker = ProgressTracker(engine="codex")
|
|
for evt in SAMPLE_EVENTS:
|
|
tracker.note_event(evt)
|
|
|
|
state = tracker.snapshot(
|
|
resume_formatter=_format_resume,
|
|
context_line="`ctx: z80 @ feat/name`",
|
|
)
|
|
formatter = MarkdownFormatter(max_actions=5)
|
|
parts = formatter.render_progress_parts(state, elapsed_s=0.0)
|
|
assert parts.footer == (
|
|
"`ctx: z80 @ feat/name`"
|
|
f"{HARD_BREAK}`codex resume 0199a213-81c0-7800-8aa1-bbab2a035a53`"
|
|
)
|
|
|
|
|
|
def test_progress_renderer_clamps_actions_and_ignores_unknown() -> None:
|
|
tracker = ProgressTracker(engine="codex")
|
|
events = [
|
|
action_completed(
|
|
f"item_{i}",
|
|
"command",
|
|
f"echo {i}",
|
|
ok=True,
|
|
detail={"exit_code": 0},
|
|
)
|
|
for i in range(6)
|
|
]
|
|
|
|
for evt in events:
|
|
assert tracker.note_event(evt) is True
|
|
|
|
state = tracker.snapshot()
|
|
formatter = MarkdownFormatter(max_actions=3, command_width=20)
|
|
parts = formatter.render_progress_parts(state, elapsed_s=0.0)
|
|
lines = parts.body.split(HARD_BREAK) if parts.body else []
|
|
assert len(lines) == 3
|
|
assert "echo 3" in lines[0]
|
|
assert "echo 5" in lines[-1]
|
|
mystery = SimpleNamespace(type="mystery")
|
|
assert tracker.note_event(cast(TakopiEvent, mystery)) is False
|
|
|
|
|
|
def test_progress_renderer_renders_commands_in_markdown() -> None:
|
|
tracker = ProgressTracker(engine="codex")
|
|
for i in (30, 31, 32):
|
|
tracker.note_event(
|
|
action_completed(
|
|
f"item_{i}",
|
|
"command",
|
|
f"echo {i}",
|
|
ok=True,
|
|
detail={"exit_code": 0},
|
|
)
|
|
)
|
|
|
|
state = tracker.snapshot()
|
|
formatter = MarkdownFormatter(max_actions=5, command_width=None)
|
|
md = assemble_markdown_parts(formatter.render_progress_parts(state, elapsed_s=0.0))
|
|
text, _ = render_markdown(md)
|
|
assert "✓ echo 30" in text
|
|
assert "✓ echo 31" in text
|
|
assert "✓ echo 32" in text
|
|
|
|
|
|
def test_progress_renderer_handles_duplicate_action_ids() -> None:
|
|
tracker = ProgressTracker(engine="codex")
|
|
events = [
|
|
action_started("dup", "command", "echo first"),
|
|
action_completed(
|
|
"dup",
|
|
"command",
|
|
"echo first",
|
|
ok=True,
|
|
detail={"exit_code": 0},
|
|
),
|
|
action_started("dup", "command", "echo second"),
|
|
action_completed(
|
|
"dup",
|
|
"command",
|
|
"echo second",
|
|
ok=True,
|
|
detail={"exit_code": 0},
|
|
),
|
|
]
|
|
|
|
for evt in events:
|
|
assert tracker.note_event(evt) is True
|
|
|
|
state = tracker.snapshot()
|
|
formatter = MarkdownFormatter(max_actions=5)
|
|
parts = formatter.render_progress_parts(state, elapsed_s=0.0)
|
|
lines = parts.body.split(HARD_BREAK) if parts.body else []
|
|
assert len(lines) == 1
|
|
assert lines[0].startswith("✓ ")
|
|
assert "echo second" in lines[0]
|
|
|
|
|
|
def test_progress_renderer_collapses_action_updates() -> None:
|
|
tracker = ProgressTracker(engine="codex")
|
|
events = [
|
|
action_started("a-1", "command", "echo one"),
|
|
action_started("a-1", "command", "echo two"),
|
|
action_completed(
|
|
"a-1",
|
|
"command",
|
|
"echo two",
|
|
ok=True,
|
|
detail={"exit_code": 0},
|
|
),
|
|
]
|
|
|
|
for evt in events:
|
|
assert tracker.note_event(evt) is True
|
|
|
|
assert tracker.action_count == 1
|
|
state = tracker.snapshot()
|
|
formatter = MarkdownFormatter(max_actions=5)
|
|
parts = formatter.render_progress_parts(state, elapsed_s=0.0)
|
|
lines = parts.body.split(HARD_BREAK) if parts.body else []
|
|
assert len(lines) == 1
|
|
assert lines[0].startswith("✓ ")
|
|
assert "echo two" in lines[0]
|
|
|
|
|
|
def test_progress_renderer_deterministic_output() -> None:
|
|
events = [
|
|
action_started("a-1", "command", "echo ok"),
|
|
action_completed(
|
|
"a-1",
|
|
"command",
|
|
"echo ok",
|
|
ok=True,
|
|
detail={"exit_code": 0},
|
|
),
|
|
]
|
|
t1 = ProgressTracker(engine="codex")
|
|
t2 = ProgressTracker(engine="codex")
|
|
|
|
for evt in events:
|
|
t1.note_event(evt)
|
|
t2.note_event(evt)
|
|
|
|
f1 = MarkdownFormatter(max_actions=5)
|
|
f2 = MarkdownFormatter(max_actions=5)
|
|
assert assemble_markdown_parts(
|
|
f1.render_progress_parts(t1.snapshot(), elapsed_s=1.0)
|
|
) == assemble_markdown_parts(f2.render_progress_parts(t2.snapshot(), elapsed_s=1.0))
|
|
|
|
|
|
def test_format_elapsed_branches() -> None:
|
|
assert format_elapsed(3661) == "1h 01m"
|
|
assert format_elapsed(61) == "1m 01s"
|
|
assert format_elapsed(1.4) == "1s"
|
|
|
|
|
|
def test_shorten_and_action_status_branches() -> None:
|
|
assert shorten("hello", None) == "hello"
|
|
assert shorten("hello", 0) == ""
|
|
shortened = shorten("hello world", 6)
|
|
assert shortened.endswith("…")
|
|
assert len(shortened) <= 6
|
|
|
|
action_ok = Action(id="ok", kind="command", title="x", detail={"exit_code": 0})
|
|
action_fail = Action(id="fail", kind="command", title="x", detail={"exit_code": 2})
|
|
|
|
assert action_status(action_ok, completed=False, ok=None) == STATUS["running"]
|
|
assert action_status(action_ok, completed=True, ok=None) == STATUS["done"]
|
|
assert action_status(action_fail, completed=True, ok=None) == STATUS["fail"]
|
|
|
|
|
|
def test_format_file_change_title_handles_overflow_and_invalid() -> None:
|
|
action = Action(
|
|
id="f",
|
|
kind="file_change",
|
|
title="files",
|
|
detail={
|
|
"changes": [
|
|
"bad",
|
|
{"path": ""},
|
|
{"path": "a", "kind": "add"},
|
|
{"path": "b"},
|
|
{"path": "c"},
|
|
{"path": "d"},
|
|
]
|
|
},
|
|
)
|
|
title = format_file_change_title(action, command_width=200)
|
|
assert title.startswith("files: ")
|
|
assert "…(" in title
|
|
|
|
fallback = format_file_change_title(
|
|
Action(id="empty", kind="file_change", title="all files"), command_width=50
|
|
)
|
|
assert fallback == "files: all files"
|
|
|
|
|
|
def test_render_event_cli_ignores_turn_actions() -> None:
|
|
event = ActionEvent(
|
|
engine="codex",
|
|
action=Action(id="turn", kind="turn", title="turn"),
|
|
phase="started",
|
|
ok=None,
|
|
)
|
|
assert render_event_cli(event) == []
|
|
|
|
|
|
def test_progress_renderer_ignores_missing_action_id() -> None:
|
|
tracker = ProgressTracker(engine="codex")
|
|
resume = ResumeToken(engine="codex", value="abc")
|
|
tracker.note_event(StartedEvent(engine="codex", resume=resume, title="Session"))
|
|
|
|
event = ActionEvent(
|
|
engine="codex",
|
|
action=Action(id="", kind="command", title="echo"),
|
|
phase="started",
|
|
ok=None,
|
|
)
|
|
assert tracker.note_event(event) is False
|
|
|
|
formatter = MarkdownFormatter()
|
|
header = assemble_markdown_parts(
|
|
formatter.render_progress_parts(tracker.snapshot(), elapsed_s=0.0)
|
|
)
|
|
assert header.startswith("working · codex · 0s")
|