11from __future__ import annotations
22
33import json
4- import tempfile
5- from datetime import datetime , timezone
4+ from datetime import UTC , datetime
65from pathlib import Path
76
8- import pytest
9-
107from pythinker_code .ui .shell .stats_collector import (
118 StepRecord ,
12- UsagePeriod ,
139 collect_session_files ,
1410 compute_period_stats ,
15- compute_insights ,
1611 get_sessions_root ,
1712 parse_wire_file ,
1813)
@@ -27,10 +22,15 @@ def _make_wire(tmp_path: Path, records: list[dict]) -> Path:
2722 return p
2823
2924
30- def _status_update (ts : float , input_other : int , output : int ,
31- cache_read : int = 0 , cache_write : int = 0 ,
32- model_name : str | None = None ,
33- provider_key : str | None = None ) -> dict :
25+ def _status_update (
26+ ts : float ,
27+ input_other : int ,
28+ output : int ,
29+ cache_read : int = 0 ,
30+ cache_write : int = 0 ,
31+ model_name : str | None = None ,
32+ provider_key : str | None = None ,
33+ ) -> dict :
3434 return {
3535 "timestamp" : ts ,
3636 "message" : {
@@ -50,11 +50,14 @@ def _status_update(ts: float, input_other: int, output: int,
5050
5151
5252def test_parse_wire_extracts_steps (tmp_path ):
53- now = datetime .now (timezone .utc ).timestamp ()
54- wire = _make_wire (tmp_path , [
55- _status_update (now , 1000 , 200 ),
56- _status_update (now + 1 , 2000 , 400 ),
57- ])
53+ now = datetime .now (UTC ).timestamp ()
54+ wire = _make_wire (
55+ tmp_path ,
56+ [
57+ _status_update (now , 1000 , 200 ),
58+ _status_update (now + 1 , 2000 , 400 ),
59+ ],
60+ )
5861 session_id = "test-session"
5962 seen = set ()
6063 steps = list (parse_wire_file (wire , session_id , seen ))
@@ -64,7 +67,7 @@ def test_parse_wire_extracts_steps(tmp_path):
6467
6568
6669def test_parse_wire_deduplicates (tmp_path ):
67- now = datetime .now (timezone . utc ).timestamp ()
70+ now = datetime .now (UTC ).timestamp ()
6871 # Same timestamp and total_tokens → duplicate
6972 record = _status_update (now , 1000 , 200 )
7073 wire = _make_wire (tmp_path , [record , record ])
@@ -74,7 +77,7 @@ def test_parse_wire_deduplicates(tmp_path):
7477
7578
7679def test_parse_wire_unknown_model_defaults (tmp_path ):
77- now = datetime .now (timezone . utc ).timestamp ()
80+ now = datetime .now (UTC ).timestamp ()
7881 wire = _make_wire (tmp_path , [_status_update (now , 500 , 100 )])
7982 seen = set ()
8083 steps = list (parse_wire_file (wire , "s" , seen ))
@@ -83,11 +86,18 @@ def test_parse_wire_unknown_model_defaults(tmp_path):
8386
8487
8588def test_compute_period_stats_today (tmp_path ):
86- now = datetime .now (timezone . utc ).timestamp ()
89+ now = datetime .now (UTC ).timestamp ()
8790 steps = [
88- StepRecord (session_id = "s1" , timestamp = now , model_name = "claude-sonnet-4-5" ,
89- provider_key = "anthropic" , input_other = 1000 , output = 200 ,
90- input_cache_read = 0 , input_cache_creation = 0 ),
91+ StepRecord (
92+ session_id = "s1" ,
93+ timestamp = now ,
94+ model_name = "claude-sonnet-4-5" ,
95+ provider_key = "anthropic" ,
96+ input_other = 1000 ,
97+ output = 200 ,
98+ input_cache_read = 0 ,
99+ input_cache_creation = 0 ,
100+ ),
91101 ]
92102 stats = compute_period_stats (steps )
93103 assert stats ["all_time" ].total_messages == 1
@@ -96,15 +106,29 @@ def test_compute_period_stats_today(tmp_path):
96106
97107
98108def test_compute_period_stats_excludes_old (tmp_path ):
99- old_ts = datetime (2020 , 1 , 1 , tzinfo = timezone . utc ).timestamp ()
100- now = datetime .now (timezone . utc ).timestamp ()
109+ old_ts = datetime (2020 , 1 , 1 , tzinfo = UTC ).timestamp ()
110+ now = datetime .now (UTC ).timestamp ()
101111 steps = [
102- StepRecord (session_id = "s1" , timestamp = old_ts , model_name = "m" ,
103- provider_key = "p" , input_other = 100 , output = 50 ,
104- input_cache_read = 0 , input_cache_creation = 0 ),
105- StepRecord (session_id = "s2" , timestamp = now , model_name = "m" ,
106- provider_key = "p" , input_other = 200 , output = 100 ,
107- input_cache_read = 0 , input_cache_creation = 0 ),
112+ StepRecord (
113+ session_id = "s1" ,
114+ timestamp = old_ts ,
115+ model_name = "m" ,
116+ provider_key = "p" ,
117+ input_other = 100 ,
118+ output = 50 ,
119+ input_cache_read = 0 ,
120+ input_cache_creation = 0 ,
121+ ),
122+ StepRecord (
123+ session_id = "s2" ,
124+ timestamp = now ,
125+ model_name = "m" ,
126+ provider_key = "p" ,
127+ input_other = 200 ,
128+ output = 100 ,
129+ input_cache_read = 0 ,
130+ input_cache_creation = 0 ,
131+ ),
108132 ]
109133 stats = compute_period_stats (steps )
110134 assert stats ["all_time" ].total_messages == 2
@@ -134,6 +158,7 @@ def test_collect_session_files_finds_wires(tmp_path):
134158
135159def test_load_all_stats_returns_all_stats (tmp_path , monkeypatch ):
136160 from pythinker_code .ui .shell .stats_collector import AllStats , load_all_stats
161+
137162 monkeypatch .setattr (
138163 "pythinker_code.ui.shell.stats_collector.get_sessions_root" ,
139164 lambda : tmp_path / "sessions" ,
0 commit comments