-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_sessionlog.py
More file actions
145 lines (110 loc) · 4.61 KB
/
Copy pathtest_sessionlog.py
File metadata and controls
145 lines (110 loc) · 4.61 KB
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
"""Tests for parsing Copilot session events into metrics."""
from __future__ import annotations
from pathlib import Path
from copilot_experiments.sessionlog import load_events, parse_metrics
def _events():
return [
{"type": "session.start", "timestamp": "2026-01-01T00:00:00.000Z", "data": {}},
{
"type": "session.model_change",
"timestamp": "2026-01-01T00:00:00.100Z",
"data": {"newModel": "gpt-5.2"},
},
{"type": "assistant.turn_start", "timestamp": "2026-01-01T00:00:00.200Z", "data": {}},
{
"type": "assistant.message",
"timestamp": "2026-01-01T00:00:00.500Z",
"data": {"text": "hi", "model": "gpt-5.2"},
},
{
"type": "tool.execution_complete",
"timestamp": "2026-01-01T00:00:00.800Z",
"data": {"success": True},
},
{
"type": "tool.execution_complete",
"timestamp": "2026-01-01T00:00:01.000Z",
"data": {"success": False},
},
{"type": "session.warning", "timestamp": "2026-01-01T00:00:01.100Z", "data": {}},
{"type": "assistant.turn_end", "timestamp": "2026-01-01T00:00:02.000Z", "data": {}},
]
def test_parse_metrics_counts():
m = parse_metrics(_events())
assert m.n_turns == 1
assert m.n_assistant_messages == 1
assert m.n_tool_calls == 2
assert m.n_tool_failures == 1
assert m.n_warnings == 1
assert "gpt-5.2" in m.models
def test_parse_metrics_duration():
m = parse_metrics(_events())
assert m.duration_s is not None
assert m.duration_s == 2.0
def test_parse_metrics_empty():
m = parse_metrics([])
assert m.n_turns == 0
assert m.duration_s is None
# --------------------------------------------------------------------------- #
# load_events / copy_events filesystem helpers
# --------------------------------------------------------------------------- #
def test_load_events_skips_blank_and_invalid_lines(tmp_path, capsys):
from copilot_experiments.sessionlog import load_events
path = tmp_path / "events.jsonl"
path.write_text(
'{"type": "session.start"}\n\n \nnot-json\n{"type": "user.message"}\n',
encoding="utf-8",
)
events = load_events(path)
assert [e["type"] for e in events] == ["session.start", "user.message"]
out, err = capsys.readouterr()
assert "Skipped 1 malformed JSON line(s)" in err
def test_load_events_missing_file_returns_empty(tmp_path):
from copilot_experiments.sessionlog import load_events
assert load_events(tmp_path / "nope.jsonl") == []
def test_copy_events_roundtrip_with_base(tmp_path):
from copilot_experiments.sessionlog import copy_events, events_path
base = tmp_path / "state"
src = events_path("sess-1", base)
src.parent.mkdir(parents=True)
src.write_text('{"type": "session.start"}\n', encoding="utf-8")
dest = tmp_path / "out" / "events.jsonl"
assert copy_events("sess-1", dest, base) is True
assert dest.read_text(encoding="utf-8") == '{"type": "session.start"}\n'
def test_copy_events_missing_source_returns_false(tmp_path):
from copilot_experiments.sessionlog import copy_events
dest = tmp_path / "out" / "events.jsonl"
assert copy_events("absent", dest, tmp_path / "state") is False
assert not dest.exists()
# --------------------------------------------------------------------------- #
# extract_economics: compaction_start peak tracking (no shutdown)
# --------------------------------------------------------------------------- #
def test_compaction_start_tracks_peak_without_shutdown():
from copilot_experiments.sessionlog import extract_economics
events = [
{
"type": "session.compaction_start",
"data": {
"systemTokens": 1000,
"conversationTokens": 8000,
"toolDefinitionsTokens": 500,
},
},
]
econ = extract_economics(events)
assert econ.total_tokens is None # no shutdown -> no authoritative totals
assert econ.peak_context_tokens == 9500 # 1000 + 8000 + 500
def test_load_events_with_malformed_lines(tmp_path: Path, capsys):
events_file = tmp_path / "events.jsonl"
events_file.write_text(
'{"type": "session.start"}\n'
'{"type": "malformed"\n' # bad json
'{"type": "session.end"}\n',
encoding="utf-8",
)
events = load_events(events_file)
assert len(events) == 2
assert events[0]["type"] == "session.start"
assert events[1]["type"] == "session.end"
out, err = capsys.readouterr()
assert "Skipped 1 malformed JSON line(s)" in err