Coverage for tests / unit / ai / test_telemetry.py: 100%

1"""Tests for the AI session telemetry collector.""" 

2 

3from __future__ import annotations 

4 

5from assertpy import assert_that 

6 

7from lintro.ai.telemetry import AITelemetry 

8 

9# -- Defaults ---------------------------------------------------------------- 

10 

11 

12def test_defaults_are_zero() -> None: 

13 """A fresh AITelemetry has all counters at zero.""" 

14 t = AITelemetry() 

15 assert_that(t.total_api_calls).is_equal_to(0) 

16 assert_that(t.total_input_tokens).is_equal_to(0) 

17 assert_that(t.total_output_tokens).is_equal_to(0) 

18 assert_that(t.total_cost_usd).is_equal_to(0.0) 

19 assert_that(t.total_latency_ms).is_equal_to(0.0) 

20 assert_that(t.successful_fixes).is_equal_to(0) 

21 assert_that(t.failed_fixes).is_equal_to(0) 

22 

23 

24# -- record_call ------------------------------------------------------------- 

25 

26 

27def test_record_call_increments_all_fields() -> None: 

28 """A single record_call increments api_calls and accumulates metrics.""" 

29 t = AITelemetry() 

30 t.record_call(input_tokens=100, output_tokens=50, cost=0.003, latency_ms=250.0) 

31 

32 assert_that(t.total_api_calls).is_equal_to(1) 

33 assert_that(t.total_input_tokens).is_equal_to(100) 

34 assert_that(t.total_output_tokens).is_equal_to(50) 

35 assert_that(t.total_cost_usd).is_equal_to(0.003) 

36 assert_that(t.total_latency_ms).is_equal_to(250.0) 

37 

38 

39def test_multiple_calls_accumulate() -> None: 

40 """Multiple record_call invocations accumulate correctly.""" 

41 t = AITelemetry() 

42 t.record_call(input_tokens=100, output_tokens=50, cost=0.003, latency_ms=200.0) 

43 t.record_call(input_tokens=200, output_tokens=80, cost=0.005, latency_ms=300.0) 

44 

45 assert_that(t.total_api_calls).is_equal_to(2) 

46 assert_that(t.total_input_tokens).is_equal_to(300) 

47 assert_that(t.total_output_tokens).is_equal_to(130) 

48 assert_that(t.total_cost_usd).is_close_to(0.008, tolerance=1e-9) 

49 assert_that(t.total_latency_ms).is_equal_to(500.0) 

50 

51 

52# -- to_dict ----------------------------------------------------------------- 

53 

54 

55def test_to_dict_returns_correct_dict() -> None: 

56 """to_dict returns a dictionary with all expected keys and values.""" 

57 t = AITelemetry() 

58 t.record_call(input_tokens=100, output_tokens=50, cost=0.003, latency_ms=250.5) 

59 t.successful_fixes = 3 

60 t.failed_fixes = 1 

61 

62 d = t.to_dict() 

63 assert_that(d).is_equal_to( 

64 { 

65 "api_calls": 1, 

66 "input_tokens": 100, 

67 "output_tokens": 50, 

68 "cost_usd": 0.003, 

69 "latency_ms": 250.5, 

70 "successful_fixes": 3, 

71 "failed_fixes": 1, 

72 }, 

73 ) 

74 

75 

76def test_to_dict_rounds_cost() -> None: 

77 """to_dict rounds cost_usd to 6 decimal places.""" 

78 t = AITelemetry() 

79 t.record_call(input_tokens=10, output_tokens=5, cost=0.0000001234, latency_ms=10.0) 

80 

81 d = t.to_dict() 

82 assert_that(d["cost_usd"]).is_equal_to(0.0) 

83 

84 

85def test_to_dict_rounds_latency() -> None: 

86 """to_dict rounds latency_ms to 1 decimal place.""" 

87 t = AITelemetry() 

88 t.record_call(input_tokens=10, output_tokens=5, cost=0.001, latency_ms=123.456) 

89 

90 d = t.to_dict() 

91 assert_that(d["latency_ms"]).is_equal_to(123.5)