337 lines
13 KiB
Python
337 lines
13 KiB
Python
"""
|
|
F4 Statistics Dashboard - Backend Unit Tests
|
|
Tests for StatisticsCalculator class in statistics.py
|
|
"""
|
|
import pytest
|
|
from datetime import datetime, timedelta, timezone
|
|
from statistics import StatisticsCalculator
|
|
|
|
|
|
class TestStatisticsCalculator:
|
|
"""Tests for StatisticsCalculator.calculate() method"""
|
|
|
|
# ========================================
|
|
# Test: Empty History
|
|
# ========================================
|
|
|
|
def test_empty_history_returns_zero_stats(self):
|
|
"""Empty history should return all zeros"""
|
|
result = StatisticsCalculator.calculate([], 30)
|
|
|
|
assert result["total_errors"] == 0
|
|
assert result["pattern_frequency"] == {}
|
|
assert result["category_breakdown"] == {}
|
|
assert result["top_patterns"] == []
|
|
assert result["resolution_rate"]["resolved"] == 0
|
|
assert result["resolution_rate"]["unresolved"] == 0
|
|
assert result["resolution_rate"]["ignored"] == 0
|
|
assert result["trend"]["last_24h"] == 0
|
|
assert result["trend"]["last_7d"] == 0
|
|
assert result["trend"]["last_30d"] == 0
|
|
|
|
# ========================================
|
|
# Test: Pattern Frequency
|
|
# ========================================
|
|
|
|
def test_pattern_frequency_counts_correctly(self):
|
|
"""Pattern frequency should count occurrences correctly"""
|
|
history = [
|
|
self._make_entry("cuda_oom_classic", "memory"),
|
|
self._make_entry("cuda_oom_classic", "memory"),
|
|
self._make_entry("missing_module", "framework"),
|
|
self._make_entry("cuda_oom_classic", "memory"),
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert result["pattern_frequency"]["cuda_oom_classic"] == 3
|
|
assert result["pattern_frequency"]["missing_module"] == 1
|
|
assert result["total_errors"] == 4
|
|
|
|
def test_repeat_count_is_weighted_in_totals(self):
|
|
"""repeat_count should contribute to totals and counts."""
|
|
now = datetime.now(timezone.utc)
|
|
history = [
|
|
{
|
|
"timestamp": (now - timedelta(minutes=5)).isoformat(),
|
|
"matched_pattern_id": "a",
|
|
"pattern_category": "memory",
|
|
"resolution_status": "unresolved",
|
|
"repeat_count": 3,
|
|
"first_seen": (now - timedelta(minutes=5)).isoformat(),
|
|
"last_seen": (now - timedelta(minutes=4)).isoformat(),
|
|
},
|
|
{
|
|
"timestamp": (now - timedelta(minutes=3)).isoformat(),
|
|
"matched_pattern_id": "b",
|
|
"pattern_category": "workflow",
|
|
"resolution_status": "resolved",
|
|
"repeat_count": 2,
|
|
"first_seen": (now - timedelta(minutes=3)).isoformat(),
|
|
"last_seen": (now - timedelta(minutes=3)).isoformat(),
|
|
},
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
assert result["total_errors"] == 5
|
|
assert result["pattern_frequency"]["a"] == 3
|
|
assert result["pattern_frequency"]["b"] == 2
|
|
assert result["category_breakdown"]["memory"] == 3
|
|
assert result["category_breakdown"]["workflow"] == 2
|
|
assert result["resolution_rate"]["unresolved"] == 3
|
|
assert result["resolution_rate"]["resolved"] == 2
|
|
|
|
def test_pattern_frequency_skips_none_pattern(self):
|
|
"""Entries without pattern_id should be skipped in frequency count"""
|
|
history = [
|
|
self._make_entry(None, None),
|
|
self._make_entry("cuda_oom", "memory"),
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
# None pattern is skipped, only cuda_oom is counted
|
|
assert "unknown" not in result["pattern_frequency"]
|
|
assert result["pattern_frequency"]["cuda_oom"] == 1
|
|
# But total_errors still includes all entries in time range
|
|
assert result["total_errors"] == 2
|
|
|
|
# ========================================
|
|
# Test: Category Breakdown
|
|
# ========================================
|
|
|
|
def test_category_breakdown_counts_correctly(self):
|
|
"""Category breakdown should group by category"""
|
|
history = [
|
|
self._make_entry("cuda_oom", "memory"),
|
|
self._make_entry("vram_full", "memory"),
|
|
self._make_entry("model_not_found", "model_loading"),
|
|
self._make_entry("connection_error", "workflow"),
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert result["category_breakdown"]["memory"] == 2
|
|
assert result["category_breakdown"]["model_loading"] == 1
|
|
assert result["category_breakdown"]["workflow"] == 1
|
|
|
|
def test_category_breakdown_skips_none_category(self):
|
|
"""Entries without category should be skipped in breakdown"""
|
|
history = [
|
|
self._make_entry("some_error", None),
|
|
self._make_entry("other_error", "memory"),
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
# None category is skipped, not counted as generic
|
|
assert "generic" not in result["category_breakdown"]
|
|
assert result["category_breakdown"]["memory"] == 1
|
|
|
|
# ========================================
|
|
# Test: Top Patterns
|
|
# ========================================
|
|
|
|
def test_top_patterns_returns_top_5(self):
|
|
"""Should return top 5 most frequent patterns"""
|
|
history = [
|
|
*[self._make_entry("pattern_a", "memory")] * 10,
|
|
*[self._make_entry("pattern_b", "workflow")] * 8,
|
|
*[self._make_entry("pattern_c", "model_loading")] * 6,
|
|
*[self._make_entry("pattern_d", "framework")] * 4,
|
|
*[self._make_entry("pattern_e", "generic")] * 2,
|
|
*[self._make_entry("pattern_f", "memory")] * 1, # Should NOT be in top 5
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert len(result["top_patterns"]) == 5
|
|
assert result["top_patterns"][0]["pattern_id"] == "pattern_a"
|
|
assert result["top_patterns"][0]["count"] == 10
|
|
assert result["top_patterns"][1]["pattern_id"] == "pattern_b"
|
|
assert result["top_patterns"][4]["pattern_id"] == "pattern_e"
|
|
|
|
def test_top_patterns_sorted_by_count_descending(self):
|
|
"""Top patterns should be sorted by count descending"""
|
|
history = [
|
|
self._make_entry("low", "memory"),
|
|
*[self._make_entry("high", "memory")] * 5,
|
|
*[self._make_entry("medium", "memory")] * 3,
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
counts = [p["count"] for p in result["top_patterns"]]
|
|
assert counts == sorted(counts, reverse=True)
|
|
|
|
# ========================================
|
|
# Test: Resolution Rate
|
|
# ========================================
|
|
|
|
def test_resolution_rate_counts_all_statuses(self):
|
|
"""Resolution rate should count all three statuses"""
|
|
history = [
|
|
self._make_entry("a", "memory", resolution_status="resolved"),
|
|
self._make_entry("b", "memory", resolution_status="resolved"),
|
|
self._make_entry("c", "memory", resolution_status="unresolved"),
|
|
self._make_entry("d", "memory", resolution_status="ignored"),
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert result["resolution_rate"]["resolved"] == 2
|
|
assert result["resolution_rate"]["unresolved"] == 1
|
|
assert result["resolution_rate"]["ignored"] == 1
|
|
|
|
def test_resolution_rate_default_is_unresolved(self):
|
|
"""Missing resolution_status should default to unresolved"""
|
|
history = [
|
|
{"timestamp": self._recent_timestamp(), "matched_pattern_id": "a"},
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert result["resolution_rate"]["unresolved"] == 1
|
|
|
|
# ========================================
|
|
# Test: Trend Calculation
|
|
# ========================================
|
|
|
|
def test_trend_last_24h_counts_recent_errors(self):
|
|
"""Trend should count errors in last 24 hours"""
|
|
now = datetime.now(timezone.utc)
|
|
history = [
|
|
self._make_entry("a", "memory", timestamp=now - timedelta(hours=1)),
|
|
self._make_entry("b", "memory", timestamp=now - timedelta(hours=12)),
|
|
self._make_entry("c", "memory", timestamp=now - timedelta(hours=25)), # Outside 24h
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert result["trend"]["last_24h"] == 2
|
|
|
|
def test_trend_last_7d_counts_weekly_errors(self):
|
|
"""Trend should count errors in last 7 days"""
|
|
now = datetime.now(timezone.utc)
|
|
history = [
|
|
self._make_entry("a", "memory", timestamp=now - timedelta(days=1)),
|
|
self._make_entry("b", "memory", timestamp=now - timedelta(days=5)),
|
|
self._make_entry("c", "memory", timestamp=now - timedelta(days=8)), # Outside 7d
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert result["trend"]["last_7d"] == 2
|
|
|
|
def test_trend_last_30d_equals_total_within_range(self):
|
|
"""Trend 30d should equal total if all within 30 days"""
|
|
now = datetime.now(timezone.utc)
|
|
history = [
|
|
self._make_entry("a", "memory", timestamp=now - timedelta(days=1)),
|
|
self._make_entry("b", "memory", timestamp=now - timedelta(days=15)),
|
|
self._make_entry("c", "memory", timestamp=now - timedelta(days=29)),
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
assert result["trend"]["last_30d"] == 3
|
|
assert result["total_errors"] == 3
|
|
|
|
# ========================================
|
|
# Test: Time Range Filtering
|
|
# ========================================
|
|
|
|
def test_time_range_filters_old_entries(self):
|
|
"""Entries older than time_range_days should be excluded"""
|
|
now = datetime.now(timezone.utc)
|
|
history = [
|
|
self._make_entry("a", "memory", timestamp=now - timedelta(days=5)),
|
|
self._make_entry("b", "memory", timestamp=now - timedelta(days=15)), # Outside 7d
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 7)
|
|
|
|
assert result["total_errors"] == 1
|
|
assert result["pattern_frequency"]["a"] == 1
|
|
assert "b" not in result["pattern_frequency"]
|
|
|
|
def test_custom_time_range(self):
|
|
"""Custom time range should work correctly"""
|
|
now = datetime.now(timezone.utc)
|
|
history = [
|
|
self._make_entry("a", "memory", timestamp=now - timedelta(days=2)),
|
|
self._make_entry("b", "memory", timestamp=now - timedelta(days=4)), # Outside 3d
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 3)
|
|
|
|
assert result["total_errors"] == 1
|
|
|
|
# ========================================
|
|
# Test: Edge Cases
|
|
# ========================================
|
|
|
|
def test_handles_malformed_timestamp(self):
|
|
"""Should handle malformed timestamps gracefully"""
|
|
history = [
|
|
{"timestamp": "invalid-date", "matched_pattern_id": "a"},
|
|
self._make_entry("b", "memory"),
|
|
]
|
|
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
|
|
# Should still process valid entries
|
|
assert result["total_errors"] >= 1
|
|
|
|
def test_handles_dict_with_missing_fields(self):
|
|
"""Should handle dicts with missing fields"""
|
|
history = [
|
|
{"timestamp": self._recent_timestamp()}, # No pattern_id
|
|
{"matched_pattern_id": "test"}, # No timestamp
|
|
]
|
|
|
|
# Should not raise exception
|
|
result = StatisticsCalculator.calculate(history, 30)
|
|
assert isinstance(result, dict)
|
|
|
|
# ========================================
|
|
# Helper Methods
|
|
# ========================================
|
|
|
|
def _make_entry(
|
|
self,
|
|
pattern_id: str,
|
|
category: str,
|
|
timestamp: datetime = None,
|
|
resolution_status: str = "unresolved"
|
|
) -> dict:
|
|
"""Create a test history entry"""
|
|
if timestamp is None:
|
|
timestamp = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
|
|
return {
|
|
"timestamp": timestamp.isoformat(),
|
|
"matched_pattern_id": pattern_id,
|
|
"pattern_category": category,
|
|
"resolution_status": resolution_status,
|
|
}
|
|
|
|
def _recent_timestamp(self) -> str:
|
|
"""Return a recent timestamp string"""
|
|
return (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
|
|
|
|
|
|
class TestStatisticsCalculatorHelpers:
|
|
"""Tests for internal helper methods"""
|
|
|
|
def test_empty_stats_structure(self):
|
|
"""_empty_stats should return correct structure"""
|
|
result = StatisticsCalculator._empty_stats()
|
|
|
|
assert "total_errors" in result
|
|
assert "pattern_frequency" in result
|
|
assert "category_breakdown" in result
|
|
assert "top_patterns" in result
|
|
assert "resolution_rate" in result
|
|
assert "trend" in result
|