Files
wehub-resource-sync f3d80b4628
Auto Release / auto-release (push) Failing after 1s
CI / lint (push) Failing after 0s
CI / screenshot-quality (push) Failing after 3s
CI / pr-policy (push) Has been skipped
CI / test (3.11) (push) Failing after 0s
CI / test (3.12) (push) Failing after 0s
CI / test (3.13) (push) Failing after 3s
CI / coverage (push) Failing after 1s
Legibility / legibility (push) Failing after 3s
chore: import upstream snapshot with attribution
2026-07-13 12:31:48 +08:00

679 lines
26 KiB
Python

#!/usr/bin/env python3
"""Playwright browser tests for global trace search in viewer.html using real trace data."""
from __future__ import annotations
import json
import re
import tempfile
from pathlib import Path
import pytest
from claude_tap.viewer import _generate_html_viewer
pw_missing = False
try:
from playwright.sync_api import sync_playwright # noqa: F401
except ImportError:
pw_missing = True
pytestmark = pytest.mark.skipif(pw_missing, reason="playwright not installed")
_WORD_RE = re.compile(r"[A-Za-z]{4,}")
_STOPWORDS = {
"this",
"that",
"with",
"from",
"have",
"will",
"into",
"your",
"http",
"https",
"json",
"true",
"false",
"null",
}
def _load_entries(trace_file: Path) -> list[dict]:
lines = trace_file.read_text(encoding="utf-8").splitlines()
return [json.loads(line) for line in lines if line.strip()]
def _pick_real_trace_file() -> Path:
traces_dir = Path(__file__).parent.parent / ".traces"
trace_files = sorted(traces_dir.glob("trace_*.jsonl"), key=lambda p: p.stat().st_size)
candidates = []
for path in trace_files:
if path.stat().st_size == 0:
continue
line_count = sum(1 for _ in path.open("r", encoding="utf-8"))
if line_count >= 4:
candidates.append(path)
if not candidates:
pytest.skip("No real trace file with >=4 entries found in .traces/")
return candidates[0]
def _normalize_messages_for_diff(body: dict | None) -> list[dict]:
if not body:
return []
if isinstance(body.get("messages"), list) and body["messages"]:
return [msg for msg in body["messages"] if isinstance(msg, dict)]
if isinstance(body.get("input"), list):
normalized = []
for item in body["input"]:
if not isinstance(item, dict) or item.get("type") != "message":
continue
normalized.append(
{
"role": item.get("role", "user"),
"content": item.get("content"),
}
)
return normalized
return []
def _normalize_content_text(content) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
normalized = []
for item in content:
if isinstance(item, dict):
normalized.append({k: v for k, v in item.items() if k != "cache_control"})
else:
normalized.append(item)
return json.dumps(normalized, ensure_ascii=False)
return json.dumps(content, ensure_ascii=False)
def _msg_hash(msg: dict) -> str:
role = msg.get("role", "")
text = _normalize_content_text(msg.get("content"))
return f"{role}:{text[:500]}"
def _is_prefix_of(shorter: list[str], longer: list[str]) -> bool:
if not shorter or len(longer) < len(shorter):
return False
return all(shorter[i] == longer[i] for i in range(len(shorter)))
def _find_prev_same_model(entries: list[dict], idx: int) -> int:
target = entries[idx]
target_body = target.get("request", {}).get("body") or {}
target_hashes = [_msg_hash(msg) for msg in _normalize_messages_for_diff(target_body)]
best_idx = -1
best_len = 0
for i in range(idx - 1, -1, -1):
candidate_body = entries[i].get("request", {}).get("body") or {}
candidate_hashes = [_msg_hash(msg) for msg in _normalize_messages_for_diff(candidate_body)]
if candidate_hashes and _is_prefix_of(candidate_hashes, target_hashes):
if len(candidate_hashes) > best_len:
best_len = len(candidate_hashes)
best_idx = i
if best_idx >= 0:
return best_idx
target_model = target_body.get("model")
for i in range(idx - 1, -1, -1):
candidate_body = entries[i].get("request", {}).get("body") or {}
candidate_model = candidate_body.get("model")
if candidate_model == target_model:
return i
return -1
def _score_diff_messages(old_msgs: list[dict], new_msgs: list[dict]) -> int:
prefix = 0
while prefix < len(old_msgs) and prefix < len(new_msgs):
if _msg_hash(old_msgs[prefix]) != _msg_hash(new_msgs[prefix]):
break
prefix += 1
old_tail = old_msgs[prefix:]
new_tail = new_msgs[prefix:]
if not old_tail and not new_tail:
return 0
max_len = 0
for msg in old_tail + new_tail:
max_len = max(max_len, len(_normalize_content_text(msg.get("content"))))
return max_len
def _pick_real_trace_file_for_diff() -> tuple[Path, int]:
traces_dir = Path(__file__).parent.parent / ".traces"
trace_files = sorted(traces_dir.glob("trace_*.jsonl"))
best_path = None
best_idx = -1
best_score = -1
for path in trace_files:
file_size = path.stat().st_size
if file_size == 0 or file_size > 5_000_000:
continue
entries = _load_entries(path)
if len(entries) < 4:
continue
for idx in range(1, len(entries)):
prev_idx = _find_prev_same_model(entries, idx)
if prev_idx < 0:
continue
old_msgs = _normalize_messages_for_diff(entries[prev_idx].get("request", {}).get("body") or {})
new_msgs = _normalize_messages_for_diff(entries[idx].get("request", {}).get("body") or {})
if len(new_msgs) < 2:
continue
score = _score_diff_messages(old_msgs, new_msgs)
if score > best_score:
best_score = score
best_path = path
best_idx = idx
if best_path and best_idx >= 0:
return best_path, best_idx
pytest.skip("No real multi-turn trace file with a message diff target found in .traces/")
def _extract_messages(body: dict | None) -> list[str]:
if not body:
return []
texts: list[str] = []
if isinstance(body.get("messages"), list):
for msg in body["messages"]:
content = msg.get("content")
if isinstance(content, str):
texts.append(content)
elif isinstance(content, list):
for block in content:
if isinstance(block, dict) and isinstance(block.get("text"), str):
texts.append(block["text"])
if isinstance(body.get("input"), list):
for item in body["input"]:
if item.get("type") != "message":
continue
content = item.get("content")
if isinstance(content, str):
texts.append(content)
elif isinstance(content, list):
for block in content:
if isinstance(block, dict) and isinstance(block.get("text"), str):
texts.append(block["text"])
return texts
def _pick_message_term(entries: list[dict]) -> tuple[str, int]:
for idx, entry in enumerate(entries):
texts = _extract_messages(entry.get("request", {}).get("body", {}))
for text in texts:
for word in _WORD_RE.findall(text):
lw = word.lower()
if lw not in _STOPWORDS:
return lw, idx
return "model", 0
def _pick_cross_entry_term(entries: list[dict]) -> str:
entry_texts = [json.dumps(entry, ensure_ascii=False).lower() for entry in entries]
by_entry_count: dict[str, int] = {}
total_counts: dict[str, int] = {}
for text in entry_texts:
seen = set()
for word in _WORD_RE.findall(text):
lw = word.lower()
if lw in _STOPWORDS:
continue
total_counts[lw] = total_counts.get(lw, 0) + text.count(lw)
seen.add(lw)
for word in seen:
by_entry_count[word] = by_entry_count.get(word, 0) + 1
scored: list[tuple[int, int, str]] = []
total_entries = len(entries)
for word, entry_hits in by_entry_count.items():
total_hits = total_counts.get(word, 0)
if entry_hits >= 2 and total_hits <= 40 and entry_hits < total_entries:
scored.append((entry_hits, total_hits, word))
if scored:
scored.sort()
return scored[0][2]
# Fallback: broad but always present in responses traces.
return "response"
@pytest.fixture(scope="module")
def trace_entries() -> tuple[Path, list[dict], str, tuple[str, int]]:
trace_file = _pick_real_trace_file()
entries = _load_entries(trace_file)
cross_term = _pick_cross_entry_term(entries)
message_term = _pick_message_term(entries)
return trace_file, entries, cross_term, message_term
@pytest.fixture(scope="module")
def html_file(trace_entries) -> Path:
trace_file, _, _, _ = trace_entries
with tempfile.TemporaryDirectory() as tmpdir:
html_path = Path(tmpdir) / "search_test_viewer.html"
_generate_html_viewer(trace_file, html_path)
html = html_path.read_text(encoding="utf-8")
# Persist file after tempdir exits for module-scoped browser fixture.
with tempfile.NamedTemporaryFile(suffix=".html", delete=False, mode="w", encoding="utf-8") as f:
f.write(html)
return Path(f.name)
@pytest.fixture(scope="module")
def browser_page(html_file):
from playwright.sync_api import sync_playwright
pw = sync_playwright().start()
browser = pw.chromium.launch(headless=True)
page = browser.new_page(viewport={"width": 1440, "height": 900})
page.goto(f"file://{html_file}")
page.wait_for_selector(".sidebar-item", timeout=10000)
yield page
browser.close()
pw.stop()
def _dispatch_find_shortcut(page, *, meta: bool, ctrl: bool) -> None:
page.evaluate(
"""([metaKey, ctrlKey]) => {
document.dispatchEvent(
new KeyboardEvent('keydown', {
key: 'f',
metaKey,
ctrlKey,
bubbles: true,
cancelable: true,
})
);
}""",
[meta, ctrl],
)
def _write_search_trace(path: Path, count: int = 1) -> None:
records = []
for idx in range(count):
records.append(
{
"timestamp": "2026-05-29T08:00:00+00:00",
"request_id": f"req_search_quote_{idx}",
"turn": idx + 1,
"duration_ms": 1200,
"request": {
"method": "POST",
"path": "/v1/messages",
"headers": {"Host": "api.anthropic.com"},
"body": {
"model": "claude-sonnet-4-6",
"messages": [
{
"role": "user",
"content": "Investigate agent routing metadata.",
}
],
"metadata": {
"subagent_type": "code-review",
"subagent_hint": "subagent_type appears in JSON tree rendering.",
},
},
},
"response": {
"status": 200,
"headers": {},
"body": {
"model": "claude-sonnet-4-6",
"content": [
{
"type": "text",
"text": "The search target is present in metadata.",
}
],
},
},
}
)
path.write_text(
"\n".join(json.dumps(record, ensure_ascii=False, separators=(",", ":")) for record in records) + "\n",
encoding="utf-8",
)
def _write_duplicate_request_id_tool_trace(path: Path) -> str:
call_id = "call_kL48GiOmxdX2R6uxH6DqTz0o"
records = [
{
"timestamp": "2026-05-29T08:00:00+00:00",
"request_id": "req_shared_search",
"turn": 1,
"duration_ms": 800,
"request": {
"method": "POST",
"path": "/v1/messages",
"headers": {"Host": "api.anthropic.com"},
"body": {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": "Check status."}],
},
},
"response": {
"status": 200,
"headers": {},
"body": {
"model": "claude-sonnet-4-6",
"content": [{"type": "text", "text": "I will inspect the repo."}],
},
},
},
{
"timestamp": "2026-05-29T08:00:01+00:00",
"request_id": "req_shared_search",
"turn": 1,
"duration_ms": 900,
"request": {
"method": "POST",
"path": "/v1/messages",
"headers": {"Host": "api.anthropic.com"},
"body": {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": "Run git status."}],
},
},
"response": {
"status": 200,
"headers": {},
"body": {
"model": "claude-sonnet-4-6",
"content": [
{
"type": "tool_use",
"id": call_id,
"name": "exec_command",
"input": {"cmd": "git status --short --branch"},
}
],
},
},
},
]
path.write_text(
"\n".join(json.dumps(record, ensure_ascii=False, separators=(",", ":")) for record in records) + "\n",
encoding="utf-8",
)
return call_id
def test_json_key_query_without_key_quotes_matches(tmp_path):
trace_path = tmp_path / "quote_trace.jsonl"
html_path = tmp_path / "quote_viewer.html"
_write_search_trace(trace_path)
_generate_html_viewer(trace_path, html_path)
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=True)
try:
page = browser.new_page(viewport={"width": 1280, "height": 760})
page.goto(f"file://{html_path}")
page.wait_for_selector(".sidebar-item", timeout=5000)
_dispatch_find_shortcut(page, meta=False, ctrl=True)
page.fill("#global-search-input", 'subagent_type: "')
page.wait_for_function("() => document.querySelector('#global-search-count')?.textContent !== '0 of 0'")
page.wait_for_function("() => document.querySelectorAll('mark.global-search-hit').length > 0")
count_text = page.inner_text("#global-search-count")
marked_text = page.locator("mark.global-search-hit.current").first.inner_text()
assert " of " in count_text
assert "0 of 0" not in count_text
assert "subagent_type" in marked_text
finally:
browser.close()
def test_global_search_distinguishes_duplicate_request_ids(tmp_path):
trace_path = tmp_path / "duplicate_request_id_trace.jsonl"
html_path = tmp_path / "duplicate_request_id_viewer.html"
call_id = _write_duplicate_request_id_tool_trace(trace_path)
_generate_html_viewer(trace_path, html_path)
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=True)
try:
page = browser.new_page(viewport={"width": 1280, "height": 760})
page.goto(f"file://{html_path}")
page.wait_for_selector(".sidebar-item", timeout=5000)
page.locator(".sidebar-item[data-idx='1']").click()
page.wait_for_function(
"callId => document.querySelector('#detail')?.innerText.includes(callId)",
arg=call_id,
)
_dispatch_find_shortcut(page, meta=False, ctrl=True)
page.fill("#global-search-input", call_id)
page.wait_for_function("() => document.querySelector('#global-search-count')?.textContent !== '0 of 0'")
page.wait_for_function(
"callId => [...document.querySelectorAll('mark.global-search-hit')].some(mark => mark.textContent === callId)",
arg=call_id,
)
count_text = page.inner_text("#global-search-count")
assert "0 of 0" not in count_text
finally:
browser.close()
def test_same_entry_search_navigation_does_not_rerender_detail(tmp_path):
trace_path = tmp_path / "same_entry_trace.jsonl"
html_path = tmp_path / "same_entry_viewer.html"
_write_search_trace(trace_path)
_generate_html_viewer(trace_path, html_path)
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=True)
try:
page = browser.new_page(viewport={"width": 1280, "height": 760})
page.goto(f"file://{html_path}")
page.wait_for_selector(".sidebar-item", timeout=5000)
page.evaluate(
"""() => {
window.__renderDetailCount = 0;
const original = window.renderDetail;
window.renderDetail = function(...args) {
window.__renderDetailCount += 1;
return original.apply(this, args);
};
}"""
)
_dispatch_find_shortcut(page, meta=False, ctrl=True)
page.fill("#global-search-input", "subagent")
page.wait_for_function("() => document.querySelectorAll('mark.global-search-hit').length > 1")
before = page.evaluate("window.__renderDetailCount")
page.keyboard.press("Enter")
page.wait_for_timeout(100)
after = page.evaluate("window.__renderDetailCount")
assert after == before
finally:
browser.close()
def test_lazy_global_search_does_not_parse_every_entry(tmp_path):
trace_path = tmp_path / "lazy_trace.jsonl"
html_path = tmp_path / "lazy_viewer.html"
_write_search_trace(trace_path, count=60)
_generate_html_viewer(trace_path, html_path)
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=True)
try:
page = browser.new_page(viewport={"width": 1280, "height": 760})
page.goto(f"file://{html_path}")
page.wait_for_selector(".sidebar-item", timeout=5000)
initial_cache_size = page.evaluate("entryCache.size")
_dispatch_find_shortcut(page, meta=False, ctrl=True)
page.fill("#global-search-input", 'subagent_type: "')
page.wait_for_function("() => document.querySelector('#global-search-count')?.textContent !== '0 of 0'")
parsed_entries = page.evaluate("entryCache.size")
assert parsed_entries < 10
assert parsed_entries <= initial_cache_size + 1
finally:
browser.close()
class TestViewerGlobalSearch:
def test_cmd_or_ctrl_f_opens_custom_search(self, browser_page):
_dispatch_find_shortcut(browser_page, meta=True, ctrl=False)
browser_page.wait_for_selector("#global-search-overlay.open", timeout=3000)
assert browser_page.evaluate("document.activeElement?.id") == "global-search-input"
browser_page.keyboard.press("Escape")
browser_page.wait_for_function(
"() => !document.querySelector('#global-search-overlay')?.classList.contains('open')"
)
_dispatch_find_shortcut(browser_page, meta=False, ctrl=True)
browser_page.wait_for_selector("#global-search-overlay.open", timeout=3000)
assert browser_page.evaluate("document.activeElement?.id") == "global-search-input"
def test_typing_highlights_and_match_counter(self, browser_page, trace_entries):
_, _, cross_term, _ = trace_entries
browser_page.fill("#global-search-input", cross_term)
browser_page.wait_for_function("() => document.querySelectorAll('mark.global-search-hit').length > 0")
count_text = browser_page.inner_text("#global-search-count")
assert " of " in count_text
assert "matches" in count_text
def test_enter_navigates_matches(self, browser_page):
before = browser_page.inner_text("#global-search-count")
browser_page.keyboard.press("Enter")
browser_page.wait_for_timeout(150)
after = browser_page.inner_text("#global-search-count")
assert before != after, f"Expected current match index to advance, got: {after}"
def test_cross_entry_navigation_switches_sidebar(self, browser_page):
start_turn = browser_page.inner_text(".sidebar-item.active .si-turn")
switched = False
for _ in range(80):
browser_page.keyboard.press("Enter")
browser_page.wait_for_timeout(80)
now_turn = browser_page.inner_text(".sidebar-item.active .si-turn")
if now_turn != start_turn:
switched = True
break
assert switched, "Expected search navigation to jump to a different sidebar entry"
def test_escape_closes_and_clears_highlights(self, browser_page):
browser_page.keyboard.press("Escape")
browser_page.wait_for_function(
"() => !document.querySelector('#global-search-overlay')?.classList.contains('open')"
)
mark_count = browser_page.evaluate("document.querySelectorAll('mark.global-search-hit').length")
assert mark_count == 0, f"Expected highlights to clear on Escape, got {mark_count}"
def test_collapsed_section_auto_expands_on_match(self, browser_page, trace_entries):
_, entries, _, message_term_info = trace_entries
message_term, _ = message_term_info
# Sidebar order can differ from raw trace order. Find an entry that renders messages.
sidebar_items = browser_page.locator(".sidebar-item")
max_items = min(sidebar_items.count(), len(entries))
found_msg_section = False
for i in range(max_items):
sidebar_items.nth(i).click()
browser_page.wait_for_timeout(120)
if browser_page.locator(".section .msg").count() > 0:
found_msg_section = True
break
assert found_msg_section, "Expected at least one sidebar entry to render message blocks"
# Collapse the messages section (identified by message blocks).
msg_section = browser_page.locator(".section", has=browser_page.locator(".msg")).first
msg_body = msg_section.locator(".section-body")
msg_section.locator(".section-header").click()
browser_page.wait_for_timeout(120)
assert "open" not in msg_body.get_attribute("class")
_dispatch_find_shortcut(browser_page, meta=False, ctrl=True)
browser_page.fill("#global-search-input", message_term)
browser_page.wait_for_function(
"""() => {
const section = [...document.querySelectorAll('.section')].find(s => s.querySelector('.msg'));
if (!section) return false;
const body = section.querySelector('.section-body');
return body && body.classList.contains('open');
}"""
)
def test_diff_overlay_content_is_scrollable(self, browser_page):
trace_file, target_idx = _pick_real_trace_file_for_diff()
with tempfile.TemporaryDirectory() as tmpdir:
html_path = Path(tmpdir) / "diff_scroll_test_viewer.html"
_generate_html_viewer(trace_file, html_path)
browser_page.set_viewport_size({"width": 1180, "height": 360})
browser_page.goto(f"file://{html_path}")
browser_page.wait_for_selector(".sidebar-item", timeout=10000)
browser_page.locator(f".sidebar-item[data-idx='{target_idx}']").click()
browser_page.wait_for_timeout(120)
browser_page.evaluate("document.querySelector('.act-btn:nth-child(3)')?.click()")
browser_page.wait_for_selector(".diff-overlay", timeout=3000)
state = browser_page.evaluate("""() => {
const overlay = document.querySelector('.diff-overlay');
const body = overlay?.querySelector('.diff-body');
const block = overlay?.querySelector('.diff-new-msg, .diff-removed-msg, .diff-modified-msg');
if (!body || !block) {
return {
hasBody: Boolean(body),
hasBlock: Boolean(block),
overlayScrollable: false,
blockScrollable: false,
overlayCanScroll: false,
blockCanScroll: false,
};
}
const bodyFiller = document.createElement('div');
bodyFiller.style.height = '900px';
body.appendChild(bodyFiller);
const blockFiller = document.createElement('div');
blockFiller.style.height = '420px';
block.appendChild(blockFiller);
const overlayScrollable = body.scrollHeight > body.clientHeight;
const blockScrollable = block.scrollHeight > block.clientHeight;
body.scrollTop = body.scrollHeight;
block.scrollTop = block.scrollHeight;
return {
hasBody: true,
hasBlock: true,
overlayScrollable,
blockScrollable,
overlayCanScroll: body.scrollTop > 0,
blockCanScroll: block.scrollTop > 0,
};
}""")
assert state["hasBody"], "Expected diff overlay body container to exist"
assert state["hasBlock"], "Expected at least one diff message block to exist"
assert state["overlayScrollable"], "Expected diff overlay body to overflow after long content is present"
assert state["overlayCanScroll"], "Expected diff overlay body to scroll when content is long"
assert state["blockScrollable"], "Expected diff message block to overflow after long content is present"
assert state["blockCanScroll"], "Expected diff message block to be scrollable"