main
py 560 lines 21.9 KB
Raw
1 import json
2 import tempfile
3 import unittest
4 from pathlib import Path
5
6 import scripts.track_token_usage as track_token_usage
7
8
9 class TrackTokenUsageTests(unittest.TestCase):
10 def test_main_estimates_tokens_and_appends_record(self) -> None:
11 tests_root = Path(__file__).resolve().parent
12 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
13 base = Path(tmpdir)
14 prompt_path = base / "prompt.txt"
15 output_path = base / "output.md"
16 usage_file = base / "data" / "metrics" / "token-usage.jsonl"
17 prompt_path.write_text("x" * 40, encoding="utf-8")
18 output_path.write_text("y" * 20, encoding="utf-8")
19
20 exit_code = track_token_usage.main(
21 [
22 "--stage",
23 "analysis",
24 "--source",
25 "copilot-cli",
26 "--model",
27 "copilot-default",
28 "--current-datetime",
29 "2026-05-19T08:00:00Z",
30 "--week",
31 "2026-W21",
32 "--prompt-file",
33 str(prompt_path),
34 "--output-file",
35 str(output_path),
36 "--usage-file",
37 str(usage_file),
38 ]
39 )
40
41 self.assertEqual(exit_code, 0)
42 records = [
43 json.loads(line)
44 for line in usage_file.read_text(encoding="utf-8").splitlines()
45 if line.strip()
46 ]
47 self.assertEqual(len(records), 1)
48 record = records[0]
49 self.assertEqual(record["stage"], "analysis")
50 self.assertEqual(record["source"], "copilot-cli")
51 self.assertEqual(record["model"], "copilot-default")
52 self.assertEqual(record["week"], "2026-W21")
53 self.assertEqual(record["input_tokens"], 10)
54 self.assertEqual(record["output_tokens"], 5)
55 self.assertEqual(record["total_tokens"], 15)
56 self.assertEqual(record["cost_usd"], 0.000105)
57 self.assertTrue(record["estimated"])
58
59 def test_main_uses_explicit_tokens_when_provided(self) -> None:
60 tests_root = Path(__file__).resolve().parent
61 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
62 base = Path(tmpdir)
63 usage_file = base / "token-usage.jsonl"
64
65 exit_code = track_token_usage.main(
66 [
67 "--stage",
68 "reskill",
69 "--source",
70 "github-models",
71 "--model",
72 "gpt-5.4-mini",
73 "--current-datetime",
74 "2026-05-19T08:00:00Z",
75 "--input-tokens",
76 "1000",
77 "--output-tokens",
78 "250",
79 "--usage-file",
80 str(usage_file),
81 ]
82 )
83
84 self.assertEqual(exit_code, 0)
85 record = json.loads(usage_file.read_text(encoding="utf-8").strip())
86 self.assertEqual(record["input_tokens"], 1000)
87 self.assertEqual(record["output_tokens"], 250)
88 self.assertEqual(record["total_tokens"], 1250)
89 self.assertEqual(record["week"], "2026-W21")
90 self.assertEqual(record["cost_usd"], 0.001875)
91 self.assertFalse(record["estimated"])
92
93 def test_input_manifest_validation_fails_when_final_usage_differs_by_more_than_10_percent(
94 self,
95 ) -> None:
96 tests_root = Path(__file__).resolve().parent
97 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
98 base = Path(tmpdir)
99 usage_file = base / "token-usage.jsonl"
100 manifest = base / "analysis-input-manifest.json"
101 manifest.write_text(
102 json.dumps(
103 {
104 "schema_version": "analysis_input_manifest_v1",
105 "rendered_prompt_estimate": {
106 "tokens": 1000,
107 "bytes": 4000,
108 "checksum_sha256": "abc",
109 },
110 "prompt_within_budget": True,
111 "degraded": False,
112 }
113 ),
114 encoding="utf-8",
115 )
116
117 exit_code = track_token_usage.main(
118 [
119 "--stage",
120 "analysis",
121 "--source",
122 "copilot-cli",
123 "--model",
124 "copilot-default",
125 "--current-datetime",
126 "2026-05-19T08:00:00Z",
127 "--input-tokens",
128 "1200",
129 "--output-tokens",
130 "1",
131 "--input-manifest",
132 str(manifest),
133 "--usage-file",
134 str(usage_file),
135 ]
136 )
137
138 self.assertEqual(exit_code, 1)
139 self.assertFalse(usage_file.exists())
140
141 def test_input_manifest_validation_accepts_exact_10_percent_low_estimate_against_final_usage(
142 self,
143 ) -> None:
144 tests_root = Path(__file__).resolve().parent
145 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
146 base = Path(tmpdir)
147 usage_file = base / "token-usage.jsonl"
148 manifest = base / "analysis-input-manifest.json"
149 manifest.write_text(
150 json.dumps(
151 {
152 "schema_version": "analysis_input_manifest_v1",
153 "rendered_prompt_estimate": {
154 "tokens": 900,
155 "bytes": 3600,
156 "checksum_sha256": "abc",
157 },
158 "prompt_within_budget": True,
159 "degraded": False,
160 }
161 ),
162 encoding="utf-8",
163 )
164
165 exit_code = track_token_usage.main(
166 [
167 "--stage",
168 "analysis",
169 "--source",
170 "copilot-cli",
171 "--model",
172 "copilot-default",
173 "--current-datetime",
174 "2026-05-19T08:00:00Z",
175 "--input-tokens",
176 "1000",
177 "--output-tokens",
178 "1",
179 "--input-manifest",
180 str(manifest),
181 "--usage-file",
182 str(usage_file),
183 ]
184 )
185
186 self.assertEqual(exit_code, 0)
187 record = json.loads(usage_file.read_text(encoding="utf-8").strip())
188 validation = record["input_manifest_validation"]
189 self.assertTrue(validation["within_10_percent"])
190 self.assertEqual(validation["delta_ratio"], 0.1)
191
192 def test_input_manifest_validation_records_degraded_over_budget_compaction_reason(self) -> None:
193 tests_root = Path(__file__).resolve().parent
194 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
195 base = Path(tmpdir)
196 usage_file = base / "token-usage.jsonl"
197 manifest = base / "analysis-input-manifest.json"
198 manifest.write_text(
199 json.dumps(
200 {
201 "schema_version": "analysis_input_manifest_v1",
202 "prompt_tokens": 1000,
203 "prompt_within_budget": False,
204 "degraded": True,
205 "degradation_reason": "Prompt was deterministically compacted to fit the configured token budget.",
206 }
207 ),
208 encoding="utf-8",
209 )
210
211 exit_code = track_token_usage.main(
212 [
213 "--stage",
214 "analysis",
215 "--source",
216 "copilot-cli",
217 "--model",
218 "copilot-default",
219 "--current-datetime",
220 "2026-05-19T08:00:00Z",
221 "--input-tokens",
222 "1200",
223 "--output-tokens",
224 "1",
225 "--input-manifest",
226 str(manifest),
227 "--usage-file",
228 str(usage_file),
229 ]
230 )
231
232 self.assertEqual(exit_code, 0)
233 record = json.loads(usage_file.read_text(encoding="utf-8").strip())
234 validation = record["input_manifest_validation"]
235 self.assertFalse(validation["within_10_percent"])
236 self.assertTrue(validation["degraded_or_compacted"])
237 self.assertIn("Manifest is degraded/compacted", validation["reason"])
238
239
240 class ParseCopilotTranscriptTests(unittest.TestCase):
241 def test_parses_input_output_tokens_pattern(self) -> None:
242 tests_root = Path(__file__).resolve().parent
243 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
244 transcript = Path(tmpdir) / "transcript.md"
245 transcript.write_text(
246 "# Copilot Session\n\nSome content here.\n\n"
247 "---\nInput tokens: 1500\nOutput tokens: 800\n",
248 encoding="utf-8",
249 )
250 result = track_token_usage.parse_copilot_transcript(transcript)
251 self.assertEqual(result, (1500, 800))
252
253 def test_parses_prompt_completion_tokens_pattern(self) -> None:
254 tests_root = Path(__file__).resolve().parent
255 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
256 transcript = Path(tmpdir) / "transcript.md"
257 transcript.write_text(
258 '```json\n{"prompt_tokens": 2000, "completion_tokens": 950}\n```\n',
259 encoding="utf-8",
260 )
261 result = track_token_usage.parse_copilot_transcript(transcript)
262 self.assertEqual(result, (2000, 950))
263
264 def test_parses_tokens_used_combined_pattern(self) -> None:
265 tests_root = Path(__file__).resolve().parent
266 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
267 transcript = Path(tmpdir) / "transcript.md"
268 transcript.write_text(
269 "## Summary\nTokens used: 3000 input, 1200 output\n",
270 encoding="utf-8",
271 )
272 result = track_token_usage.parse_copilot_transcript(transcript)
273 self.assertEqual(result, (3000, 1200))
274
275 def test_parses_usage_slash_pattern(self) -> None:
276 tests_root = Path(__file__).resolve().parent
277 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
278 transcript = Path(tmpdir) / "transcript.md"
279 transcript.write_text(
280 "Usage: 500/200 tokens (input/output)\n",
281 encoding="utf-8",
282 )
283 result = track_token_usage.parse_copilot_transcript(transcript)
284 self.assertEqual(result, (500, 200))
285
286 def test_returns_none_when_no_pattern_found(self) -> None:
287 tests_root = Path(__file__).resolve().parent
288 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
289 transcript = Path(tmpdir) / "transcript.md"
290 transcript.write_text(
291 "# Just a normal transcript\nNo usage info here.\n", encoding="utf-8"
292 )
293 result = track_token_usage.parse_copilot_transcript(transcript)
294 self.assertIsNone(result)
295
296 def test_returns_none_for_missing_file(self) -> None:
297 result = track_token_usage.parse_copilot_transcript(Path("/nonexistent/path.md"))
298 self.assertIsNone(result)
299
300
301 class ParseApiResponseTests(unittest.TestCase):
302 def test_parses_openai_compatible_usage(self) -> None:
303 tests_root = Path(__file__).resolve().parent
304 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
305 response_file = Path(tmpdir) / "response.json"
306 response_file.write_text(
307 json.dumps(
308 {
309 "id": "chatcmpl-abc123",
310 "choices": [{"message": {"content": "Hello"}}],
311 "usage": {
312 "prompt_tokens": 450,
313 "completion_tokens": 120,
314 "total_tokens": 570,
315 },
316 }
317 ),
318 encoding="utf-8",
319 )
320 result = track_token_usage.parse_api_response(response_file)
321 self.assertEqual(result, (450, 120))
322
323 def test_returns_none_for_missing_usage(self) -> None:
324 tests_root = Path(__file__).resolve().parent
325 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
326 response_file = Path(tmpdir) / "response.json"
327 response_file.write_text(json.dumps({"choices": []}), encoding="utf-8")
328 result = track_token_usage.parse_api_response(response_file)
329 self.assertIsNone(result)
330
331 def test_returns_none_for_invalid_json(self) -> None:
332 tests_root = Path(__file__).resolve().parent
333 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
334 response_file = Path(tmpdir) / "response.json"
335 response_file.write_text("not json at all", encoding="utf-8")
336 result = track_token_usage.parse_api_response(response_file)
337 self.assertIsNone(result)
338
339 def test_returns_none_for_missing_file(self) -> None:
340 result = track_token_usage.parse_api_response(Path("/nonexistent/response.json"))
341 self.assertIsNone(result)
342
343
344 class TokenSourcePriorityTests(unittest.TestCase):
345 """Test the priority ordering: explicit > transcript/api > file-size estimate."""
346
347 def test_transcript_overrides_file_estimate(self) -> None:
348 tests_root = Path(__file__).resolve().parent
349 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
350 base = Path(tmpdir)
351 usage_file = base / "token-usage.jsonl"
352 prompt_path = base / "prompt.txt"
353 output_path = base / "output.md"
354 transcript = base / "transcript.md"
355 prompt_path.write_text("x" * 400, encoding="utf-8")
356 output_path.write_text("y" * 200, encoding="utf-8")
357 transcript.write_text("Input tokens: 5000\nOutput tokens: 2500\n", encoding="utf-8")
358
359 exit_code = track_token_usage.main(
360 [
361 "--stage",
362 "analysis",
363 "--source",
364 "copilot-cli",
365 "--model",
366 "claude-sonnet-4",
367 "--current-datetime",
368 "2026-05-19T08:00:00Z",
369 "--prompt-file",
370 str(prompt_path),
371 "--output-file",
372 str(output_path),
373 "--transcript",
374 str(transcript),
375 "--usage-file",
376 str(usage_file),
377 ]
378 )
379
380 self.assertEqual(exit_code, 0)
381 record = json.loads(usage_file.read_text(encoding="utf-8").strip())
382 self.assertEqual(record["input_tokens"], 5000)
383 self.assertEqual(record["output_tokens"], 2500)
384 self.assertFalse(record["estimated"])
385
386 def test_api_response_overrides_file_estimate(self) -> None:
387 tests_root = Path(__file__).resolve().parent
388 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
389 base = Path(tmpdir)
390 usage_file = base / "token-usage.jsonl"
391 prompt_path = base / "prompt.txt"
392 api_response = base / "response.json"
393 prompt_path.write_text("x" * 400, encoding="utf-8")
394 api_response.write_text(
395 json.dumps(
396 {
397 "usage": {
398 "prompt_tokens": 800,
399 "completion_tokens": 300,
400 "total_tokens": 1100,
401 }
402 }
403 ),
404 encoding="utf-8",
405 )
406
407 exit_code = track_token_usage.main(
408 [
409 "--stage",
410 "reskill",
411 "--source",
412 "github-models",
413 "--model",
414 "gpt-5.4-mini",
415 "--current-datetime",
416 "2026-05-19T08:00:00Z",
417 "--prompt-file",
418 str(prompt_path),
419 "--api-response",
420 str(api_response),
421 "--usage-file",
422 str(usage_file),
423 ]
424 )
425
426 self.assertEqual(exit_code, 0)
427 record = json.loads(usage_file.read_text(encoding="utf-8").strip())
428 self.assertEqual(record["input_tokens"], 800)
429 self.assertEqual(record["output_tokens"], 300)
430 self.assertFalse(record["estimated"])
431
432 def test_explicit_tokens_override_transcript(self) -> None:
433 tests_root = Path(__file__).resolve().parent
434 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
435 base = Path(tmpdir)
436 usage_file = base / "token-usage.jsonl"
437 transcript = base / "transcript.md"
438 transcript.write_text("Input tokens: 5000\nOutput tokens: 2500\n", encoding="utf-8")
439
440 exit_code = track_token_usage.main(
441 [
442 "--stage",
443 "analysis",
444 "--source",
445 "copilot-cli",
446 "--model",
447 "claude-sonnet-4",
448 "--current-datetime",
449 "2026-05-19T08:00:00Z",
450 "--input-tokens",
451 "9999",
452 "--output-tokens",
453 "4444",
454 "--transcript",
455 str(transcript),
456 "--usage-file",
457 str(usage_file),
458 ]
459 )
460
461 self.assertEqual(exit_code, 0)
462 record = json.loads(usage_file.read_text(encoding="utf-8").strip())
463 self.assertEqual(record["input_tokens"], 9999)
464 self.assertEqual(record["output_tokens"], 4444)
465 self.assertFalse(record["estimated"])
466
467 def test_fallback_to_estimate_when_transcript_has_no_usage(self) -> None:
468 tests_root = Path(__file__).resolve().parent
469 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
470 base = Path(tmpdir)
471 usage_file = base / "token-usage.jsonl"
472 prompt_path = base / "prompt.txt"
473 output_path = base / "output.md"
474 transcript = base / "transcript.md"
475 prompt_path.write_text("x" * 40, encoding="utf-8")
476 output_path.write_text("y" * 20, encoding="utf-8")
477 transcript.write_text("# No usage info here\n", encoding="utf-8")
478
479 exit_code = track_token_usage.main(
480 [
481 "--stage",
482 "analysis",
483 "--source",
484 "copilot-cli",
485 "--model",
486 "claude-sonnet-4",
487 "--current-datetime",
488 "2026-05-19T08:00:00Z",
489 "--prompt-file",
490 str(prompt_path),
491 "--output-file",
492 str(output_path),
493 "--transcript",
494 str(transcript),
495 "--usage-file",
496 str(usage_file),
497 ]
498 )
499
500 self.assertEqual(exit_code, 0)
501 record = json.loads(usage_file.read_text(encoding="utf-8").strip())
502 self.assertEqual(record["input_tokens"], 10)
503 self.assertEqual(record["output_tokens"], 5)
504 self.assertTrue(record["estimated"])
505
506
507 class ModelPricingTests(unittest.TestCase):
508 def test_prices_representative_current_models(self) -> None:
509 self.assertEqual(
510 track_token_usage.estimate_cost_usd("gpt-5-mini", 1_000_000, 1_000_000), 2.25
511 )
512 self.assertEqual(
513 track_token_usage.estimate_cost_usd("claude-haiku-4.5", 1_000_000, 1_000_000), 6.0
514 )
515 self.assertEqual(
516 track_token_usage.estimate_cost_usd("gemini-3-flash", 1_000_000, 1_000_000), 3.5
517 )
518 self.assertEqual(
519 track_token_usage.estimate_cost_usd("mai-code-1-flash", 1_000_000, 1_000_000), 5.25
520 )
521
522 def test_long_context_threshold_rates_apply(self) -> None:
523 self.assertEqual(track_token_usage.estimate_cost_usd("gpt-5.4", 272_000, 1_000), 0.695)
524 self.assertEqual(track_token_usage.estimate_cost_usd("gpt-5.4", 272_001, 1_000), 1.382505)
525 self.assertEqual(
526 track_token_usage.estimate_cost_usd("gemini-3.1-pro", 200_001, 1_000), 0.818004
527 )
528
529 def test_cached_and_cache_write_tokens_are_supported(self) -> None:
530 cost = track_token_usage.estimate_cost_usd(
531 "claude-sonnet-4.6",
532 input_tokens=1_000_000,
533 output_tokens=1_000_000,
534 cached_input_tokens=1_000_000,
535 cache_write_tokens=1_000_000,
536 )
537 self.assertEqual(cost, 22.05)
538
539 def test_cached_tokens_do_not_trigger_long_context_rates(self) -> None:
540 cost = track_token_usage.estimate_cost_usd(
541 "gpt-5.4",
542 input_tokens=272_000,
543 output_tokens=1_000,
544 cached_input_tokens=1,
545 )
546 self.assertEqual(cost, 0.695)
547
548 def test_unsupported_cache_write_tokens_return_unknown(self) -> None:
549 self.assertIsNone(
550 track_token_usage.estimate_cost_usd(
551 "gpt-5-mini",
552 input_tokens=1_000_000,
553 output_tokens=1_000_000,
554 cache_write_tokens=1,
555 )
556 )
557
558
559 if __name__ == "__main__":
560 unittest.main()