124 lines
4.4 KiB
Python
124 lines
4.4 KiB
Python
"""Tests for deterministic trajectory metrics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from swe_data_processing.metrics import canonical_training_text, extract_metrics
|
|
|
|
|
|
def _record(outputs: list[str]) -> dict:
|
|
trajectory = [{"role": "user", "content": "Fix the bug."}]
|
|
for index, output in enumerate(outputs, 1):
|
|
trajectory.extend(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"id": f"call-{index}",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "bash",
|
|
"arguments": json.dumps({"command": "pytest tests"}),
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "content": output},
|
|
]
|
|
)
|
|
return {
|
|
"trajectory_id": "sample-1",
|
|
"instance_id": "repo-1",
|
|
"resolved": 1,
|
|
"tools": [{"type": "function", "function": {"name": "bash"}}],
|
|
"trajectory": trajectory,
|
|
}
|
|
|
|
|
|
def test_canonical_text_is_stable_and_excludes_metadata() -> None:
|
|
record = _record(["exit code 0"])
|
|
record["metadata"] = {"reference_patch": {"patch": "SECRET"}}
|
|
text = canonical_training_text(record)
|
|
assert "SECRET" not in text
|
|
assert text == canonical_training_text(record)
|
|
|
|
|
|
def test_profile_keeps_source_parquet_out_of_canonical_text() -> None:
|
|
record = _record(["exit code 0"])
|
|
record["_source_parquet"] = "/data/openhands/train.parquet"
|
|
metrics = extract_metrics(record)
|
|
assert metrics["source_group"] == "openhands"
|
|
assert "/data/openhands" not in canonical_training_text(record)
|
|
|
|
|
|
def test_extracts_failure_count_type_distribution_and_streak() -> None:
|
|
record = _record(
|
|
[
|
|
"2 failed\n[The command completed with exit code 1.]",
|
|
"permission denied\nexit status 2",
|
|
"tests passed\nexit code 0",
|
|
"timeout while waiting",
|
|
]
|
|
)
|
|
metrics = extract_metrics(record)
|
|
tools = metrics["tools"]
|
|
assert tools["tool_call_count"] == 4
|
|
assert tools["failed_tool_call_count"] == 3
|
|
assert tools["longest_consecutive_failure_run"] == 2
|
|
assert tools["error_type_counts"]["nonzero_exit"] == 2
|
|
assert tools["error_type_counts"]["test_failure"] == 1
|
|
assert tools["error_type_counts"]["permission_denied"] == 1
|
|
assert tools["error_type_counts"]["timeout"] == 1
|
|
assert tools["error_positions"]["bins_5"] == [1, 1, 0, 0, 1]
|
|
|
|
|
|
def test_explicit_success_suppresses_incidental_error_words() -> None:
|
|
record = _record(["Read fixture containing 'permission denied'.\nexit code 0"])
|
|
assert extract_metrics(record)["tools"]["failed_tool_call_count"] == 0
|
|
|
|
|
|
def test_detects_broken_tool_structure() -> None:
|
|
record = _record(["exit code 0"])
|
|
call = record["trajectory"][1]["tool_calls"][0]
|
|
call["function"]["arguments"] = "{broken"
|
|
record["trajectory"].pop()
|
|
metrics = extract_metrics(record)
|
|
assert metrics["structure"]["malformed_tool_call_count"] == 1
|
|
assert metrics["structure"]["missing_tool_result_count"] == 1
|
|
assert metrics["tools"]["failed_tool_call_count"] == 1
|
|
|
|
|
|
def test_terminal_finish_without_result_is_valid() -> None:
|
|
record = _record([])
|
|
record["trajectory"].append(
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"function": {"name": "finish", "arguments": "{}"},
|
|
"id": "finish-1",
|
|
"type": "function",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
record["tools"].append({"type": "function", "function": {"name": "finish"}})
|
|
metrics = extract_metrics(record)
|
|
assert metrics["structure"]["missing_tool_result_count"] == 0
|
|
assert metrics["tools"]["failed_tool_call_count"] == 0
|
|
|
|
|
|
def test_editor_file_content_does_not_trigger_shell_patterns() -> None:
|
|
record = _record(["source code contains timeout and permission denied"])
|
|
call = record["trajectory"][1]["tool_calls"][0]
|
|
call["function"]["name"] = "str_replace_editor"
|
|
call["function"]["arguments"] = '{"command":"view","path":"src/a.py"}'
|
|
record["tools"] = [
|
|
{"type": "function", "function": {"name": "str_replace_editor"}}
|
|
]
|
|
assert extract_metrics(record)["tools"]["failed_tool_call_count"] == 0
|