Spaces:
Sleeping
Sleeping
File size: 5,186 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """Tests for LoadTracesStep — generic JSONL file loader."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from ace.core.context import ACEStepContext
from ace.steps.load_traces import LoadTracesStep
@pytest.fixture
def load_step():
return LoadTracesStep()
@pytest.fixture
def sample_jsonl(tmp_path: Path) -> Path:
"""Create a sample JSONL file with valid events."""
path = tmp_path / "session.jsonl"
events = [
{"type": "session", "id": "s1", "timestamp": "2026-01-01T00:00:00Z"},
{
"type": "message",
"id": "m1",
"timestamp": "2026-01-01T00:00:01Z",
"message": {"role": "user", "content": [{"type": "text", "text": "hello"}]},
},
{
"type": "message",
"id": "m2",
"timestamp": "2026-01-01T00:00:02Z",
"message": {
"role": "assistant",
"content": [{"type": "text", "text": "hi there"}],
},
},
]
path.write_text("\n".join(json.dumps(e) for e in events) + "\n")
return path
class TestLoadTracesStep:
def test_requires_provides(self, load_step: LoadTracesStep):
assert load_step.requires == frozenset({"sample"})
assert load_step.provides == frozenset({"trace"})
def test_valid_jsonl(self, load_step: LoadTracesStep, sample_jsonl: Path):
ctx = ACEStepContext(sample=str(sample_jsonl))
result = load_step(ctx)
assert isinstance(result.trace, list)
assert len(result.trace) == 3
assert result.trace[0]["type"] == "session"
assert result.trace[1]["type"] == "message"
assert result.trace[2]["type"] == "message"
def test_empty_file(self, load_step: LoadTracesStep, tmp_path: Path):
path = tmp_path / "empty.jsonl"
path.write_text("")
ctx = ACEStepContext(sample=str(path))
result = load_step(ctx)
assert result.trace == []
def test_missing_file(self, load_step: LoadTracesStep, tmp_path: Path):
path = tmp_path / "nonexistent.jsonl"
ctx = ACEStepContext(sample=str(path))
result = load_step(ctx)
assert result.trace == []
def test_skips_unparseable_lines(self, load_step: LoadTracesStep, tmp_path: Path):
path = tmp_path / "mixed.jsonl"
path.write_text(
'{"type": "session", "id": "s1"}\n'
"this is not json\n"
'{"type": "message", "id": "m1"}\n'
"\n" # blank line
"also not json\n"
)
ctx = ACEStepContext(sample=str(path))
result = load_step(ctx)
assert len(result.trace) == 2
assert result.trace[0]["type"] == "session"
assert result.trace[1]["type"] == "message"
def test_preserves_full_event_data(self, load_step: LoadTracesStep, tmp_path: Path):
"""Verify no truncation of event fields."""
long_text = "x" * 10000
event = {
"type": "message",
"id": "m1",
"timestamp": "2026-01-01T00:00:00Z",
"message": {
"role": "assistant",
"content": [
{"type": "thinking", "thinking": long_text},
{
"type": "toolCall",
"id": "tc1",
"name": "Read",
"arguments": {"file_path": "/a/b/c", "data": long_text},
},
],
},
}
path = tmp_path / "full.jsonl"
path.write_text(json.dumps(event) + "\n")
ctx = ACEStepContext(sample=str(path))
result = load_step(ctx)
assert len(result.trace) == 1
msg = result.trace[0]["message"]
assert msg["content"][0]["thinking"] == long_text
assert msg["content"][1]["arguments"]["data"] == long_text
def test_returns_new_context(self, load_step: LoadTracesStep, sample_jsonl: Path):
"""Step should return a new context, not mutate the original."""
ctx = ACEStepContext(sample=str(sample_jsonl))
result = load_step(ctx)
assert result is not ctx
assert ctx.trace is None # original unchanged
assert result.trace is not None
def test_with_real_sample_fixture(self, load_step: LoadTracesStep):
"""Test with the actual sample JSONL from the examples directory."""
sample = Path(__file__).resolve().parents[1] / "examples" / "openclaw"
jsonl_files = list(sample.glob("*.jsonl"))
if not jsonl_files:
pytest.skip("No sample JSONL files in examples/openclaw/")
ctx = ACEStepContext(sample=str(jsonl_files[0]))
result = load_step(ctx)
assert isinstance(result.trace, list)
assert len(result.trace) > 0
# Should contain message events (sample file may have corrupted first line)
types = {e.get("type") for e in result.trace}
assert "message" in types
|