mirrors / liaohch3--claude-tap

项目文件夹

liaohch3--claude-tap
文件
main
T
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

370 行
14 KiB
Python
原始文件 永久链接 Blame 文件历史

"""Tests for Bedrock EventStream trace normalization in the HTML viewer."""
from __future__ import annotations
import base64
import json
import pytest
from claude_tap.viewer import _normalize_record_for_viewer
pw_missing = False
try:
from playwright.sync_api import sync_playwright # noqa: F401
except ImportError:
pw_missing = True
def _bedrock_frame(payload: dict) -> str:
encoded = base64.b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode()
return "\x00\x00binary-prefix" + json.dumps({"bytes": encoded, "p": "abcdefghijk"}) + "\ufffd"
def _write_trace(trace_path, records: list[dict]) -> None:
with trace_path.open("w", encoding="utf-8") as f:
for record in records:
f.write(json.dumps(record) + "\n")
def test_normalize_record_for_viewer_decodes_bedrock_eventstream() -> None:
body = "".join(
[
_bedrock_frame(
{
"type": "message_start",
"message": {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-opus-4-6",
"content": [],
"usage": {
"input_tokens": 3,
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": 0,
"output_tokens": 0,
},
},
}
),
_bedrock_frame({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}),
_bedrock_frame({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "OK"}}),
_bedrock_frame({"type": "content_block_stop", "index": 0}),
_bedrock_frame(
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 1},
}
),
_bedrock_frame({"type": "message_stop", "amazon-bedrock-invocationMetrics": {"inputTokenCount": 3}}),
]
)
record = {
"turn": 1,
"request": {
"method": "POST",
"path": "/model/global.anthropic.claude-opus-4-6-v1/invoke-with-response-stream",
"body": {"messages": [{"role": "user", "content": [{"type": "text", "text": "ping"}]}]},
},
"response": {"status": 200, "headers": {}, "body": body},
}
normalized = json.loads(_normalize_record_for_viewer(json.dumps(record)))
assert [event["event"] for event in normalized["response"]["sse_events"]] == [
"message_start",
"content_block_start",
"content_block_delta",
"content_block_stop",
"message_delta",
"message_stop",
]
assert normalized["response"]["body"]["content"] == [{"type": "text", "text": "OK"}]
assert normalized["response"]["body"]["usage"]["input_tokens"] == 3
assert normalized["response"]["body"]["usage"]["output_tokens"] == 1
def test_normalize_record_for_viewer_decodes_bedrock_converse_stream() -> None:
body = "".join(
[
_bedrock_frame({"messageStart": {"role": "assistant"}}),
_bedrock_frame({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "OK"}}}),
_bedrock_frame({"contentBlockStop": {"contentBlockIndex": 0}}),
_bedrock_frame({"messageStop": {"stopReason": "end_turn"}}),
_bedrock_frame({"metadata": {"usage": {"inputTokens": 4, "outputTokens": 2, "totalTokens": 6}}}),
]
)
record = {
"turn": 1,
"request": {
"method": "POST",
"path": "/model/anthropic.claude-sonnet-4-20250514-v1:0/converse-stream",
"body": {"messages": [{"role": "user", "content": [{"text": "ping"}]}]},
},
"response": {"status": 200, "headers": {}, "body": body},
}
normalized = json.loads(_normalize_record_for_viewer(json.dumps(record)))
assert [event["event"] for event in normalized["response"]["sse_events"]] == [
"message_start",
"content_block_delta",
"content_block_stop",
"message_delta",
"message_delta",
]
assert normalized["response"]["body"]["content"] == [{"type": "text", "text": "OK"}]
assert normalized["response"]["body"]["stop_reason"] == "end_turn"
assert normalized["response"]["body"]["usage"]["input_tokens"] == 4
assert normalized["response"]["body"]["usage"]["output_tokens"] == 2
assert normalized["response"]["body"]["usage"]["total_tokens"] == 6
def test_normalize_record_for_viewer_decodes_raw_bedrock_converse_stream_payloads() -> None:
body = "".join(
json.dumps(payload, separators=(",", ":"))
for payload in (
{"messageStart": {"role": "assistant"}},
{"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "OK"}}},
{"metadata": {"usage": {"inputTokens": 4, "outputTokens": 2, "totalTokens": 6}}},
)
)
record = {
"turn": 1,
"request": {
"method": "POST",
"path": "/model/anthropic.claude-sonnet-4-20250514-v1:0/converse-stream",
"body": {"messages": [{"role": "user", "content": [{"text": "ping"}]}]},
},
"response": {"status": 200, "headers": {}, "body": body},
}
normalized = json.loads(_normalize_record_for_viewer(json.dumps(record)))
assert normalized["response"]["body"]["content"] == [{"type": "text", "text": "OK"}]
assert normalized["response"]["body"]["usage"]["input_tokens"] == 4
assert normalized["response"]["body"]["usage"]["output_tokens"] == 2
def test_normalize_record_for_viewer_preserves_bedrock_reasoning_signature() -> None:
body = "".join(
[
_bedrock_frame({"messageStart": {"role": "assistant"}}),
_bedrock_frame(
{
"contentBlockDelta": {
"contentBlockIndex": 0,
"delta": {"reasoningContent": {"text": "thinking", "signature": "sig-123"}},
}
}
),
]
)
record = {
"turn": 1,
"request": {
"method": "POST",
"path": "/model/anthropic.claude-sonnet-4-20250514-v1:0/converse-stream",
"body": {"messages": [{"role": "user", "content": [{"text": "ping"}]}]},
},
"response": {"status": 200, "headers": {}, "body": body},
}
normalized = json.loads(_normalize_record_for_viewer(json.dumps(record)))
delta = normalized["response"]["sse_events"][1]["data"]["delta"]
assert delta == {"type": "thinking_delta", "thinking": "thinking", "signature": "sig-123"}
assert normalized["response"]["body"]["content"] == [
{"type": "thinking", "thinking": "thinking", "signature": "sig-123"}
]
@pytest.mark.skipif(pw_missing, reason="playwright not installed")
def test_bedrock_invoke_path_is_primary_filter(tmp_path) -> None:
from playwright.sync_api import sync_playwright
from claude_tap.viewer import _generate_html_viewer
bedrock_path = "/model/global.anthropic.claude-opus-4-6-v1/invoke-with-response-stream"
paths = [
bedrock_path,
"/mcp-registry/v0/servers",
"/inference-profiles",
"/auxiliary/one",
]
trace_path = tmp_path / "trace.jsonl"
_write_trace(
trace_path,
[
{
"timestamp": f"2026-04-27T09:15:{turn:02d}+00:00",
"request_id": f"req_{turn}",
"turn": turn,
"duration_ms": 100,
"request": {
"method": "POST" if path == bedrock_path else "GET",
"path": path,
"headers": {},
"body": {
"messages": [{"role": "user", "content": [{"type": "text", "text": "ping"}]}],
}
if path == bedrock_path
else None,
},
"response": {"status": 200, "headers": {}, "body": {"content": [], "usage": {}}},
}
for turn, path in enumerate(paths, 1)
],
)
html_path = tmp_path / "trace.html"
_generate_html_viewer(trace_path, html_path)
with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.goto(html_path.resolve().as_uri(), wait_until="networkidle")
chip_text = page.locator("#path-filter .filter-chip").first.inner_text()
sidebar_count = page.locator(".sidebar-item").count()
more_text = page.locator("#path-filter .filter-chip-toggle").inner_text()
browser.close()
assert "invoke-with-response-stream" in chip_text
assert sidebar_count == 1
assert "+3" in more_text
@pytest.mark.skipif(pw_missing, reason="playwright not installed")
def test_bedrock_billing_header_does_not_become_task_label(tmp_path) -> None:
from playwright.sync_api import sync_playwright
from claude_tap.viewer import _generate_html_viewer
trace_path = tmp_path / "trace.jsonl"
_write_trace(
trace_path,
[
{
"timestamp": "2026-04-27T09:15:01+00:00",
"request_id": "req_1",
"turn": 1,
"duration_ms": 100,
"request": {
"method": "POST",
"path": "/model/global.anthropic.claude-haiku-4-5-20251001-v1:0/invoke-with-response-stream",
"headers": {},
"body": {
"system": [
{
"type": "text",
"text": (
"x-anthropic-billing-header: cc_version=2.1.119.6a6; "
"cc_entrypoint=sdk-ts-e2b-runner;\n"
"You are a Claude agent, built on Anthropic's Claude Agent SDK."
),
}
],
"messages": [{"role": "user", "content": [{"type": "text", "text": "ping"}]}],
},
},
"response": {"status": 200, "headers": {}, "body": {"content": [], "usage": {}}},
}
],
)
html_path = tmp_path / "trace.html"
_generate_html_viewer(trace_path, html_path)
with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.goto(html_path.resolve().as_uri(), wait_until="networkidle")
label = page.locator(".sidebar-item .si-task").first.inner_text()
browser.close()
assert label == "Claude Agent"
@pytest.mark.skipif(pw_missing, reason="playwright not installed")
def test_bedrock_converse_response_output_and_usage_render(tmp_path) -> None:
from playwright.sync_api import sync_playwright
from claude_tap.viewer import _generate_html_viewer
trace_path = tmp_path / "trace.jsonl"
_write_trace(
trace_path,
[
{
"timestamp": "2026-04-27T09:15:01+00:00",
"request_id": "req_1",
"turn": 1,
"duration_ms": 100,
"request": {
"method": "POST",
"path": "/model/anthropic.claude-sonnet-4-20250514-v1:0/converse",
"headers": {},
"body": {
"messages": [{"role": "user", "content": [{"text": "ping"}]}],
},
},
"response": {
"status": 200,
"headers": {},
"body": {
"output": {
"message": {
"role": "assistant",
"content": [
{"text": "Bedrock says OK"},
{
"toolUse": {
"toolUseId": "tool-1",
"name": "lookup",
"input": {"query": "ping"},
}
},
],
}
},
"usage": {
"inputTokens": 9,
"outputTokens": 4,
"totalTokens": 13,
"cacheReadInputTokens": 3,
"cacheWriteInputTokens": 2,
},
},
},
}
],
)
html_path = tmp_path / "trace.html"
_generate_html_viewer(trace_path, html_path)
with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.goto(html_path.resolve().as_uri(), wait_until="networkidle")
result = page.evaluate(
"""
() => ({
output: getResponseOutput(entries[0]).content,
usage: getUsage(entries[0]),
})
"""
)
browser.close()
assert result["output"] == [
{"type": "text", "text": "Bedrock says OK"},
{"type": "tool_use", "id": "tool-1", "name": "lookup", "input": {"query": "ping"}},
]
assert result["usage"]["input_tokens"] == 9
assert result["usage"]["output_tokens"] == 4
assert result["usage"]["cache_read_input_tokens"] == 3
assert result["usage"]["cache_creation_input_tokens"] == 2
在新工单中引用 查看 Git Blame 复制永久链接