Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 35s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 39s
Tests / e2e (pull_request) Successful in 3m59s
Tests / test (pull_request) Failing after 45m10s
112 lines
3.2 KiB
Python
112 lines
3.2 KiB
Python
"""
|
|
Tests for session analytics
|
|
|
|
Issue: #753
|
|
"""
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from agent.session_analytics import (
|
|
SessionTracker,
|
|
SessionStats,
|
|
format_stats,
|
|
get_daily_stats,
|
|
format_daily_report,
|
|
)
|
|
|
|
|
|
class TestSessionStats(unittest.TestCase):
|
|
|
|
def test_defaults(self):
|
|
stats = SessionStats(session_id="test", start_time="2026-01-01")
|
|
self.assertEqual(stats.input_tokens, 0)
|
|
self.assertEqual(stats.output_tokens, 0)
|
|
self.assertEqual(stats.tool_calls, 0)
|
|
|
|
def test_to_dict(self):
|
|
stats = SessionStats(session_id="test", start_time="2026-01-01")
|
|
d = stats.to_dict()
|
|
self.assertEqual(d["session_id"], "test")
|
|
self.assertIn("input_tokens", d)
|
|
|
|
|
|
class TestSessionTracker(unittest.TestCase):
|
|
|
|
def test_record_tokens(self):
|
|
tracker = SessionTracker("test", provider="openai")
|
|
tracker.record_tokens(100, 50)
|
|
stats = tracker.get_current_stats()
|
|
self.assertEqual(stats.input_tokens, 100)
|
|
self.assertEqual(stats.output_tokens, 50)
|
|
self.assertGreater(stats.estimated_cost_usd, 0)
|
|
|
|
def test_record_tool_call(self):
|
|
tracker = SessionTracker("test")
|
|
tracker.record_tool_call(success=True)
|
|
tracker.record_tool_call(success=False)
|
|
stats = tracker.get_current_stats()
|
|
self.assertEqual(stats.tool_calls, 2)
|
|
self.assertEqual(stats.tool_errors, 1)
|
|
|
|
def test_free_provider(self):
|
|
tracker = SessionTracker("test", provider="ollama")
|
|
tracker.record_tokens(1000, 500)
|
|
stats = tracker.get_current_stats()
|
|
self.assertEqual(stats.estimated_cost_usd, 0.0)
|
|
|
|
def test_finish(self):
|
|
tracker = SessionTracker("test")
|
|
stats = tracker.finish()
|
|
self.assertIsNotNone(stats.end_time)
|
|
self.assertGreater(stats.wall_time_seconds, 0)
|
|
|
|
|
|
class TestFormatStats(unittest.TestCase):
|
|
|
|
def test_format(self):
|
|
stats = SessionStats(
|
|
session_id="test123",
|
|
start_time="2026-01-01",
|
|
input_tokens=1000,
|
|
output_tokens=500,
|
|
total_tokens=1500,
|
|
tool_calls=5,
|
|
tool_errors=1,
|
|
wall_time_seconds=30.5,
|
|
api_calls=3
|
|
)
|
|
formatted = format_stats(stats)
|
|
self.assertIn("1,000", formatted)
|
|
self.assertIn("500", formatted)
|
|
|
|
|
|
class TestDailyStats(unittest.TestCase):
|
|
|
|
def test_empty(self):
|
|
with patch("agent.session_analytics.ANALYTICS_DIR", Path(tempfile.mkdtemp())):
|
|
stats = get_daily_stats("2020-01-01")
|
|
self.assertEqual(stats["sessions"], 0)
|
|
|
|
def test_format_report(self):
|
|
stats = {
|
|
"date": "2026-04-14",
|
|
"sessions": 10,
|
|
"total_tokens": 50000,
|
|
"total_cost_usd": 0.50,
|
|
"total_wall_time_seconds": 300,
|
|
"total_tool_calls": 100,
|
|
"total_tool_errors": 5,
|
|
"avg_tokens_per_session": 5000,
|
|
"avg_cost_per_session": 0.05,
|
|
}
|
|
report = format_daily_report(stats)
|
|
self.assertIn("10", report)
|
|
self.assertIn("50,000", report)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|