test(blackbox): add comprehensive test coverage for dashboard
This commit is contained in:
Kevin Turcios
parent
1b56333a8a
commit
9afb317bb2
6 changed files with 1025 additions and 15 deletions
|
|
@ -20,13 +20,14 @@ from blackbox.dashboard.rendering import (
|
|||
passes_filter,
|
||||
render_log_html,
|
||||
)
|
||||
from blackbox.dashboard.transcript import parse_transcript, scan_sessions
|
||||
from blackbox.dashboard.transcript import parse_transcript, parse_transcript_tail, scan_sessions
|
||||
from blackbox.models import SessionInfo
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
from blackbox.dashboard.watcher import SessionWatcher
|
||||
from blackbox.models import LogEntry
|
||||
|
||||
TEMPLATES_DIR = Path(__file__).parent / "templates"
|
||||
PROJECTS_DIR = Path.home() / ".claude" / "projects"
|
||||
|
|
@ -107,7 +108,17 @@ def build_session_info(path: Path, session_id: str, project_path: str) -> Sessio
|
|||
)
|
||||
|
||||
|
||||
async def log_stream( # noqa: C901
|
||||
def filter_and_render(entries: list[LogEntry], filter_name: str, allowed: set[str] | None) -> list[str]:
|
||||
return [
|
||||
html
|
||||
for entry in entries
|
||||
if passes_filter(entry, filter_name, allowed)
|
||||
for html in [render_log_html(entry)]
|
||||
if html
|
||||
]
|
||||
|
||||
|
||||
async def log_stream(
|
||||
transcript_path: Path,
|
||||
filter_name: str,
|
||||
) -> AsyncIterator[ServerSentEvent]:
|
||||
|
|
@ -126,26 +137,19 @@ async def log_stream( # noqa: C901
|
|||
if batch:
|
||||
yield ServerSentEvent(data="\n".join(batch), event="log")
|
||||
|
||||
last_size = transcript_path.stat().st_size
|
||||
offset = transcript_path.stat().st_size
|
||||
while True:
|
||||
await asyncio.sleep(1.0)
|
||||
try:
|
||||
current_size = transcript_path.stat().st_size
|
||||
except OSError:
|
||||
break
|
||||
if current_size <= last_size:
|
||||
if current_size <= offset:
|
||||
continue
|
||||
new_entries = await asyncio.to_thread(parse_transcript, transcript_path)
|
||||
new_batch: list[str] = []
|
||||
for entry in new_entries[len(entries) :]:
|
||||
if passes_filter(entry, filter_name, allowed):
|
||||
html = render_log_html(entry)
|
||||
if html:
|
||||
new_batch.append(html)
|
||||
entries = new_entries
|
||||
last_size = current_size
|
||||
if new_batch:
|
||||
yield ServerSentEvent(data="\n".join(new_batch), event="log")
|
||||
new_entries, offset = await asyncio.to_thread(parse_transcript_tail, transcript_path, offset)
|
||||
rendered = filter_and_render(new_entries, filter_name, allowed)
|
||||
if rendered:
|
||||
yield ServerSentEvent(data="\n".join(rendered), event="log")
|
||||
|
||||
|
||||
@router.get("/sessions/{project_path}/{session_id}/logs")
|
||||
|
|
|
|||
|
|
@ -58,6 +58,24 @@ def parse_transcript(path: Path) -> list[LogEntry]:
|
|||
return entries
|
||||
|
||||
|
||||
def parse_transcript_tail(path: Path, offset: int) -> tuple[list[LogEntry], int]:
|
||||
"""Parse only new bytes appended after *offset*. Returns (entries, new_offset)."""
|
||||
with path.open("rb") as f:
|
||||
f.seek(offset)
|
||||
tail = f.read()
|
||||
new_offset = offset + len(tail)
|
||||
entries: list[LogEntry] = []
|
||||
for line in tail.decode("utf-8", errors="replace").splitlines():
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
raw = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
entries.extend(parse_entry(raw))
|
||||
return entries, new_offset
|
||||
|
||||
|
||||
def parse_entry(raw: dict[str, Any]) -> list[LogEntry]:
|
||||
entry_type = raw.get("type", "")
|
||||
ts = ts_to_epoch(raw.get("timestamp"))
|
||||
|
|
|
|||
271
packages/blackbox/tests/test_rendering.py
Normal file
271
packages/blackbox/tests/test_rendering.py
Normal file
|
|
@ -0,0 +1,271 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from blackbox.dashboard.rendering import (
|
||||
esc,
|
||||
esc_md,
|
||||
fmt_duration,
|
||||
fmt_relative,
|
||||
fmt_time,
|
||||
passes_filter,
|
||||
render_log_html,
|
||||
shorten_paths,
|
||||
tool_call_html,
|
||||
)
|
||||
from blackbox.models import LogEntry
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# fmt_time
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFmtTime:
|
||||
def test_epoch_zero(self) -> None:
|
||||
assert "00:00:00" == fmt_time(0.0)
|
||||
|
||||
def test_known_timestamp(self) -> None:
|
||||
assert "01:46:40" == fmt_time(1_000_000_000.0)
|
||||
|
||||
def test_fractional_seconds_truncated(self) -> None:
|
||||
assert "00:00:00" == fmt_time(0.999)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# fmt_duration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFmtDuration:
|
||||
def test_zero_seconds(self) -> None:
|
||||
assert "0s" == fmt_duration(100.0, 100.0)
|
||||
|
||||
def test_seconds_only(self) -> None:
|
||||
assert "45s" == fmt_duration(0.0, 45.0)
|
||||
|
||||
def test_minutes_and_seconds(self) -> None:
|
||||
assert "2m30s" == fmt_duration(0.0, 150.0)
|
||||
|
||||
def test_hours_and_minutes(self) -> None:
|
||||
assert "1h30m" == fmt_duration(0.0, 5400.0)
|
||||
|
||||
def test_negative_clamps_to_zero(self) -> None:
|
||||
assert "0s" == fmt_duration(100.0, 50.0)
|
||||
|
||||
def test_none_finished_uses_current_time(self) -> None:
|
||||
result = fmt_duration(time.time() - 10, None)
|
||||
assert result.endswith("s")
|
||||
|
||||
def test_exactly_60_seconds(self) -> None:
|
||||
assert "1m00s" == fmt_duration(0.0, 60.0)
|
||||
|
||||
def test_exactly_one_hour(self) -> None:
|
||||
assert "1h00m" == fmt_duration(0.0, 3600.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# fmt_relative
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFmtRelative:
|
||||
def test_just_now(self) -> None:
|
||||
assert "just now" == fmt_relative(time.time())
|
||||
|
||||
def test_minutes_ago(self) -> None:
|
||||
assert "5m ago" == fmt_relative(time.time() - 300)
|
||||
|
||||
def test_hours_ago(self) -> None:
|
||||
assert "2h ago" == fmt_relative(time.time() - 7200)
|
||||
|
||||
def test_days_ago(self) -> None:
|
||||
assert "3d ago" == fmt_relative(time.time() - 259200)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# esc / esc_md
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEsc:
|
||||
def test_ampersand(self) -> None:
|
||||
assert "a & b" == esc("a & b")
|
||||
|
||||
def test_angle_brackets(self) -> None:
|
||||
assert "<div>" == esc("<div>")
|
||||
|
||||
def test_newlines_become_br(self) -> None:
|
||||
assert "a<br>b" == esc("a\nb")
|
||||
|
||||
def test_combined(self) -> None:
|
||||
assert "<b>hi</b><br>&" == esc("<b>hi</b>\n&")
|
||||
|
||||
|
||||
class TestEscMd:
|
||||
def test_bold_converted(self) -> None:
|
||||
result = esc_md("hello **world**")
|
||||
assert '<strong class="text-white">world</strong>' in result
|
||||
|
||||
def test_html_still_escaped(self) -> None:
|
||||
result = esc_md("<script>**bold**")
|
||||
assert "<script>" in result
|
||||
assert '<strong class="text-white">bold</strong>' in result
|
||||
|
||||
def test_no_bold(self) -> None:
|
||||
assert "plain text" == esc_md("plain text")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# shorten_paths
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestShortenPaths:
|
||||
def test_removes_tmp_paths(self) -> None:
|
||||
assert "file: " == shorten_paths("file: /tmp/abc123/foo.py")
|
||||
|
||||
def test_removes_private_tmp_paths(self) -> None:
|
||||
assert "file: " == shorten_paths("file: /private/tmp/abc123/bar.py")
|
||||
|
||||
def test_no_match_unchanged(self) -> None:
|
||||
assert "/home/user/code" == shorten_paths("/home/user/code")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# passes_filter
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPassesFilter:
|
||||
def make(self, level: str = "info", message: str = "hi", source: str = "user") -> LogEntry:
|
||||
return LogEntry(timestamp=0.0, source=source, level=level, message=message)
|
||||
|
||||
def test_empty_message_rejected(self) -> None:
|
||||
assert not passes_filter(self.make(message=""), "all", None)
|
||||
|
||||
def test_whitespace_only_rejected(self) -> None:
|
||||
assert not passes_filter(self.make(message=" "), "all", None)
|
||||
|
||||
def test_all_filter_accepts_everything(self) -> None:
|
||||
assert passes_filter(self.make(), "all", None)
|
||||
|
||||
def test_skip_levels_rejected_in_compact(self) -> None:
|
||||
for level in ("delta", "stream", "block_stop", "block_start", "thinking_delta", "tool_start"):
|
||||
assert not passes_filter(self.make(level=level), "compact", None)
|
||||
|
||||
def test_allowed_set_filters(self) -> None:
|
||||
allowed = {"error"}
|
||||
assert passes_filter(self.make(level="error"), "errors", allowed)
|
||||
assert not passes_filter(self.make(level="info"), "errors", allowed)
|
||||
|
||||
def test_thinking_rejected_in_compact(self) -> None:
|
||||
entry = self.make(level="assistant", message="(thinking)")
|
||||
assert not passes_filter(entry, "compact", None)
|
||||
|
||||
def test_assistant_non_thinking_accepted(self) -> None:
|
||||
entry = self.make(level="assistant", message="Hello there")
|
||||
assert passes_filter(entry, "compact", None)
|
||||
|
||||
def test_skip_levels_pass_in_all_mode(self) -> None:
|
||||
entry = self.make(level="delta", message="x")
|
||||
assert passes_filter(entry, "all", None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# tool_call_html
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestToolCallHtml:
|
||||
def test_short_preview_no_details(self) -> None:
|
||||
html = tool_call_html("ls -la")
|
||||
assert "<details" not in html
|
||||
assert "ls -la" in html
|
||||
|
||||
def test_three_lines_no_details(self) -> None:
|
||||
html = tool_call_html("line1\nline2\nline3")
|
||||
assert "<details" not in html
|
||||
|
||||
def test_long_preview_has_details(self) -> None:
|
||||
text = "\n".join(f"line {i}" for i in range(10))
|
||||
html = tool_call_html(text)
|
||||
assert "<details" in html
|
||||
assert "+9 lines" in html
|
||||
|
||||
def test_tmp_paths_shortened(self) -> None:
|
||||
html = tool_call_html("/tmp/abc123/foo.py")
|
||||
assert "/tmp/" not in html
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# render_log_html
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRenderLogHtml:
|
||||
def make(self, **kw: object) -> LogEntry:
|
||||
defaults = {
|
||||
"timestamp": 1_000_000_000.0,
|
||||
"source": "claude",
|
||||
"level": "assistant",
|
||||
"message": "hello",
|
||||
"data": {},
|
||||
}
|
||||
defaults.update(kw)
|
||||
return LogEntry(**defaults) # type: ignore[arg-type]
|
||||
|
||||
def test_user_message_green(self) -> None:
|
||||
html = render_log_html(self.make(source="user", level="info"))
|
||||
assert "text-green-300" in html
|
||||
|
||||
def test_assistant_message(self) -> None:
|
||||
html = render_log_html(self.make(level="assistant", message="hi"))
|
||||
assert "text-gray-100" in html
|
||||
|
||||
def test_thinking_italic(self) -> None:
|
||||
html = render_log_html(self.make(level="assistant", message="(thinking)"))
|
||||
assert "italic" in html
|
||||
|
||||
def test_error_red(self) -> None:
|
||||
html = render_log_html(self.make(level="error", message="fail"))
|
||||
assert "text-red-400" in html
|
||||
|
||||
def test_tool_call_amber_badge(self) -> None:
|
||||
html = render_log_html(
|
||||
self.make(
|
||||
level="tool_call",
|
||||
message="Bash: ls",
|
||||
data={"tool": "Bash", "input_preview": "ls"},
|
||||
)
|
||||
)
|
||||
assert "bg-amber-500" in html
|
||||
assert "Bash" in html
|
||||
|
||||
def test_tool_result_has_res_badge(self) -> None:
|
||||
html = render_log_html(self.make(level="tool_result", message="output"))
|
||||
assert "RES" in html
|
||||
|
||||
def test_tool_result_truncates_long_messages(self) -> None:
|
||||
html = render_log_html(self.make(level="tool_result", message="x" * 600))
|
||||
assert "..." in html
|
||||
|
||||
def test_contains_timestamp(self) -> None:
|
||||
html = render_log_html(self.make())
|
||||
assert "01:46:40" in html
|
||||
|
||||
def test_source_badges(self) -> None:
|
||||
for source, label in [("claude", "CLU"), ("user", "USR"), ("system", "SYS")]:
|
||||
html = render_log_html(self.make(source=source, level="info"))
|
||||
assert label in html
|
||||
|
||||
def test_tool_levels_have_indent_and_opacity(self) -> None:
|
||||
html = render_log_html(self.make(level="tool_call", data={"tool": "Read", "input_preview": "x"}))
|
||||
assert "opacity-60" in html
|
||||
assert "pl-4" in html
|
||||
|
||||
def test_non_tool_no_indent(self) -> None:
|
||||
html = render_log_html(self.make(level="info"))
|
||||
assert "pl-4" not in html
|
||||
77
packages/blackbox/tests/test_routes.py
Normal file
77
packages/blackbox/tests/test_routes.py
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import attrs
|
||||
|
||||
from blackbox.dashboard.routes import build_session_info, mark_live
|
||||
from blackbox.models import SessionInfo
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# mark_live
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMarkLive:
|
||||
def make_session(self, session_id: str = "sess-1") -> SessionInfo:
|
||||
return SessionInfo(
|
||||
session_id=session_id,
|
||||
project_path="proj",
|
||||
project_name="proj",
|
||||
transcript_path="/tmp/proj/sess.jsonl",
|
||||
started_at=1_000_000.0,
|
||||
)
|
||||
|
||||
def test_no_live_ids_returns_same_list(self) -> None:
|
||||
watcher = MagicMock()
|
||||
watcher.live_session_ids.return_value = set()
|
||||
sessions = [self.make_session()]
|
||||
result = mark_live(sessions, watcher)
|
||||
assert result is sessions
|
||||
|
||||
def test_marks_matching_session_as_live(self) -> None:
|
||||
watcher = MagicMock()
|
||||
watcher.live_session_ids.return_value = {"sess-1"}
|
||||
sessions = [self.make_session("sess-1"), self.make_session("sess-2")]
|
||||
result = mark_live(sessions, watcher)
|
||||
assert result[0].is_live is True
|
||||
assert result[1].is_live is False
|
||||
|
||||
def test_non_matching_sessions_unchanged(self) -> None:
|
||||
watcher = MagicMock()
|
||||
watcher.live_session_ids.return_value = {"other"}
|
||||
sessions = [self.make_session("sess-1")]
|
||||
result = mark_live(sessions, watcher)
|
||||
assert result[0].is_live is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# build_session_info
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildSessionInfo:
|
||||
def write_transcript(self, path: Path, entries: list[dict[str, Any]]) -> None:
|
||||
path.write_text("\n".join(json.dumps(e) for e in entries))
|
||||
|
||||
def test_returns_info_from_transcript(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "abc.jsonl"
|
||||
self.write_transcript(
|
||||
path,
|
||||
[{"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hello"}}],
|
||||
)
|
||||
info = build_session_info(path, "abc", "proj")
|
||||
assert "abc" == info.session_id
|
||||
assert "hello" == info.first_prompt
|
||||
|
||||
def test_fallback_for_empty_transcript(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "empty.jsonl"
|
||||
path.write_text("")
|
||||
info = build_session_info(path, "empty", "proj")
|
||||
assert "empty" == info.session_id
|
||||
assert "proj" == info.project_name
|
||||
assert "" == info.first_prompt
|
||||
555
packages/blackbox/tests/test_transcript.py
Normal file
555
packages/blackbox/tests/test_transcript.py
Normal file
|
|
@ -0,0 +1,555 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from blackbox.dashboard.transcript import (
|
||||
decode_project_name,
|
||||
extract_text_content,
|
||||
extract_tool_results,
|
||||
extract_tool_uses,
|
||||
parse_assistant_entry,
|
||||
parse_entry,
|
||||
parse_transcript,
|
||||
parse_transcript_tail,
|
||||
parse_user_entry,
|
||||
quick_session_info,
|
||||
scan_sessions,
|
||||
tool_input_preview,
|
||||
ts_to_epoch,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ts_to_epoch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTsToEpoch:
|
||||
def test_none_returns_zero(self) -> None:
|
||||
assert 0.0 == ts_to_epoch(None)
|
||||
|
||||
def test_empty_string_returns_zero(self) -> None:
|
||||
assert 0.0 == ts_to_epoch("")
|
||||
|
||||
def test_valid_iso_timestamp(self) -> None:
|
||||
result = ts_to_epoch("2024-01-01T00:00:00Z")
|
||||
assert result > 0
|
||||
|
||||
def test_naive_datetime_treated_as_utc(self) -> None:
|
||||
result = ts_to_epoch("2024-01-01T00:00:00")
|
||||
assert result > 0
|
||||
|
||||
def test_invalid_format_returns_zero(self) -> None:
|
||||
assert 0.0 == ts_to_epoch("not-a-date")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# extract_text_content
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestExtractTextContent:
|
||||
def test_string_passthrough(self) -> None:
|
||||
assert "hello" == extract_text_content("hello")
|
||||
|
||||
def test_list_of_text_blocks(self) -> None:
|
||||
content = [
|
||||
{"type": "text", "text": "hello"},
|
||||
{"type": "text", "text": "world"},
|
||||
]
|
||||
assert "hello\nworld" == extract_text_content(content)
|
||||
|
||||
def test_non_text_blocks_skipped(self) -> None:
|
||||
content = [
|
||||
{"type": "tool_use", "name": "Bash"},
|
||||
{"type": "text", "text": "only this"},
|
||||
]
|
||||
assert "only this" == extract_text_content(content)
|
||||
|
||||
def test_empty_list(self) -> None:
|
||||
assert "" == extract_text_content([])
|
||||
|
||||
def test_non_string_non_list(self) -> None:
|
||||
assert "" == extract_text_content(42)
|
||||
|
||||
def test_non_dict_items_skipped(self) -> None:
|
||||
assert "" == extract_text_content(["not a dict"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# extract_tool_uses / extract_tool_results
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestExtractToolUses:
|
||||
def test_extracts_tool_use_blocks(self) -> None:
|
||||
content = [{"type": "tool_use", "name": "Read"}, {"type": "text", "text": "x"}]
|
||||
assert [{"type": "tool_use", "name": "Read"}] == extract_tool_uses(content)
|
||||
|
||||
def test_non_list_returns_empty(self) -> None:
|
||||
assert [] == extract_tool_uses("string")
|
||||
|
||||
def test_empty_list(self) -> None:
|
||||
assert [] == extract_tool_uses([])
|
||||
|
||||
|
||||
class TestExtractToolResults:
|
||||
def test_extracts_tool_result_blocks(self) -> None:
|
||||
content = [{"type": "tool_result", "content": "ok"}, {"type": "text", "text": "x"}]
|
||||
assert [{"type": "tool_result", "content": "ok"}] == extract_tool_results(content)
|
||||
|
||||
def test_non_list_returns_empty(self) -> None:
|
||||
assert [] == extract_tool_results(42)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# tool_input_preview
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestToolInputPreview:
|
||||
def test_bash_shows_command(self) -> None:
|
||||
assert "ls -la" == tool_input_preview("Bash", {"command": "ls -la"})
|
||||
|
||||
def test_read_shows_path(self) -> None:
|
||||
assert "/foo.py" == tool_input_preview("Read", {"file_path": "/foo.py"})
|
||||
|
||||
def test_write_shows_path(self) -> None:
|
||||
assert "/bar.py" == tool_input_preview("Write", {"file_path": "/bar.py"})
|
||||
|
||||
def test_edit_shows_path_and_old_string(self) -> None:
|
||||
result = tool_input_preview("Edit", {"file_path": "/f.py", "old_string": "x" * 100})
|
||||
assert "/f.py" in result
|
||||
assert result.endswith("...")
|
||||
|
||||
def test_agent_shows_description(self) -> None:
|
||||
result = tool_input_preview("Agent", {"description": "find bugs", "prompt": "long prompt"})
|
||||
assert "find bugs" == result
|
||||
|
||||
def test_agent_falls_back_to_prompt(self) -> None:
|
||||
result = tool_input_preview("Agent", {"prompt": "do stuff"})
|
||||
assert "do stuff" == result
|
||||
|
||||
def test_skill_shows_skill_name(self) -> None:
|
||||
assert "commit" == tool_input_preview("Skill", {"skill": "commit"})
|
||||
|
||||
def test_unknown_tool_json_preview(self) -> None:
|
||||
result = tool_input_preview("CustomTool", {"key": "value"})
|
||||
assert "key" in result
|
||||
assert "value" in result
|
||||
|
||||
def test_unknown_tool_truncated_at_200(self) -> None:
|
||||
result = tool_input_preview("CustomTool", {"key": "x" * 300})
|
||||
assert len(result) <= 200
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse_entry / parse_user_entry / parse_assistant_entry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseEntry:
|
||||
def test_user_entry(self) -> None:
|
||||
raw = {"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hello"}}
|
||||
entries = parse_entry(raw)
|
||||
assert 1 == len(entries)
|
||||
assert "user" == entries[0].source
|
||||
assert "hello" == entries[0].message
|
||||
|
||||
def test_assistant_text_entry(self) -> None:
|
||||
raw = {
|
||||
"type": "assistant",
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"message": {"content": [{"type": "text", "text": "hi"}]},
|
||||
}
|
||||
entries = parse_entry(raw)
|
||||
assert 1 == len(entries)
|
||||
assert "claude" == entries[0].source
|
||||
assert "assistant" == entries[0].level
|
||||
|
||||
def test_assistant_tool_use(self) -> None:
|
||||
raw = {
|
||||
"type": "assistant",
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"message": {"content": [{"type": "tool_use", "name": "Read", "input": {"file_path": "/x.py"}}]},
|
||||
}
|
||||
entries = parse_entry(raw)
|
||||
assert 1 == len(entries)
|
||||
assert "tool_call" == entries[0].level
|
||||
assert "Read" in entries[0].message
|
||||
|
||||
def test_assistant_thinking_block(self) -> None:
|
||||
raw = {
|
||||
"type": "assistant",
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"message": {"content": [{"type": "thinking"}]},
|
||||
}
|
||||
entries = parse_entry(raw)
|
||||
assert 1 == len(entries)
|
||||
assert "(thinking)" == entries[0].message
|
||||
|
||||
def test_system_entry(self) -> None:
|
||||
raw = {"type": "system", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "init"}}
|
||||
entries = parse_entry(raw)
|
||||
assert 1 == len(entries)
|
||||
assert "system" == entries[0].source
|
||||
|
||||
def test_unknown_type_returns_empty(self) -> None:
|
||||
assert [] == parse_entry({"type": "unknown"})
|
||||
|
||||
|
||||
class TestParseUserEntry:
|
||||
def test_tool_result_with_error(self) -> None:
|
||||
content = [{"type": "tool_result", "content": "fail", "is_error": True}]
|
||||
raw: dict[str, Any] = {"toolUseResult": {"stderr": "bad command"}}
|
||||
entries = parse_user_entry(0.0, {"content": content}, raw)
|
||||
assert 1 == len(entries)
|
||||
assert "error" == entries[0].level
|
||||
assert "bad command" == entries[0].message
|
||||
|
||||
def test_tool_result_success(self) -> None:
|
||||
content = [{"type": "tool_result", "content": "output text"}]
|
||||
entries = parse_user_entry(0.0, {"content": content}, {})
|
||||
assert 1 == len(entries)
|
||||
assert "tool_result" == entries[0].level
|
||||
|
||||
def test_tool_result_with_stdout(self) -> None:
|
||||
content = [{"type": "tool_result", "content": "ignored"}]
|
||||
raw: dict[str, Any] = {"toolUseResult": {"stdout": "real output"}}
|
||||
entries = parse_user_entry(0.0, {"content": content}, raw)
|
||||
assert "real output" == entries[0].message
|
||||
|
||||
def test_tool_result_content_as_list(self) -> None:
|
||||
content = [{"type": "tool_result", "content": [{"text": "a"}, {"text": "b"}]}]
|
||||
entries = parse_user_entry(0.0, {"content": content}, {})
|
||||
assert "a b" == entries[0].message
|
||||
|
||||
def test_non_dict_message_returns_empty(self) -> None:
|
||||
assert [] == parse_user_entry(0.0, "not a dict", {})
|
||||
|
||||
def test_tool_result_message_truncated_at_2000(self) -> None:
|
||||
content = [{"type": "tool_result", "content": "x" * 3000}]
|
||||
entries = parse_user_entry(0.0, {"content": content}, {})
|
||||
assert 2000 == len(entries[0].message)
|
||||
|
||||
def test_tool_use_result_non_dict_ignored(self) -> None:
|
||||
content = [{"type": "tool_result", "content": "ok", "is_error": True}]
|
||||
raw: dict[str, Any] = {"toolUseResult": "not a dict"}
|
||||
entries = parse_user_entry(0.0, {"content": content}, raw)
|
||||
assert 1 == len(entries)
|
||||
|
||||
|
||||
class TestParseAssistantEntry:
|
||||
def test_non_dict_message_returns_empty(self) -> None:
|
||||
assert [] == parse_assistant_entry(0.0, "not a dict")
|
||||
|
||||
def test_string_content(self) -> None:
|
||||
entries = parse_assistant_entry(0.0, {"content": "hello"})
|
||||
assert 1 == len(entries)
|
||||
assert "hello" == entries[0].message
|
||||
|
||||
def test_empty_content(self) -> None:
|
||||
assert [] == parse_assistant_entry(0.0, {"content": ""})
|
||||
assert [] == parse_assistant_entry(0.0, {"content": []})
|
||||
|
||||
def test_non_dict_blocks_skipped(self) -> None:
|
||||
entries = parse_assistant_entry(0.0, {"content": ["not a dict"]})
|
||||
assert [] == entries
|
||||
|
||||
def test_mixed_content(self) -> None:
|
||||
content = [
|
||||
{"type": "text", "text": "thinking about it"},
|
||||
{"type": "tool_use", "name": "Bash", "input": {"command": "ls"}},
|
||||
{"type": "thinking"},
|
||||
]
|
||||
entries = parse_assistant_entry(0.0, {"content": content})
|
||||
assert 3 == len(entries)
|
||||
assert "assistant" == entries[0].level
|
||||
assert "tool_call" == entries[1].level
|
||||
assert "(thinking)" == entries[2].message
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse_transcript
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseTranscript:
|
||||
def test_parses_jsonl(self, tmp_path: Path) -> None:
|
||||
lines = [
|
||||
json.dumps({"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hi"}}),
|
||||
json.dumps(
|
||||
{
|
||||
"type": "assistant",
|
||||
"timestamp": "2024-01-01T00:00:01Z",
|
||||
"message": {"content": [{"type": "text", "text": "hello"}]},
|
||||
}
|
||||
),
|
||||
]
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text("\n".join(lines))
|
||||
entries = parse_transcript(path)
|
||||
assert 2 == len(entries)
|
||||
assert "user" == entries[0].source
|
||||
assert "claude" == entries[1].source
|
||||
|
||||
def test_skips_blank_lines(self, tmp_path: Path) -> None:
|
||||
lines = [
|
||||
json.dumps({"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hi"}}),
|
||||
"",
|
||||
" ",
|
||||
json.dumps(
|
||||
{
|
||||
"type": "assistant",
|
||||
"timestamp": "2024-01-01T00:00:01Z",
|
||||
"message": {"content": [{"type": "text", "text": "ok"}]},
|
||||
}
|
||||
),
|
||||
]
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text("\n".join(lines))
|
||||
assert 2 == len(parse_transcript(path))
|
||||
|
||||
def test_skips_invalid_json(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text("not json\n{bad json}\n")
|
||||
assert [] == parse_transcript(path)
|
||||
|
||||
def test_empty_file(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text("")
|
||||
assert [] == parse_transcript(path)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse_transcript_tail
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseTranscriptTail:
|
||||
def test_reads_from_offset(self, tmp_path: Path) -> None:
|
||||
line1 = json.dumps({"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "first"}})
|
||||
line2 = json.dumps({"type": "user", "timestamp": "2024-01-01T00:01:00Z", "message": {"content": "second"}})
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text(line1 + "\n")
|
||||
offset = path.stat().st_size
|
||||
|
||||
with path.open("a") as f:
|
||||
f.write(line2 + "\n")
|
||||
|
||||
entries, new_offset = parse_transcript_tail(path, offset)
|
||||
assert 1 == len(entries)
|
||||
assert "second" == entries[0].message
|
||||
assert new_offset > offset
|
||||
|
||||
def test_offset_zero_reads_full_file(self, tmp_path: Path) -> None:
|
||||
line = json.dumps({"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hi"}})
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text(line + "\n")
|
||||
entries, offset = parse_transcript_tail(path, 0)
|
||||
assert 1 == len(entries)
|
||||
assert offset == path.stat().st_size
|
||||
|
||||
def test_no_new_data_returns_empty(self, tmp_path: Path) -> None:
|
||||
line = json.dumps({"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hi"}})
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text(line + "\n")
|
||||
offset = path.stat().st_size
|
||||
entries, new_offset = parse_transcript_tail(path, offset)
|
||||
assert [] == entries
|
||||
assert new_offset == offset
|
||||
|
||||
def test_skips_invalid_json_in_tail(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text("")
|
||||
offset = 0
|
||||
with path.open("a") as f:
|
||||
f.write("bad json\n")
|
||||
f.write(
|
||||
json.dumps({"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "ok"}}) + "\n"
|
||||
)
|
||||
entries, _ = parse_transcript_tail(path, offset)
|
||||
assert 1 == len(entries)
|
||||
assert "ok" == entries[0].message
|
||||
|
||||
def test_multiple_appends(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "session.jsonl"
|
||||
path.write_text("")
|
||||
offset = 0
|
||||
|
||||
for i in range(3):
|
||||
with path.open("a") as f:
|
||||
f.write(
|
||||
json.dumps(
|
||||
{"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": f"msg-{i}"}}
|
||||
)
|
||||
+ "\n"
|
||||
)
|
||||
entries, offset = parse_transcript_tail(path, offset)
|
||||
assert 1 == len(entries)
|
||||
assert f"msg-{i}" == entries[0].message
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# decode_project_name
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDecodeProjectName:
|
||||
def test_encoded_path(self) -> None:
|
||||
assert "work/myproject" == decode_project_name("-Users-kevin-Desktop-work-myproject")
|
||||
|
||||
def test_filters_common_parts(self) -> None:
|
||||
result = decode_project_name("-Users-private-tmp-")
|
||||
assert result == "-Users-private-tmp-"
|
||||
|
||||
def test_simple_name_passthrough(self) -> None:
|
||||
assert "myproject" == decode_project_name("myproject")
|
||||
|
||||
def test_short_meaningful_parts(self) -> None:
|
||||
assert "kevin/myproject" == decode_project_name("-Users-kevin-myproject")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# quick_session_info
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestQuickSessionInfo:
|
||||
def write_transcript(self, path: Path, entries: list[dict[str, Any]]) -> None:
|
||||
path.write_text("\n".join(json.dumps(e) for e in entries))
|
||||
|
||||
def test_basic_session(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "abc123.jsonl"
|
||||
self.write_transcript(
|
||||
path,
|
||||
[
|
||||
{
|
||||
"type": "user",
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"message": {"content": "help me debug this"},
|
||||
"cwd": "/home/user/projects/myapp",
|
||||
},
|
||||
{
|
||||
"type": "assistant",
|
||||
"timestamp": "2024-01-01T00:05:00Z",
|
||||
"message": {"content": [{"type": "text", "text": "sure"}]},
|
||||
},
|
||||
],
|
||||
)
|
||||
info = quick_session_info(path, "abc123", "encoded-project", "myproject")
|
||||
assert info is not None
|
||||
assert "abc123" == info.session_id
|
||||
assert "help me debug this" == info.first_prompt
|
||||
assert 1 == info.message_count
|
||||
assert "projects/myapp" == info.project_name
|
||||
|
||||
def test_skips_tool_results_for_first_prompt(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "abc123.jsonl"
|
||||
self.write_transcript(
|
||||
path,
|
||||
[
|
||||
{
|
||||
"type": "user",
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"message": {"content": [{"type": "tool_result", "content": "output"}]},
|
||||
},
|
||||
{
|
||||
"type": "user",
|
||||
"timestamp": "2024-01-01T00:01:00Z",
|
||||
"message": {"content": "real prompt"},
|
||||
},
|
||||
],
|
||||
)
|
||||
info = quick_session_info(path, "abc123", "enc", "proj")
|
||||
assert info is not None
|
||||
assert "real prompt" == info.first_prompt
|
||||
|
||||
def test_returns_none_for_empty_file(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "empty.jsonl"
|
||||
path.write_text("")
|
||||
assert quick_session_info(path, "empty", "enc", "proj") is None
|
||||
|
||||
def test_returns_none_for_missing_file(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "missing.jsonl"
|
||||
assert quick_session_info(path, "missing", "enc", "proj") is None
|
||||
|
||||
def test_first_prompt_truncated_at_120(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "abc.jsonl"
|
||||
self.write_transcript(
|
||||
path,
|
||||
[{"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "x" * 200}}],
|
||||
)
|
||||
info = quick_session_info(path, "abc", "enc", "proj")
|
||||
assert info is not None
|
||||
assert 120 == len(info.first_prompt)
|
||||
|
||||
def test_uses_mtime_for_finished_at(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "abc.jsonl"
|
||||
self.write_transcript(
|
||||
path,
|
||||
[{"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hi"}}],
|
||||
)
|
||||
info = quick_session_info(path, "abc", "enc", "proj")
|
||||
assert info is not None
|
||||
assert info.finished_at is not None
|
||||
assert info.finished_at >= info.started_at
|
||||
|
||||
def test_cwd_used_for_display_name(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "abc.jsonl"
|
||||
self.write_transcript(
|
||||
path,
|
||||
[
|
||||
{
|
||||
"type": "user",
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"message": {"content": "hi"},
|
||||
"cwd": "/Users/kevin/Desktop/work/myapp",
|
||||
},
|
||||
],
|
||||
)
|
||||
info = quick_session_info(path, "abc", "enc", "proj")
|
||||
assert info is not None
|
||||
assert "work/myapp" == info.project_name
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# scan_sessions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestScanSessions:
|
||||
def test_empty_dir(self, tmp_path: Path) -> None:
|
||||
assert [] == scan_sessions(tmp_path)
|
||||
|
||||
def test_nonexistent_dir(self, tmp_path: Path) -> None:
|
||||
assert [] == scan_sessions(tmp_path / "nonexistent")
|
||||
|
||||
def test_finds_sessions(self, tmp_path: Path) -> None:
|
||||
project_dir = tmp_path / "encoded-project"
|
||||
project_dir.mkdir()
|
||||
transcript = project_dir / "sess-123.jsonl"
|
||||
transcript.write_text(
|
||||
json.dumps({"type": "user", "timestamp": "2024-01-01T00:00:00Z", "message": {"content": "hi"}})
|
||||
)
|
||||
sessions = scan_sessions(tmp_path)
|
||||
assert 1 == len(sessions)
|
||||
assert "sess-123" == sessions[0].session_id
|
||||
|
||||
def test_sorted_by_started_at_descending(self, tmp_path: Path) -> None:
|
||||
project_dir = tmp_path / "proj"
|
||||
project_dir.mkdir()
|
||||
for i, ts in enumerate(["2024-01-01T00:00:00Z", "2024-06-01T00:00:00Z", "2024-03-01T00:00:00Z"]):
|
||||
path = project_dir / f"sess-{i}.jsonl"
|
||||
path.write_text(json.dumps({"type": "user", "timestamp": ts, "message": {"content": "hi"}}))
|
||||
sessions = scan_sessions(tmp_path)
|
||||
assert 3 == len(sessions)
|
||||
assert sessions[0].started_at >= sessions[1].started_at >= sessions[2].started_at
|
||||
|
||||
def test_skips_non_directory_entries(self, tmp_path: Path) -> None:
|
||||
(tmp_path / "not_a_dir.txt").write_text("hello")
|
||||
assert [] == scan_sessions(tmp_path)
|
||||
85
packages/blackbox/tests/test_watcher.py
Normal file
85
packages/blackbox/tests/test_watcher.py
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from blackbox.dashboard.watcher import LIVE_THRESHOLD_S, SessionWatcher
|
||||
|
||||
|
||||
class TestSessionWatcherLiveIds:
|
||||
def test_empty_initially(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
assert set() == w.live_session_ids()
|
||||
|
||||
def test_recent_modification_is_live(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
event = MagicMock()
|
||||
event.is_directory = False
|
||||
event.src_path = str(tmp_path / "abc123.jsonl")
|
||||
w.on_modified(event)
|
||||
assert {"abc123"} == w.live_session_ids()
|
||||
|
||||
def test_old_modification_expires(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
with w._lock:
|
||||
w._last_modified["old-session"] = time.time() - LIVE_THRESHOLD_S - 10
|
||||
assert set() == w.live_session_ids()
|
||||
|
||||
def test_non_jsonl_ignored(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
event = MagicMock()
|
||||
event.is_directory = False
|
||||
event.src_path = str(tmp_path / "notes.txt")
|
||||
w.on_modified(event)
|
||||
assert set() == w.live_session_ids()
|
||||
|
||||
def test_directory_events_ignored(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
event = MagicMock()
|
||||
event.is_directory = True
|
||||
event.src_path = str(tmp_path / "somedir")
|
||||
w.on_modified(event)
|
||||
assert set() == w.live_session_ids()
|
||||
|
||||
def test_on_created_delegates_to_on_modified(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
event = MagicMock()
|
||||
event.is_directory = False
|
||||
event.src_path = str(tmp_path / "new-session.jsonl")
|
||||
w.on_created(event)
|
||||
assert {"new-session"} == w.live_session_ids()
|
||||
|
||||
def test_expired_entries_cleaned_up(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
with w._lock:
|
||||
w._last_modified["expired"] = time.time() - LIVE_THRESHOLD_S - 100
|
||||
w._last_modified["fresh"] = time.time()
|
||||
live = w.live_session_ids()
|
||||
assert {"fresh"} == live
|
||||
with w._lock:
|
||||
assert "expired" not in w._last_modified
|
||||
|
||||
|
||||
class TestSessionWatcherStartStop:
|
||||
def test_start_without_directory_is_noop(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path / "nonexistent")
|
||||
w.start()
|
||||
assert w._observer is None
|
||||
|
||||
def test_stop_without_start_is_safe(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
w.stop()
|
||||
|
||||
def test_start_and_stop(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
w.start()
|
||||
assert w._observer is not None
|
||||
w.stop()
|
||||
assert w._observer is None
|
||||
|
||||
def test_multiple_stops_safe(self, tmp_path: Path) -> None:
|
||||
w = SessionWatcher(tmp_path)
|
||||
w.start()
|
||||
w.stop()
|
||||
w.stop()
|
||||
Loading…
Reference in a new issue