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
402 lines
16 KiB
Python
402 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from claude_tap.usage import normalize_usage
|
|
from claude_tap.viewer import _extract_metadata, _extract_request_messages, _generate_html_viewer
|
|
|
|
pw_missing = False
|
|
try:
|
|
from playwright.sync_api import sync_playwright # noqa: F401
|
|
except ImportError:
|
|
pw_missing = True
|
|
|
|
|
|
def _sse_frame(payload: dict) -> str:
|
|
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
|
|
|
|
|
def _gemini_record() -> dict:
|
|
return {
|
|
"timestamp": "2026-05-13T12:00:00+00:00",
|
|
"request_id": "req_gemini",
|
|
"turn": 1,
|
|
"duration_ms": 1234,
|
|
"request": {
|
|
"method": "POST",
|
|
"path": "/v1internal:streamGenerateContent?alt=sse",
|
|
"headers": {"Host": "cloudcode-pa.googleapis.com"},
|
|
"body": {
|
|
"model": "gemini-3-flash-preview",
|
|
"project": "test-project",
|
|
"request": {
|
|
"systemInstruction": {
|
|
"role": "user",
|
|
"parts": [
|
|
{
|
|
"text": "You are Gemini CLI, an autonomous CLI agent specializing in software engineering tasks."
|
|
}
|
|
],
|
|
},
|
|
"contents": [
|
|
{
|
|
"role": "user",
|
|
"parts": [
|
|
{"text": "<session_context>Workspace: /repo</session_context>"},
|
|
{"text": "Use shell to inspect the workspace."},
|
|
],
|
|
},
|
|
{
|
|
"role": "model",
|
|
"parts": [
|
|
{
|
|
"functionCall": {
|
|
"name": "run_shell_command",
|
|
"args": {"command": "pwd", "description": "Check current directory."},
|
|
}
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"parts": [
|
|
{
|
|
"functionResponse": {
|
|
"id": "run_shell_command_1",
|
|
"name": "run_shell_command",
|
|
"response": {"output": "Output: /repo\nProcess Group PGID: 123"},
|
|
}
|
|
}
|
|
],
|
|
},
|
|
],
|
|
"tools": [
|
|
{
|
|
"functionDeclarations": [
|
|
{
|
|
"name": "run_shell_command",
|
|
"description": "Runs a shell command.",
|
|
"parametersJsonSchema": {
|
|
"type": "object",
|
|
"properties": {"command": {"type": "string"}},
|
|
"required": ["command"],
|
|
},
|
|
}
|
|
]
|
|
}
|
|
],
|
|
},
|
|
},
|
|
},
|
|
"response": {
|
|
"status": 200,
|
|
"body": (
|
|
_sse_frame(
|
|
{
|
|
"response": {
|
|
"candidates": [
|
|
{
|
|
"content": {
|
|
"role": "model",
|
|
"parts": [
|
|
{"thought": True, "text": "I need"},
|
|
],
|
|
}
|
|
}
|
|
],
|
|
"usageMetadata": {
|
|
"promptTokenCount": 100,
|
|
"candidatesTokenCount": 4,
|
|
"cachedContentTokenCount": 40,
|
|
},
|
|
}
|
|
}
|
|
)
|
|
+ _sse_frame(
|
|
{
|
|
"response": {
|
|
"candidates": [
|
|
{
|
|
"content": {
|
|
"role": "model",
|
|
"parts": [
|
|
{"thought": True, "text": " to run a shell command."},
|
|
{
|
|
"functionCall": {
|
|
"name": "run_shell_command",
|
|
"args": {"command": "pwd"},
|
|
}
|
|
},
|
|
],
|
|
}
|
|
}
|
|
],
|
|
"usageMetadata": {
|
|
"promptTokenCount": 100,
|
|
"candidatesTokenCount": 8,
|
|
"cachedContentTokenCount": 40,
|
|
},
|
|
}
|
|
}
|
|
)
|
|
+ _sse_frame(
|
|
{
|
|
"response": {
|
|
"candidates": [
|
|
{"content": {"role": "model", "parts": [{"text": "Final OK from Gemini."}]}}
|
|
],
|
|
"usageMetadata": {
|
|
"promptTokenCount": 110,
|
|
"candidatesTokenCount": 12,
|
|
"cachedContentTokenCount": 40,
|
|
},
|
|
}
|
|
}
|
|
)
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
def test_normalize_usage_maps_gemini_usage_metadata() -> None:
|
|
usage = normalize_usage(
|
|
{
|
|
"promptTokenCount": 110,
|
|
"candidatesTokenCount": 12,
|
|
"cachedContentTokenCount": 40,
|
|
}
|
|
)
|
|
|
|
assert usage["input_tokens"] == 110
|
|
assert usage["output_tokens"] == 12
|
|
assert usage["cache_read_input_tokens"] == 40
|
|
|
|
|
|
def test_extract_request_messages_normalizes_gemini_contents() -> None:
|
|
messages = _extract_request_messages(_gemini_record()["request"]["body"])
|
|
|
|
assert [message["role"] for message in messages] == ["user", "assistant", "tool"]
|
|
assert messages[0]["content"][0]["text"].startswith("<session_context>")
|
|
assert messages[1]["content"][0] == {
|
|
"type": "tool_use",
|
|
"id": "",
|
|
"name": "run_shell_command",
|
|
"input": {"command": "pwd", "description": "Check current directory."},
|
|
}
|
|
assert messages[2]["content"][0]["type"] == "tool_result"
|
|
assert messages[2]["content"][0]["tool_use_id"] == "run_shell_command_1"
|
|
assert "Output: /repo" in messages[2]["content"][0]["content"]
|
|
|
|
|
|
def test_extract_metadata_understands_gemini_system_tools_output_and_usage() -> None:
|
|
meta = _extract_metadata(json.dumps(_gemini_record(), ensure_ascii=False))
|
|
|
|
assert meta is not None
|
|
assert meta["model"] == "gemini-3-flash-preview"
|
|
assert meta["has_system"] is True
|
|
assert meta["sys_hint"].startswith("You are Gemini CLI")
|
|
assert meta["message_count"] == 3
|
|
assert meta["tool_names"] == ["run_shell_command"]
|
|
assert meta["response_tool_names"] == ["run_shell_command"]
|
|
assert meta["input_tokens"] == 110
|
|
assert meta["output_tokens"] == 12
|
|
assert meta["cache_read_input_tokens"] == 40
|
|
|
|
|
|
def _direct_gemini_native_record() -> dict:
|
|
record = json.loads(json.dumps(_gemini_record()))
|
|
record["request"]["path"] = "/v1beta/models/gemini-3.5-flash:generateContent"
|
|
record["request"]["body"] = record["request"]["body"]["request"]
|
|
return record
|
|
|
|
|
|
def test_extract_request_messages_normalizes_direct_gemini_native_body() -> None:
|
|
messages = _extract_request_messages(_direct_gemini_native_record()["request"]["body"])
|
|
|
|
assert [message["role"] for message in messages] == ["user", "assistant", "tool"]
|
|
assert messages[1]["content"][0]["name"] == "run_shell_command"
|
|
assert messages[2]["content"][0]["tool_use_id"] == "run_shell_command_1"
|
|
|
|
|
|
def test_extract_metadata_understands_direct_gemini_native_body() -> None:
|
|
meta = _extract_metadata(json.dumps(_direct_gemini_native_record(), ensure_ascii=False))
|
|
|
|
assert meta is not None
|
|
assert meta["model"] == "gemini-3.5-flash"
|
|
assert meta["tool_names"] == ["run_shell_command"]
|
|
assert meta["response_tool_names"] == ["run_shell_command"]
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def gemini_html_file() -> Path:
|
|
trace_path = Path(tempfile.mktemp(suffix=".jsonl"))
|
|
html_path = Path(tempfile.mktemp(suffix=".html"))
|
|
trace_path.write_text(json.dumps(_gemini_record(), ensure_ascii=False) + "\n", encoding="utf-8")
|
|
_generate_html_viewer(trace_path, html_path)
|
|
yield html_path
|
|
trace_path.unlink(missing_ok=True)
|
|
html_path.unlink(missing_ok=True)
|
|
|
|
|
|
@pytest.mark.skipif(pw_missing, reason="playwright not installed")
|
|
def test_viewer_renders_gemini_semantic_sections(gemini_html_file: Path) -> None:
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=True)
|
|
page = browser.new_page()
|
|
page.goto(f"file://{gemini_html_file}", timeout=10000)
|
|
page.wait_for_selector(".sidebar-item", timeout=5000)
|
|
page.locator(".sidebar-item").first.click()
|
|
page.wait_for_selector("#detail .section", timeout=5000)
|
|
|
|
result = page.evaluate(
|
|
"""() => {
|
|
const entry = entries[0];
|
|
const body = entry.request.body;
|
|
return {
|
|
tier: pathTier(entry.request.path),
|
|
primary: isPathPrimary(entry.request.path),
|
|
system: extractSystem(body),
|
|
roles: getMessages(body).map(message => message.role),
|
|
tools: getRequestTools(body).map(toolDisplayName),
|
|
output: getResponseOutput(entry).content,
|
|
usage: getUsage(entry),
|
|
eventCount: getResponseEvents(entry).length,
|
|
detail: document.querySelector('#detail').innerText,
|
|
};
|
|
}"""
|
|
)
|
|
browser.close()
|
|
|
|
assert result["tier"] == 0
|
|
assert result["primary"] is True
|
|
assert result["system"].startswith("You are Gemini CLI")
|
|
assert result["roles"] == ["user", "assistant", "tool"]
|
|
assert result["tools"] == ["run_shell_command"]
|
|
assert [block["type"] for block in result["output"]] == ["thinking", "tool_use", "text"]
|
|
assert result["output"][0]["thinking"] == "I need to run a shell command."
|
|
assert result["output"][1]["name"] == "run_shell_command"
|
|
assert result["usage"]["input_tokens"] == 110
|
|
assert result["usage"]["output_tokens"] == 12
|
|
assert result["usage"]["cache_read_input_tokens"] == 40
|
|
assert result["eventCount"] == 3
|
|
detail = result["detail"]
|
|
assert "System Prompt" in detail
|
|
assert "Messages" in detail
|
|
assert "Tools" in detail
|
|
assert "Response" in detail
|
|
assert "You are Gemini CLI" in detail
|
|
assert "Use shell to inspect the workspace." in detail
|
|
assert "Output: /repo" in detail
|
|
assert "run_shell_command" in detail
|
|
assert "Final OK from Gemini." in detail
|
|
|
|
|
|
@pytest.mark.skipif(pw_missing, reason="playwright not installed")
|
|
def test_viewer_keeps_gemini_native_paths_visible_with_chat_completions(tmp_path: Path) -> None:
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
gemini_record = json.loads(json.dumps(_gemini_record()))
|
|
gemini_record["request"]["path"] = "/v1beta/models/gemini-3.5-flash:generateContent"
|
|
gemini_record["request"]["body"]["model"] = "gemini-3.5-flash"
|
|
|
|
chat_record = {
|
|
"timestamp": "2026-05-13T12:01:00+00:00",
|
|
"request_id": "req_grading",
|
|
"turn": 2,
|
|
"duration_ms": 456,
|
|
"request": {
|
|
"method": "POST",
|
|
"path": "/v1/chat/completions",
|
|
"headers": {"Host": "api.openai.test"},
|
|
"body": {
|
|
"model": "gemini-3.5-flash",
|
|
"messages": [{"role": "user", "content": "Grade this answer."}],
|
|
},
|
|
},
|
|
"response": {
|
|
"status": 200,
|
|
"body": {
|
|
"model": "gemini-3.5-flash",
|
|
"choices": [{"message": {"role": "assistant", "content": "score: 0"}}],
|
|
"usage": {"prompt_tokens": 10, "completion_tokens": 2},
|
|
},
|
|
},
|
|
}
|
|
|
|
trace_path = tmp_path / "mixed-gemini-native-and-chat.jsonl"
|
|
html_path = tmp_path / "mixed-gemini-native-and-chat.html"
|
|
trace_path.write_text(
|
|
"".join(json.dumps(record, ensure_ascii=False) + "\n" for record in (gemini_record, chat_record)),
|
|
encoding="utf-8",
|
|
)
|
|
_generate_html_viewer(trace_path, html_path)
|
|
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=True)
|
|
page = browser.new_page()
|
|
page.goto(f"file://{html_path}", timeout=10000)
|
|
page.wait_for_selector(".sidebar-item", timeout=5000)
|
|
|
|
result = page.evaluate(
|
|
"""() => ({
|
|
turns: document.querySelector('#stat-turns').textContent,
|
|
activePaths: Array.from(activePaths).sort(),
|
|
filteredPaths: filtered.map(getPath).sort(),
|
|
pathFilter: document.querySelector('#path-filter').innerText.replace(/\\s+/g, ' ').trim(),
|
|
})"""
|
|
)
|
|
browser.close()
|
|
|
|
assert result["turns"] == "2"
|
|
assert result["activePaths"] == [
|
|
"/v1/chat/completions",
|
|
"/v1beta/models/gemini-3.5-flash:generateContent",
|
|
]
|
|
assert result["filteredPaths"] == result["activePaths"]
|
|
assert "+1 more" not in result["pathFilter"]
|
|
|
|
|
|
@pytest.mark.skipif(pw_missing, reason="playwright not installed")
|
|
def test_viewer_renders_direct_gemini_native_body(tmp_path: Path) -> None:
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
trace_path = tmp_path / "direct-gemini-native.jsonl"
|
|
html_path = tmp_path / "direct-gemini-native.html"
|
|
trace_path.write_text(json.dumps(_direct_gemini_native_record(), ensure_ascii=False) + "\n", encoding="utf-8")
|
|
_generate_html_viewer(trace_path, html_path)
|
|
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=True)
|
|
page = browser.new_page()
|
|
page.goto(f"file://{html_path}", timeout=10000)
|
|
page.wait_for_selector(".sidebar-item", timeout=5000)
|
|
page.locator(".sidebar-item").first.click()
|
|
page.wait_for_selector("#detail .section", timeout=5000)
|
|
|
|
result = page.evaluate(
|
|
"""() => {
|
|
const entry = entries[0];
|
|
const body = entry.request.body;
|
|
return {
|
|
roles: getMessages(body).map(message => message.role),
|
|
tools: getRequestTools(body).map(toolDisplayName),
|
|
output: getResponseOutput(entry).content,
|
|
detail: document.querySelector('#detail').innerText,
|
|
};
|
|
}"""
|
|
)
|
|
browser.close()
|
|
|
|
assert result["roles"] == ["user", "assistant", "tool"]
|
|
assert result["tools"] == ["run_shell_command"]
|
|
assert result["output"][1]["name"] == "run_shell_command"
|
|
assert "unknown" not in "\n".join(result["tools"]).lower()
|
|
assert "run_shell_command" in result["detail"]
|
|
assert "Final OK from Gemini." in result["detail"]
|