test_insights.py
1 """Tests for agent/insights.py — InsightsEngine analytics and reporting.""" 2 3 import time 4 import pytest 5 from pathlib import Path 6 7 from hermes_state import SessionDB 8 from agent.insights import ( 9 InsightsEngine, 10 _estimate_cost, 11 _format_duration, 12 _bar_chart, 13 _has_known_pricing, 14 _DEFAULT_PRICING, 15 ) 16 17 18 @pytest.fixture() 19 def db(tmp_path): 20 """Create a SessionDB with a temp database file.""" 21 db_path = tmp_path / "test_insights.db" 22 session_db = SessionDB(db_path=db_path) 23 yield session_db 24 session_db.close() 25 26 27 @pytest.fixture() 28 def populated_db(db): 29 """Create a DB with realistic session data for insights testing.""" 30 now = time.time() 31 day = 86400 32 33 # Session 1: CLI, claude-sonnet, ended, 2 days ago 34 db.create_session( 35 session_id="s1", source="cli", 36 model="anthropic/claude-sonnet-4-20250514", user_id="user1", 37 ) 38 # Backdate the started_at 39 db._conn.execute("UPDATE sessions SET started_at = ? WHERE id = 's1'", (now - 2 * day,)) 40 db.end_session("s1", end_reason="user_exit") 41 db._conn.execute("UPDATE sessions SET ended_at = ? WHERE id = 's1'", (now - 2 * day + 3600,)) 42 db.update_token_counts("s1", input_tokens=50000, output_tokens=15000) 43 db.append_message("s1", role="user", content="Hello, help me fix a bug") 44 db.append_message("s1", role="assistant", content="Sure, let me look into that.") 45 db.append_message("s1", role="assistant", content="Let me search the files.", 46 tool_calls=[{"function": {"name": "search_files"}}]) 47 db.append_message("s1", role="tool", content="Found 3 matches", tool_name="search_files") 48 db.append_message("s1", role="assistant", content="Let me read the file.", 49 tool_calls=[{"function": {"name": "read_file"}}]) 50 db.append_message("s1", role="tool", content="file contents...", tool_name="read_file") 51 db.append_message("s1", role="assistant", content="I found the bug. Let me fix it.", 52 tool_calls=[{"function": {"name": "patch"}}]) 53 db.append_message("s1", role="tool", content="patched successfully", tool_name="patch") 54 db.append_message( 55 "s1", 56 role="assistant", 57 content="Let me load the PR workflow skill.", 58 tool_calls=[{"function": {"name": "skill_view", "arguments": '{"name":"github-pr-workflow"}'}}], 59 ) 60 db.append_message("s1", role="user", content="Thanks!") 61 db.append_message("s1", role="assistant", content="You're welcome!") 62 63 # Session 2: Telegram, gpt-4o, ended, 5 days ago 64 db.create_session( 65 session_id="s2", source="telegram", 66 model="gpt-4o", user_id="user1", 67 ) 68 db._conn.execute("UPDATE sessions SET started_at = ? WHERE id = 's2'", (now - 5 * day,)) 69 db.end_session("s2", end_reason="timeout") 70 db._conn.execute("UPDATE sessions SET ended_at = ? WHERE id = 's2'", (now - 5 * day + 1800,)) 71 db.update_token_counts("s2", input_tokens=20000, output_tokens=8000) 72 db.append_message("s2", role="user", content="Search the web for something") 73 db.append_message("s2", role="assistant", content="Searching...", 74 tool_calls=[{"function": {"name": "web_search"}}]) 75 db.append_message("s2", role="tool", content="results...", tool_name="web_search") 76 db.append_message("s2", role="assistant", content="Here's what I found") 77 78 # Session 3: CLI, deepseek-chat, ended, 10 days ago 79 db.create_session( 80 session_id="s3", source="cli", 81 model="deepseek-chat", user_id="user1", 82 ) 83 db._conn.execute("UPDATE sessions SET started_at = ? WHERE id = 's3'", (now - 10 * day,)) 84 db.end_session("s3", end_reason="user_exit") 85 db._conn.execute("UPDATE sessions SET ended_at = ? WHERE id = 's3'", (now - 10 * day + 7200,)) 86 db.update_token_counts("s3", input_tokens=100000, output_tokens=40000) 87 db.append_message("s3", role="user", content="Run this terminal command") 88 db.append_message("s3", role="assistant", content="Running...", 89 tool_calls=[{"function": {"name": "terminal"}}]) 90 db.append_message("s3", role="tool", content="output...", tool_name="terminal") 91 db.append_message("s3", role="assistant", content="Let me run another", 92 tool_calls=[{"function": {"name": "terminal"}}]) 93 db.append_message("s3", role="tool", content="more output...", tool_name="terminal") 94 db.append_message("s3", role="assistant", content="And search files", 95 tool_calls=[{"function": {"name": "search_files"}}]) 96 db.append_message("s3", role="tool", content="found stuff", tool_name="search_files") 97 db.append_message( 98 "s3", 99 role="assistant", 100 content="Load the debugging skill.", 101 tool_calls=[{"function": {"name": "skill_view", "arguments": '{"name":"systematic-debugging"}'}}], 102 ) 103 104 # Session 4: Discord, same model as s1, ended, 1 day ago 105 db.create_session( 106 session_id="s4", source="discord", 107 model="anthropic/claude-sonnet-4-20250514", user_id="user2", 108 ) 109 db._conn.execute("UPDATE sessions SET started_at = ? WHERE id = 's4'", (now - 1 * day,)) 110 db.end_session("s4", end_reason="user_exit") 111 db._conn.execute("UPDATE sessions SET ended_at = ? WHERE id = 's4'", (now - 1 * day + 900,)) 112 db.update_token_counts("s4", input_tokens=10000, output_tokens=5000) 113 db.append_message("s4", role="user", content="Quick question") 114 db.append_message("s4", role="assistant", content="Sure, go ahead") 115 db.append_message( 116 "s4", 117 role="assistant", 118 content="Load and update GitHub skills.", 119 tool_calls=[ 120 {"function": {"name": "skill_view", "arguments": '{"name":"github-pr-workflow"}'}}, 121 {"function": {"name": "skill_manage", "arguments": '{"name":"github-code-review"}'}}, 122 ], 123 ) 124 125 # Session 5: Old session, 45 days ago (should be excluded from 30-day window) 126 db.create_session( 127 session_id="s_old", source="cli", 128 model="gpt-4o-mini", user_id="user1", 129 ) 130 db._conn.execute("UPDATE sessions SET started_at = ? WHERE id = 's_old'", (now - 45 * day,)) 131 db.end_session("s_old", end_reason="user_exit") 132 db._conn.execute("UPDATE sessions SET ended_at = ? WHERE id = 's_old'", (now - 45 * day + 600,)) 133 db.update_token_counts("s_old", input_tokens=5000, output_tokens=2000) 134 db.append_message("s_old", role="user", content="old message") 135 db.append_message("s_old", role="assistant", content="old reply") 136 137 db._conn.commit() 138 return db 139 140 141 class TestHasKnownPricing: 142 def test_known_commercial_model(self): 143 assert _has_known_pricing("gpt-4o", provider="openai") is True 144 assert _has_known_pricing("anthropic/claude-sonnet-4-20250514") is True 145 assert _has_known_pricing("gpt-4.1", provider="openai") is True 146 147 def test_unknown_custom_model(self): 148 assert _has_known_pricing("FP16_Hermes_4.5") is False 149 assert _has_known_pricing("my-custom-model") is False 150 assert _has_known_pricing("glm-5") is False 151 assert _has_known_pricing("") is False 152 assert _has_known_pricing(None) is False 153 154 def test_heuristic_matched_models_are_not_considered_known(self): 155 assert _has_known_pricing("some-opus-model") is False 156 assert _has_known_pricing("future-sonnet-v2") is False 157 158 159 class TestEstimateCost: 160 def test_basic_cost(self): 161 cost, status = _estimate_cost( 162 "anthropic/claude-sonnet-4-20250514", 163 1_000_000, 164 1_000_000, 165 provider="anthropic", 166 ) 167 assert status == "estimated" 168 assert cost == pytest.approx(18.0, abs=0.01) 169 170 def test_zero_tokens(self): 171 cost, status = _estimate_cost("gpt-4o", 0, 0, provider="openai") 172 assert status == "estimated" 173 assert cost == 0.0 174 175 def test_cache_aware_usage(self): 176 cost, status = _estimate_cost( 177 "anthropic/claude-sonnet-4-20250514", 178 1000, 179 500, 180 cache_read_tokens=2000, 181 cache_write_tokens=400, 182 provider="anthropic", 183 ) 184 assert status == "estimated" 185 expected = (1000 * 3.0 + 500 * 15.0 + 2000 * 0.30 + 400 * 3.75) / 1_000_000 186 assert cost == pytest.approx(expected, abs=0.0001) 187 188 189 # ========================================================================= 190 # Format helpers 191 # ========================================================================= 192 193 class TestFormatDuration: 194 def test_seconds(self): 195 assert _format_duration(45) == "45s" 196 197 def test_minutes(self): 198 assert _format_duration(300) == "5m" 199 200 def test_hours_with_minutes(self): 201 result = _format_duration(5400) # 1.5 hours 202 assert result == "1h 30m" 203 204 def test_exact_hours(self): 205 assert _format_duration(7200) == "2h" 206 207 def test_days(self): 208 result = _format_duration(172800) # 2 days 209 assert result == "2.0d" 210 211 212 class TestBarChart: 213 def test_basic_bars(self): 214 bars = _bar_chart([10, 5, 0, 20], max_width=10) 215 assert len(bars) == 4 216 assert len(bars[3]) == 10 # max value gets full width 217 assert len(bars[0]) == 5 # half of max 218 assert bars[2] == "" # zero gets empty 219 220 def test_empty_values(self): 221 bars = _bar_chart([], max_width=10) 222 assert bars == [] 223 224 def test_all_zeros(self): 225 bars = _bar_chart([0, 0, 0], max_width=10) 226 assert all(b == "" for b in bars) 227 228 def test_single_value(self): 229 bars = _bar_chart([5], max_width=10) 230 assert len(bars) == 1 231 assert len(bars[0]) == 10 232 233 234 # ========================================================================= 235 # InsightsEngine — empty DB 236 # ========================================================================= 237 238 class TestInsightsEmpty: 239 def test_empty_db_returns_empty_report(self, db): 240 engine = InsightsEngine(db) 241 report = engine.generate(days=30) 242 assert report["empty"] is True 243 assert report["overview"] == {} 244 245 def test_empty_db_terminal_format(self, db): 246 engine = InsightsEngine(db) 247 report = engine.generate(days=30) 248 text = engine.format_terminal(report) 249 assert "No sessions found" in text 250 251 def test_empty_db_gateway_format(self, db): 252 engine = InsightsEngine(db) 253 report = engine.generate(days=30) 254 text = engine.format_gateway(report) 255 assert "No sessions found" in text 256 257 258 # ========================================================================= 259 # InsightsEngine — populated DB 260 # ========================================================================= 261 262 class TestInsightsPopulated: 263 def test_generate_returns_all_sections(self, populated_db): 264 engine = InsightsEngine(populated_db) 265 report = engine.generate(days=30) 266 267 assert report["empty"] is False 268 assert "overview" in report 269 assert "models" in report 270 assert "platforms" in report 271 assert "tools" in report 272 assert "activity" in report 273 assert "top_sessions" in report 274 275 def test_overview_session_count(self, populated_db): 276 engine = InsightsEngine(populated_db) 277 report = engine.generate(days=30) 278 overview = report["overview"] 279 280 # s1, s2, s3, s4 are within 30 days; s_old is 45 days ago 281 assert overview["total_sessions"] == 4 282 283 def test_overview_token_totals(self, populated_db): 284 engine = InsightsEngine(populated_db) 285 report = engine.generate(days=30) 286 overview = report["overview"] 287 288 expected_input = 50000 + 20000 + 100000 + 10000 289 expected_output = 15000 + 8000 + 40000 + 5000 290 assert overview["total_input_tokens"] == expected_input 291 assert overview["total_output_tokens"] == expected_output 292 assert overview["total_tokens"] == expected_input + expected_output 293 294 def test_overview_cost_positive(self, populated_db): 295 engine = InsightsEngine(populated_db) 296 report = engine.generate(days=30) 297 assert report["overview"]["estimated_cost"] > 0 298 299 def test_overview_duration_stats(self, populated_db): 300 engine = InsightsEngine(populated_db) 301 report = engine.generate(days=30) 302 overview = report["overview"] 303 304 # All 4 sessions have durations 305 assert overview["total_hours"] > 0 306 assert overview["avg_session_duration"] > 0 307 308 def test_model_breakdown(self, populated_db): 309 engine = InsightsEngine(populated_db) 310 report = engine.generate(days=30) 311 models = report["models"] 312 313 # Should have 3 distinct models (claude-sonnet x2, gpt-4o, deepseek-chat) 314 model_names = [m["model"] for m in models] 315 assert "claude-sonnet-4-20250514" in model_names 316 assert "gpt-4o" in model_names 317 assert "deepseek-chat" in model_names 318 319 # Claude-sonnet has 2 sessions (s1 + s4) 320 claude = next(m for m in models if "claude-sonnet" in m["model"]) 321 assert claude["sessions"] == 2 322 323 def test_platform_breakdown(self, populated_db): 324 engine = InsightsEngine(populated_db) 325 report = engine.generate(days=30) 326 platforms = report["platforms"] 327 328 platform_names = [p["platform"] for p in platforms] 329 assert "cli" in platform_names 330 assert "telegram" in platform_names 331 assert "discord" in platform_names 332 333 cli = next(p for p in platforms if p["platform"] == "cli") 334 assert cli["sessions"] == 2 # s1 + s3 335 336 def test_tool_breakdown(self, populated_db): 337 engine = InsightsEngine(populated_db) 338 report = engine.generate(days=30) 339 tools = report["tools"] 340 341 tool_names = [t["tool"] for t in tools] 342 assert "terminal" in tool_names 343 assert "search_files" in tool_names 344 assert "read_file" in tool_names 345 assert "patch" in tool_names 346 assert "web_search" in tool_names 347 348 # terminal was used 2x in s3 349 terminal = next(t for t in tools if t["tool"] == "terminal") 350 assert terminal["count"] == 2 351 352 # Percentages should sum to ~100% 353 total_pct = sum(t["percentage"] for t in tools) 354 assert total_pct == pytest.approx(100.0, abs=0.1) 355 356 def test_skill_breakdown(self, populated_db): 357 engine = InsightsEngine(populated_db) 358 report = engine.generate(days=30) 359 skills = report["skills"] 360 361 assert skills["summary"]["distinct_skills_used"] == 3 362 assert skills["summary"]["total_skill_loads"] == 3 363 assert skills["summary"]["total_skill_edits"] == 1 364 assert skills["summary"]["total_skill_actions"] == 4 365 366 top_skill = skills["top_skills"][0] 367 assert top_skill["skill"] == "github-pr-workflow" 368 assert top_skill["view_count"] == 2 369 assert top_skill["manage_count"] == 0 370 assert top_skill["total_count"] == 2 371 assert top_skill["last_used_at"] is not None 372 373 def test_skill_breakdown_respects_days_filter(self, populated_db): 374 engine = InsightsEngine(populated_db) 375 report = engine.generate(days=3) 376 skills = report["skills"] 377 378 assert skills["summary"]["distinct_skills_used"] == 2 379 assert skills["summary"]["total_skill_loads"] == 2 380 assert skills["summary"]["total_skill_edits"] == 1 381 382 skill_names = [s["skill"] for s in skills["top_skills"]] 383 assert "systematic-debugging" not in skill_names 384 385 def test_activity_patterns(self, populated_db): 386 engine = InsightsEngine(populated_db) 387 report = engine.generate(days=30) 388 activity = report["activity"] 389 390 assert len(activity["by_day"]) == 7 391 assert len(activity["by_hour"]) == 24 392 assert activity["active_days"] >= 1 393 assert activity["busiest_day"] is not None 394 assert activity["busiest_hour"] is not None 395 396 def test_top_sessions(self, populated_db): 397 engine = InsightsEngine(populated_db) 398 report = engine.generate(days=30) 399 top = report["top_sessions"] 400 401 labels = [t["label"] for t in top] 402 assert "Longest session" in labels 403 assert "Most messages" in labels 404 assert "Most tokens" in labels 405 assert "Most tool calls" in labels 406 407 def test_source_filter_cli(self, populated_db): 408 engine = InsightsEngine(populated_db) 409 report = engine.generate(days=30, source="cli") 410 411 assert report["overview"]["total_sessions"] == 2 # s1, s3 412 413 def test_source_filter_telegram(self, populated_db): 414 engine = InsightsEngine(populated_db) 415 report = engine.generate(days=30, source="telegram") 416 417 assert report["overview"]["total_sessions"] == 1 # s2 418 419 def test_source_filter_nonexistent(self, populated_db): 420 engine = InsightsEngine(populated_db) 421 report = engine.generate(days=30, source="slack") 422 423 assert report["empty"] is True 424 425 def test_days_filter_short(self, populated_db): 426 engine = InsightsEngine(populated_db) 427 report = engine.generate(days=3) 428 429 # Only s1 (2 days ago) and s4 (1 day ago) should be included 430 assert report["overview"]["total_sessions"] == 2 431 432 def test_days_filter_long(self, populated_db): 433 engine = InsightsEngine(populated_db) 434 report = engine.generate(days=60) 435 436 # All 5 sessions should be included 437 assert report["overview"]["total_sessions"] == 5 438 439 440 # ========================================================================= 441 # Formatting 442 # ========================================================================= 443 444 class TestTerminalFormatting: 445 def test_terminal_format_has_sections(self, populated_db): 446 engine = InsightsEngine(populated_db) 447 report = engine.generate(days=30) 448 text = engine.format_terminal(report) 449 450 assert "Hermes Insights" in text 451 assert "Overview" in text 452 assert "Models Used" in text 453 assert "Top Tools" in text 454 assert "Top Skills" in text 455 assert "Activity Patterns" in text 456 assert "Notable Sessions" in text 457 458 def test_terminal_format_shows_tokens(self, populated_db): 459 engine = InsightsEngine(populated_db) 460 report = engine.generate(days=30) 461 text = engine.format_terminal(report) 462 463 assert "Input tokens" in text 464 assert "Output tokens" in text 465 # Cost and cache metrics are intentionally hidden (pricing was unreliable). 466 assert "Est. cost" not in text 467 assert "Cache read" not in text 468 assert "Cache write" not in text 469 470 def test_terminal_format_shows_platforms(self, populated_db): 471 engine = InsightsEngine(populated_db) 472 report = engine.generate(days=30) 473 text = engine.format_terminal(report) 474 475 # Multi-platform, so Platforms section should show 476 assert "Platforms" in text 477 assert "cli" in text 478 assert "telegram" in text 479 480 def test_terminal_format_shows_bar_chart(self, populated_db): 481 engine = InsightsEngine(populated_db) 482 report = engine.generate(days=30) 483 text = engine.format_terminal(report) 484 485 assert "█" in text # Bar chart characters 486 487 def test_terminal_format_hides_cost_for_custom_models(self, db): 488 """Cost display is hidden entirely — custom models no longer show 'N/A' either.""" 489 db.create_session(session_id="s1", source="cli", model="my-custom-model") 490 db.update_token_counts("s1", input_tokens=1000, output_tokens=500) 491 db._conn.commit() 492 493 engine = InsightsEngine(db) 494 report = engine.generate(days=30) 495 text = engine.format_terminal(report) 496 497 assert "N/A" not in text 498 assert "custom/self-hosted" not in text 499 assert "Cost" not in text 500 501 502 class TestGatewayFormatting: 503 def test_gateway_format_is_shorter(self, populated_db): 504 engine = InsightsEngine(populated_db) 505 report = engine.generate(days=30) 506 terminal_text = engine.format_terminal(report) 507 gateway_text = engine.format_gateway(report) 508 509 assert len(gateway_text) < len(terminal_text) 510 511 def test_gateway_format_has_bold(self, populated_db): 512 engine = InsightsEngine(populated_db) 513 report = engine.generate(days=30) 514 text = engine.format_gateway(report) 515 516 assert "**" in text # Markdown bold 517 518 def test_gateway_format_hides_cost(self, populated_db): 519 """Gateway format omits dollar figures and internal cache details.""" 520 engine = InsightsEngine(populated_db) 521 report = engine.generate(days=30) 522 text = engine.format_gateway(report) 523 524 assert "$" not in text 525 assert "cache" not in text.lower() 526 527 def test_gateway_format_shows_models(self, populated_db): 528 engine = InsightsEngine(populated_db) 529 report = engine.generate(days=30) 530 text = engine.format_gateway(report) 531 532 assert "Models" in text 533 assert "sessions" in text 534 535 536 # ========================================================================= 537 # Edge cases 538 # ========================================================================= 539 540 class TestEdgeCases: 541 def test_session_with_no_tokens(self, db): 542 """Sessions with zero tokens should not crash.""" 543 db.create_session(session_id="s1", source="cli", model="test-model") 544 db._conn.commit() 545 546 engine = InsightsEngine(db) 547 report = engine.generate(days=30) 548 assert report["empty"] is False 549 assert report["overview"]["total_tokens"] == 0 550 assert report["overview"]["estimated_cost"] == 0.0 551 552 def test_session_with_no_end_time(self, db): 553 """Active (non-ended) sessions should be included but duration = 0.""" 554 db.create_session(session_id="s1", source="cli", model="test-model") 555 db.update_token_counts("s1", input_tokens=1000, output_tokens=500) 556 db._conn.commit() 557 558 engine = InsightsEngine(db) 559 report = engine.generate(days=30) 560 # Session included 561 assert report["overview"]["total_sessions"] == 1 562 assert report["overview"]["total_tokens"] == 1500 563 # But no duration stats (session not ended) 564 assert report["overview"]["total_hours"] == 0 565 566 def test_session_with_no_model(self, db): 567 """Sessions with NULL model should not crash.""" 568 db.create_session(session_id="s1", source="cli") 569 db.update_token_counts("s1", input_tokens=1000, output_tokens=500) 570 db._conn.commit() 571 572 engine = InsightsEngine(db) 573 report = engine.generate(days=30) 574 assert report["empty"] is False 575 576 models = report["models"] 577 assert len(models) == 1 578 assert models[0]["model"] == "unknown" 579 assert models[0]["has_pricing"] is False 580 581 def test_custom_model_shows_zero_cost(self, db): 582 """Custom/self-hosted models should show $0 cost, not fake estimates.""" 583 db.create_session(session_id="s1", source="cli", model="FP16_Hermes_4.5") 584 db.update_token_counts("s1", input_tokens=100000, output_tokens=50000) 585 db._conn.commit() 586 587 engine = InsightsEngine(db) 588 report = engine.generate(days=30) 589 assert report["overview"]["estimated_cost"] == 0.0 590 assert "FP16_Hermes_4.5" in report["overview"]["models_without_pricing"] 591 592 models = report["models"] 593 custom = next(m for m in models if m["model"] == "FP16_Hermes_4.5") 594 assert custom["cost"] == 0.0 595 assert custom["has_pricing"] is False 596 597 def test_tool_usage_from_tool_calls_json(self, db): 598 """Tool usage should be extracted from tool_calls JSON when tool_name is NULL.""" 599 import json as _json 600 db.create_session(session_id="s1", source="cli", model="test") 601 # Assistant message with tool_calls (this is what CLI produces) 602 db.append_message("s1", role="assistant", content="Let me search", 603 tool_calls=[{"id": "call_1", "type": "function", 604 "function": {"name": "search_files", "arguments": "{}"}}]) 605 # Tool response WITHOUT tool_name (this is the CLI bug) 606 db.append_message("s1", role="tool", content="found results", 607 tool_call_id="call_1") 608 db.append_message("s1", role="assistant", content="Now reading", 609 tool_calls=[{"id": "call_2", "type": "function", 610 "function": {"name": "read_file", "arguments": "{}"}}]) 611 db.append_message("s1", role="tool", content="file content", 612 tool_call_id="call_2") 613 db.append_message("s1", role="assistant", content="And searching again", 614 tool_calls=[{"id": "call_3", "type": "function", 615 "function": {"name": "search_files", "arguments": "{}"}}]) 616 db.append_message("s1", role="tool", content="more results", 617 tool_call_id="call_3") 618 db._conn.commit() 619 620 engine = InsightsEngine(db) 621 report = engine.generate(days=30) 622 tools = report["tools"] 623 624 # Should find tools from tool_calls JSON even though tool_name is NULL 625 tool_names = [t["tool"] for t in tools] 626 assert "search_files" in tool_names 627 assert "read_file" in tool_names 628 629 # search_files was called twice 630 sf = next(t for t in tools if t["tool"] == "search_files") 631 assert sf["count"] == 2 632 633 def test_overview_pricing_sets_are_lists(self, db): 634 """models_with/without_pricing should be JSON-serializable lists.""" 635 import json as _json 636 db.create_session(session_id="s1", source="cli", model="gpt-4o") 637 db.create_session(session_id="s2", source="cli", model="my-custom") 638 db._conn.commit() 639 640 engine = InsightsEngine(db) 641 report = engine.generate(days=30) 642 overview = report["overview"] 643 644 assert isinstance(overview["models_with_pricing"], list) 645 assert isinstance(overview["models_without_pricing"], list) 646 # Should be JSON-serializable 647 _json.dumps(report["overview"]) # would raise if sets present 648 649 def test_mixed_commercial_and_custom_models(self, db): 650 """Mix of commercial and custom models: only commercial ones get costs.""" 651 db.create_session(session_id="s1", source="cli", model="anthropic/claude-sonnet-4-20250514") 652 db.update_token_counts( 653 "s1", 654 input_tokens=10000, 655 output_tokens=5000, 656 billing_provider="anthropic", 657 ) 658 db.create_session(session_id="s2", source="cli", model="my-local-llama") 659 db.update_token_counts("s2", input_tokens=10000, output_tokens=5000) 660 db._conn.commit() 661 662 engine = InsightsEngine(db) 663 report = engine.generate(days=30) 664 665 # Cost should only come from gpt-4o, not from the custom model 666 overview = report["overview"] 667 assert overview["estimated_cost"] > 0 668 assert "claude-sonnet-4-20250514" in overview["models_with_pricing"] # list now, not set 669 assert "my-local-llama" in overview["models_without_pricing"] 670 671 # Verify individual model entries 672 claude = next(m for m in report["models"] if m["model"] == "claude-sonnet-4-20250514") 673 assert claude["has_pricing"] is True 674 assert claude["cost"] > 0 675 676 llama = next(m for m in report["models"] if m["model"] == "my-local-llama") 677 assert llama["has_pricing"] is False 678 assert llama["cost"] == 0.0 679 680 def test_single_session_streak(self, db): 681 """Single session should have streak of 0 or 1.""" 682 db.create_session(session_id="s1", source="cli", model="test") 683 db._conn.commit() 684 685 engine = InsightsEngine(db) 686 report = engine.generate(days=30) 687 assert report["activity"]["max_streak"] <= 1 688 689 def test_no_tool_calls(self, db): 690 """Sessions with no tool calls should produce empty tools list.""" 691 db.create_session(session_id="s1", source="cli", model="test") 692 db.append_message("s1", role="user", content="hello") 693 db.append_message("s1", role="assistant", content="hi there") 694 db._conn.commit() 695 696 engine = InsightsEngine(db) 697 report = engine.generate(days=30) 698 assert report["tools"] == [] 699 700 def test_only_one_platform(self, db): 701 """Single-platform usage should still work.""" 702 db.create_session(session_id="s1", source="cli", model="test") 703 db._conn.commit() 704 705 engine = InsightsEngine(db) 706 report = engine.generate(days=30) 707 assert len(report["platforms"]) == 1 708 assert report["platforms"][0]["platform"] == "cli" 709 710 # Terminal format should NOT show platform section for single platform 711 text = engine.format_terminal(report) 712 # (it still shows platforms section if there's only cli and nothing else) 713 # Actually the condition is > 1 platforms OR non-cli, so single cli won't show 714 715 def test_large_days_value(self, db): 716 """Very large days value should not crash.""" 717 db.create_session(session_id="s1", source="cli", model="test") 718 db._conn.commit() 719 720 engine = InsightsEngine(db) 721 report = engine.generate(days=365) 722 assert report["empty"] is False 723 724 def test_zero_days(self, db): 725 """Zero days should return empty (nothing is in the future).""" 726 db.create_session(session_id="s1", source="cli", model="test") 727 db._conn.commit() 728 729 engine = InsightsEngine(db) 730 report = engine.generate(days=0) 731 # Depending on timing, might catch the session if created <1s ago 732 # Just verify it doesn't crash 733 assert "empty" in report