|
| 1 | +""" |
| 2 | +Tests for empathy_os.cost_tracker module. |
| 3 | +
|
| 4 | +Tests cover: |
| 5 | +- CostTracker initialization and storage |
| 6 | +- Request logging and cost calculation |
| 7 | +- Daily totals tracking |
| 8 | +- Summary generation |
| 9 | +- Report formatting |
| 10 | +- Tier detection |
| 11 | +- CLI command handler |
| 12 | +""" |
| 13 | + |
| 14 | +import json |
| 15 | +import pytest |
| 16 | +from datetime import datetime, timedelta |
| 17 | +from pathlib import Path |
| 18 | +from unittest.mock import patch, MagicMock |
| 19 | + |
| 20 | +from empathy_os.cost_tracker import ( |
| 21 | + CostTracker, |
| 22 | + get_tracker, |
| 23 | + log_request, |
| 24 | + cmd_costs, |
| 25 | + MODEL_PRICING, |
| 26 | + BASELINE_MODEL, |
| 27 | + _build_model_pricing, |
| 28 | +) |
| 29 | + |
| 30 | + |
| 31 | +class TestCostTrackerInit: |
| 32 | + """Test CostTracker initialization.""" |
| 33 | + |
| 34 | + def test_creates_storage_directory(self, tmp_path): |
| 35 | + """Test that storage directory is created if it doesn't exist.""" |
| 36 | + storage_dir = tmp_path / "new_empathy_dir" |
| 37 | + tracker = CostTracker(storage_dir=str(storage_dir)) |
| 38 | + assert storage_dir.exists() |
| 39 | + assert tracker.storage_dir == storage_dir |
| 40 | + |
| 41 | + def test_loads_existing_data(self, tmp_path): |
| 42 | + """Test loading existing cost data from file.""" |
| 43 | + storage_dir = tmp_path / ".empathy" |
| 44 | + storage_dir.mkdir() |
| 45 | + costs_file = storage_dir / "costs.json" |
| 46 | + |
| 47 | + existing_data = { |
| 48 | + "requests": [{"model": "test", "timestamp": "2025-01-01T00:00:00"}], |
| 49 | + "daily_totals": {"2025-01-01": {"requests": 1}}, |
| 50 | + "created_at": "2025-01-01T00:00:00", |
| 51 | + "last_updated": "2025-01-01T00:00:00", |
| 52 | + } |
| 53 | + with open(costs_file, "w") as f: |
| 54 | + json.dump(existing_data, f) |
| 55 | + |
| 56 | + tracker = CostTracker(storage_dir=str(storage_dir)) |
| 57 | + assert len(tracker.data["requests"]) == 1 |
| 58 | + assert "2025-01-01" in tracker.data["daily_totals"] |
| 59 | + |
| 60 | + def test_handles_corrupted_file(self, tmp_path): |
| 61 | + """Test handling of corrupted JSON file.""" |
| 62 | + storage_dir = tmp_path / ".empathy" |
| 63 | + storage_dir.mkdir() |
| 64 | + costs_file = storage_dir / "costs.json" |
| 65 | + |
| 66 | + with open(costs_file, "w") as f: |
| 67 | + f.write("not valid json {{{") |
| 68 | + |
| 69 | + tracker = CostTracker(storage_dir=str(storage_dir)) |
| 70 | + assert tracker.data["requests"] == [] |
| 71 | + |
| 72 | + def test_default_data_structure(self, tmp_path): |
| 73 | + """Test default data structure is created.""" |
| 74 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 75 | + assert "requests" in tracker.data |
| 76 | + assert "daily_totals" in tracker.data |
| 77 | + assert "created_at" in tracker.data |
| 78 | + assert "last_updated" in tracker.data |
| 79 | + |
| 80 | + |
| 81 | +class TestCostCalculation: |
| 82 | + """Test cost calculation logic.""" |
| 83 | + |
| 84 | + def test_calculate_cost_known_model(self, tmp_path): |
| 85 | + """Test cost calculation for known model.""" |
| 86 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 87 | + # 1M input tokens, 1M output tokens for claude-3-haiku |
| 88 | + cost = tracker._calculate_cost("claude-3-haiku-20240307", 1_000_000, 1_000_000) |
| 89 | + # Haiku: $0.25/M input + $1.25/M output = $1.50 |
| 90 | + assert cost == pytest.approx(1.50, rel=0.01) |
| 91 | + |
| 92 | + def test_calculate_cost_unknown_model_defaults(self, tmp_path): |
| 93 | + """Test that unknown models default to capable tier pricing.""" |
| 94 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 95 | + cost = tracker._calculate_cost("unknown-model-xyz", 1_000_000, 1_000_000) |
| 96 | + # Should use capable tier pricing |
| 97 | + assert cost > 0 |
| 98 | + |
| 99 | + def test_calculate_cost_tier_alias(self, tmp_path): |
| 100 | + """Test cost calculation using tier aliases.""" |
| 101 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 102 | + cheap_cost = tracker._calculate_cost("cheap", 1_000_000, 1_000_000) |
| 103 | + capable_cost = tracker._calculate_cost("capable", 1_000_000, 1_000_000) |
| 104 | + premium_cost = tracker._calculate_cost("premium", 1_000_000, 1_000_000) |
| 105 | + |
| 106 | + # Cheap should be less than capable, which should be less than premium |
| 107 | + assert cheap_cost < capable_cost < premium_cost |
| 108 | + |
| 109 | + |
| 110 | +class TestRequestLogging: |
| 111 | + """Test request logging functionality.""" |
| 112 | + |
| 113 | + def test_log_request_creates_record(self, tmp_path): |
| 114 | + """Test that log_request creates a proper record.""" |
| 115 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 116 | + result = tracker.log_request( |
| 117 | + model="claude-3-haiku-20240307", |
| 118 | + input_tokens=1000, |
| 119 | + output_tokens=500, |
| 120 | + task_type="summarize", |
| 121 | + ) |
| 122 | + |
| 123 | + assert result["model"] == "claude-3-haiku-20240307" |
| 124 | + assert result["input_tokens"] == 1000 |
| 125 | + assert result["output_tokens"] == 500 |
| 126 | + assert result["task_type"] == "summarize" |
| 127 | + assert result["tier"] == "cheap" |
| 128 | + assert "actual_cost" in result |
| 129 | + assert "baseline_cost" in result |
| 130 | + assert "savings" in result |
| 131 | + assert result["savings"] > 0 # Haiku saves vs Opus |
| 132 | + |
| 133 | + def test_log_request_updates_daily_totals(self, tmp_path): |
| 134 | + """Test that logging updates daily totals.""" |
| 135 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 136 | + tracker.log_request("claude-3-haiku-20240307", 1000, 500, "test") |
| 137 | + tracker.log_request("claude-3-haiku-20240307", 2000, 1000, "test") |
| 138 | + |
| 139 | + today = datetime.now().strftime("%Y-%m-%d") |
| 140 | + daily = tracker.data["daily_totals"][today] |
| 141 | + assert daily["requests"] == 2 |
| 142 | + assert daily["input_tokens"] == 3000 |
| 143 | + assert daily["output_tokens"] == 1500 |
| 144 | + |
| 145 | + def test_log_request_saves_to_file(self, tmp_path): |
| 146 | + """Test that logging saves data to file.""" |
| 147 | + storage_dir = tmp_path / ".empathy" |
| 148 | + tracker = CostTracker(storage_dir=str(storage_dir)) |
| 149 | + tracker.log_request("claude-3-haiku-20240307", 1000, 500, "test") |
| 150 | + |
| 151 | + # Reload from file |
| 152 | + with open(storage_dir / "costs.json") as f: |
| 153 | + saved_data = json.load(f) |
| 154 | + |
| 155 | + assert len(saved_data["requests"]) == 1 |
| 156 | + |
| 157 | + def test_log_request_limits_stored_requests(self, tmp_path): |
| 158 | + """Test that only last 1000 requests are kept.""" |
| 159 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 160 | + |
| 161 | + # Log 1050 requests |
| 162 | + for i in range(1050): |
| 163 | + tracker.log_request("claude-3-haiku-20240307", 100, 50, f"task_{i}") |
| 164 | + |
| 165 | + assert len(tracker.data["requests"]) == 1000 |
| 166 | + |
| 167 | + def test_log_request_with_tier_override(self, tmp_path): |
| 168 | + """Test logging with explicit tier override.""" |
| 169 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 170 | + result = tracker.log_request( |
| 171 | + model="some-model", |
| 172 | + input_tokens=1000, |
| 173 | + output_tokens=500, |
| 174 | + task_type="test", |
| 175 | + tier="premium", |
| 176 | + ) |
| 177 | + assert result["tier"] == "premium" |
| 178 | + |
| 179 | + |
| 180 | +class TestTierDetection: |
| 181 | + """Test tier detection from model names.""" |
| 182 | + |
| 183 | + def test_detects_haiku_as_cheap(self, tmp_path): |
| 184 | + """Test that haiku models are detected as cheap tier.""" |
| 185 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 186 | + assert tracker._get_tier("claude-3-haiku-20240307") == "cheap" |
| 187 | + assert tracker._get_tier("claude-3-5-haiku-20241022") == "cheap" |
| 188 | + |
| 189 | + def test_detects_opus_as_premium(self, tmp_path): |
| 190 | + """Test that opus models are detected as premium tier.""" |
| 191 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 192 | + assert tracker._get_tier("claude-opus-4-5-20251101") == "premium" |
| 193 | + assert tracker._get_tier("claude-3-opus-20240229") == "premium" |
| 194 | + |
| 195 | + def test_detects_other_as_capable(self, tmp_path): |
| 196 | + """Test that other models default to capable tier.""" |
| 197 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 198 | + assert tracker._get_tier("claude-3-sonnet-20240229") == "capable" |
| 199 | + assert tracker._get_tier("gpt-4") == "capable" |
| 200 | + |
| 201 | + |
| 202 | +class TestSummary: |
| 203 | + """Test summary generation.""" |
| 204 | + |
| 205 | + def test_get_summary_empty(self, tmp_path): |
| 206 | + """Test summary with no data.""" |
| 207 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 208 | + summary = tracker.get_summary(days=7) |
| 209 | + |
| 210 | + assert summary["requests"] == 0 |
| 211 | + assert summary["actual_cost"] == 0 |
| 212 | + assert summary["savings_percent"] == 0 |
| 213 | + |
| 214 | + def test_get_summary_with_data(self, tmp_path): |
| 215 | + """Test summary with logged requests.""" |
| 216 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 217 | + tracker.log_request("claude-3-haiku-20240307", 1000, 500, "summarize") |
| 218 | + tracker.log_request("claude-3-5-sonnet-20241022", 2000, 1000, "generate") |
| 219 | + |
| 220 | + summary = tracker.get_summary(days=7) |
| 221 | + |
| 222 | + assert summary["requests"] == 2 |
| 223 | + assert summary["input_tokens"] == 3000 |
| 224 | + assert summary["output_tokens"] == 1500 |
| 225 | + assert summary["actual_cost"] > 0 |
| 226 | + assert summary["by_task"]["summarize"] == 1 |
| 227 | + assert summary["by_task"]["generate"] == 1 |
| 228 | + |
| 229 | + def test_get_summary_respects_days_filter(self, tmp_path): |
| 230 | + """Test that summary respects the days filter.""" |
| 231 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 232 | + |
| 233 | + # Add data for today |
| 234 | + tracker.log_request("claude-3-haiku-20240307", 1000, 500, "today_task") |
| 235 | + |
| 236 | + # Manually add old data |
| 237 | + old_date = (datetime.now() - timedelta(days=10)).strftime("%Y-%m-%d") |
| 238 | + tracker.data["daily_totals"][old_date] = { |
| 239 | + "requests": 100, |
| 240 | + "input_tokens": 100000, |
| 241 | + "output_tokens": 50000, |
| 242 | + "actual_cost": 1.0, |
| 243 | + "baseline_cost": 10.0, |
| 244 | + "savings": 9.0, |
| 245 | + } |
| 246 | + |
| 247 | + summary = tracker.get_summary(days=7) |
| 248 | + assert summary["requests"] == 1 # Only today's request |
| 249 | + |
| 250 | + |
| 251 | +class TestReport: |
| 252 | + """Test report generation.""" |
| 253 | + |
| 254 | + def test_get_report_format(self, tmp_path): |
| 255 | + """Test report formatting.""" |
| 256 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 257 | + tracker.log_request("claude-3-haiku-20240307", 1000, 500, "summarize") |
| 258 | + |
| 259 | + report = tracker.get_report(days=7) |
| 260 | + |
| 261 | + assert "COST TRACKING REPORT" in report |
| 262 | + assert "SUMMARY" in report |
| 263 | + assert "COSTS" in report |
| 264 | + assert "BY MODEL TIER" in report |
| 265 | + assert "BY TASK TYPE" in report |
| 266 | + |
| 267 | + def test_get_report_shows_savings(self, tmp_path): |
| 268 | + """Test that report shows savings percentage.""" |
| 269 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 270 | + tracker.log_request("claude-3-haiku-20240307", 100000, 50000, "test") |
| 271 | + |
| 272 | + report = tracker.get_report(days=7) |
| 273 | + assert "You saved:" in report |
| 274 | + assert "%" in report |
| 275 | + |
| 276 | + |
| 277 | +class TestGetToday: |
| 278 | + """Test get_today functionality.""" |
| 279 | + |
| 280 | + def test_get_today_empty(self, tmp_path): |
| 281 | + """Test get_today with no data for today.""" |
| 282 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 283 | + today = tracker.get_today() |
| 284 | + |
| 285 | + assert today["requests"] == 0 |
| 286 | + assert today["actual_cost"] == 0 |
| 287 | + |
| 288 | + def test_get_today_with_data(self, tmp_path): |
| 289 | + """Test get_today returns today's data.""" |
| 290 | + tracker = CostTracker(storage_dir=str(tmp_path / ".empathy")) |
| 291 | + tracker.log_request("claude-3-haiku-20240307", 1000, 500, "test") |
| 292 | + |
| 293 | + today = tracker.get_today() |
| 294 | + assert today["requests"] == 1 |
| 295 | + assert today["input_tokens"] == 1000 |
| 296 | + |
| 297 | + |
| 298 | +class TestGlobalTracker: |
| 299 | + """Test global tracker functionality.""" |
| 300 | + |
| 301 | + def test_get_tracker_singleton(self, tmp_path): |
| 302 | + """Test that get_tracker returns same instance.""" |
| 303 | + import empathy_os.cost_tracker as ct |
| 304 | + |
| 305 | + # Reset singleton |
| 306 | + ct._tracker = None |
| 307 | + |
| 308 | + tracker1 = get_tracker(storage_dir=str(tmp_path / ".empathy")) |
| 309 | + tracker2 = get_tracker(storage_dir=str(tmp_path / ".empathy")) |
| 310 | + |
| 311 | + assert tracker1 is tracker2 |
| 312 | + |
| 313 | + # Cleanup |
| 314 | + ct._tracker = None |
| 315 | + |
| 316 | + def test_log_request_convenience_function(self, tmp_path): |
| 317 | + """Test log_request convenience function.""" |
| 318 | + import empathy_os.cost_tracker as ct |
| 319 | + |
| 320 | + ct._tracker = None |
| 321 | + |
| 322 | + with patch.object(ct, "_tracker", None): |
| 323 | + with patch.object(ct, "get_tracker") as mock_get: |
| 324 | + mock_tracker = MagicMock() |
| 325 | + mock_get.return_value = mock_tracker |
| 326 | + |
| 327 | + log_request("model", 100, 50, "task") |
| 328 | + mock_tracker.log_request.assert_called_once() |
| 329 | + |
| 330 | + ct._tracker = None |
| 331 | + |
| 332 | + |
| 333 | +class TestCLICommand: |
| 334 | + """Test CLI command handler.""" |
| 335 | + |
| 336 | + def test_cmd_costs_text_output(self, tmp_path, capsys): |
| 337 | + """Test cmd_costs with text output.""" |
| 338 | + args = MagicMock() |
| 339 | + args.empathy_dir = str(tmp_path / ".empathy") |
| 340 | + args.days = 7 |
| 341 | + args.json = False |
| 342 | + |
| 343 | + result = cmd_costs(args) |
| 344 | + |
| 345 | + assert result == 0 |
| 346 | + captured = capsys.readouterr() |
| 347 | + assert "COST TRACKING REPORT" in captured.out |
| 348 | + |
| 349 | + def test_cmd_costs_json_output(self, tmp_path, capsys): |
| 350 | + """Test cmd_costs with JSON output.""" |
| 351 | + args = MagicMock() |
| 352 | + args.empathy_dir = str(tmp_path / ".empathy") |
| 353 | + args.days = 7 |
| 354 | + args.json = True |
| 355 | + |
| 356 | + result = cmd_costs(args) |
| 357 | + |
| 358 | + assert result == 0 |
| 359 | + captured = capsys.readouterr() |
| 360 | + data = json.loads(captured.out) |
| 361 | + assert "requests" in data |
| 362 | + assert "savings_percent" in data |
| 363 | + |
| 364 | + |
| 365 | +class TestModelPricing: |
| 366 | + """Test model pricing configuration.""" |
| 367 | + |
| 368 | + def test_model_pricing_has_tiers(self): |
| 369 | + """Test that MODEL_PRICING has tier aliases.""" |
| 370 | + assert "cheap" in MODEL_PRICING |
| 371 | + assert "capable" in MODEL_PRICING |
| 372 | + assert "premium" in MODEL_PRICING |
| 373 | + |
| 374 | + def test_model_pricing_has_legacy_models(self): |
| 375 | + """Test that MODEL_PRICING includes legacy models.""" |
| 376 | + assert "claude-3-haiku-20240307" in MODEL_PRICING |
| 377 | + assert "gpt-4-turbo" in MODEL_PRICING |
| 378 | + |
| 379 | + def test_baseline_model_defined(self): |
| 380 | + """Test that BASELINE_MODEL is defined and valid.""" |
| 381 | + assert BASELINE_MODEL is not None |
| 382 | + assert "opus" in BASELINE_MODEL.lower() |
| 383 | + |
| 384 | + def test_build_model_pricing_returns_dict(self): |
| 385 | + """Test _build_model_pricing returns valid dict.""" |
| 386 | + pricing = _build_model_pricing() |
| 387 | + assert isinstance(pricing, dict) |
| 388 | + assert len(pricing) > 0 |
| 389 | + |
| 390 | + # Each entry should have input and output costs |
| 391 | + for model, costs in pricing.items(): |
| 392 | + assert "input" in costs |
| 393 | + assert "output" in costs |
| 394 | + assert costs["input"] >= 0 |
| 395 | + assert costs["output"] >= 0 |
0 commit comments