283 lines
12 KiB
Python
283 lines
12 KiB
Python
"""
|
|
Unit tests for FeatureEnrichmentService
|
|
========================================
|
|
Tests all 6 enrichment methods with mocked DB cursor:
|
|
1. compute_team_stats
|
|
2. compute_h2h
|
|
3. compute_form_streaks
|
|
4. compute_referee_stats
|
|
5. compute_league_averages
|
|
6. compute_momentum
|
|
"""
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
AI_ENGINE_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(AI_ENGINE_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(AI_ENGINE_ROOT))
|
|
|
|
from services.feature_enrichment import FeatureEnrichmentService, _safe_avg
|
|
|
|
|
|
def _make_cursor(rows=None, side_effect=None):
|
|
"""Create a mock RealDictCursor."""
|
|
cur = MagicMock()
|
|
if side_effect:
|
|
cur.execute.side_effect = side_effect
|
|
else:
|
|
cur.fetchall.return_value = rows or []
|
|
cur.fetchone.return_value = rows[0] if rows else None
|
|
return cur
|
|
|
|
|
|
class TestSafeAvg(unittest.TestCase):
|
|
def test_returns_average(self):
|
|
self.assertAlmostEqual(_safe_avg([2.0, 4.0, 6.0], 0.0), 4.0)
|
|
|
|
def test_returns_default_on_empty(self):
|
|
self.assertEqual(_safe_avg([], 99.0), 99.0)
|
|
|
|
def test_single_value(self):
|
|
self.assertAlmostEqual(_safe_avg([7.5], 0.0), 7.5)
|
|
|
|
|
|
class TestComputeTeamStats(unittest.TestCase):
|
|
def setUp(self):
|
|
self.svc = FeatureEnrichmentService()
|
|
self.ts = 1700000000000
|
|
|
|
def test_returns_defaults_when_no_team_id(self):
|
|
result = self.svc.compute_team_stats(MagicMock(), '', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_TEAM_STATS)
|
|
|
|
def test_returns_defaults_when_no_rows(self):
|
|
cur = _make_cursor(rows=[])
|
|
result = self.svc.compute_team_stats(cur, 'team1', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_TEAM_STATS)
|
|
|
|
def test_returns_defaults_on_db_error(self):
|
|
cur = _make_cursor(side_effect=Exception('DB down'))
|
|
result = self.svc.compute_team_stats(cur, 'team1', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_TEAM_STATS)
|
|
|
|
def test_calculates_averages_correctly(self):
|
|
rows = [
|
|
{'possession_percentage': 60.0, 'shots_on_target': 5, 'total_shots': 10, 'corners': 7},
|
|
{'possession_percentage': 40.0, 'shots_on_target': 3, 'total_shots': 12, 'corners': 3},
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_team_stats(cur, 'team1', self.ts)
|
|
|
|
self.assertAlmostEqual(result['avg_possession'], 50.0)
|
|
self.assertAlmostEqual(result['avg_shots_on_target'], 4.0)
|
|
self.assertAlmostEqual(result['shot_conversion'], (5 / 10 + 3 / 12) / 2, places=4)
|
|
self.assertAlmostEqual(result['avg_corners'], 5.0)
|
|
|
|
def test_handles_none_subfields_gracefully(self):
|
|
"""Rows with None values should be skipped, not crash."""
|
|
rows = [
|
|
{'possession_percentage': 55.0, 'shots_on_target': None, 'total_shots': None, 'corners': 4},
|
|
{'possession_percentage': None, 'shots_on_target': 2, 'total_shots': 8, 'corners': None},
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_team_stats(cur, 'team1', self.ts)
|
|
|
|
self.assertAlmostEqual(result['avg_possession'], 55.0)
|
|
self.assertAlmostEqual(result['avg_shots_on_target'], 2.0)
|
|
self.assertAlmostEqual(result['avg_corners'], 4.0)
|
|
|
|
|
|
class TestComputeH2H(unittest.TestCase):
|
|
def setUp(self):
|
|
self.svc = FeatureEnrichmentService()
|
|
self.ts = 1700000000000
|
|
|
|
def test_returns_defaults_when_no_ids(self):
|
|
result = self.svc.compute_h2h(MagicMock(), '', 'away1', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_H2H)
|
|
|
|
def test_returns_defaults_when_no_rows(self):
|
|
cur = _make_cursor(rows=[])
|
|
result = self.svc.compute_h2h(cur, 'home1', 'away1', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_H2H)
|
|
|
|
def test_calculates_h2h_stats(self):
|
|
rows = [
|
|
{'home_team_id': 'home1', 'away_team_id': 'away1', 'score_home': 2, 'score_away': 1}, # home win, btts, over25
|
|
{'home_team_id': 'home1', 'away_team_id': 'away1', 'score_home': 0, 'score_away': 0}, # draw, no btts, no over25
|
|
{'home_team_id': 'away1', 'away_team_id': 'home1', 'score_home': 1, 'score_away': 3}, # reversed: home wins again, btts, over25
|
|
{'home_team_id': 'away1', 'away_team_id': 'home1', 'score_home': 2, 'score_away': 0}, # reversed: away(=home1) lost
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_h2h(cur, 'home1', 'away1', self.ts)
|
|
|
|
self.assertEqual(result['total_matches'], 4)
|
|
self.assertAlmostEqual(result['home_win_rate'], 2 / 4)
|
|
self.assertAlmostEqual(result['draw_rate'], 1 / 4)
|
|
self.assertAlmostEqual(result['btts_rate'], 2 / 4)
|
|
self.assertAlmostEqual(result['over25_rate'], 2 / 4)
|
|
|
|
def test_returns_defaults_on_db_error(self):
|
|
cur = _make_cursor(side_effect=Exception('connection lost'))
|
|
result = self.svc.compute_h2h(cur, 'home1', 'away1', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_H2H)
|
|
|
|
|
|
class TestComputeFormStreaks(unittest.TestCase):
|
|
def setUp(self):
|
|
self.svc = FeatureEnrichmentService()
|
|
self.ts = 1700000000000
|
|
|
|
def test_returns_defaults_when_no_team_id(self):
|
|
result = self.svc.compute_form_streaks(MagicMock(), '', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_FORM)
|
|
|
|
def test_calculates_streaks_correctly(self):
|
|
"""Most recent first: W, W, D, L → winning_streak=2, unbeaten_streak=3."""
|
|
rows = [
|
|
{'home_team_id': 'team1', 'away_team_id': 'x', 'score_home': 2, 'score_away': 0}, # W (clean sheet, scored)
|
|
{'home_team_id': 'team1', 'away_team_id': 'x', 'score_home': 1, 'score_away': 0}, # W (clean sheet, scored)
|
|
{'home_team_id': 'x', 'away_team_id': 'team1', 'score_home': 1, 'score_away': 1}, # D (scored, conceded)
|
|
{'home_team_id': 'team1', 'away_team_id': 'x', 'score_home': 0, 'score_away': 2}, # L (not scored, conceded)
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_form_streaks(cur, 'team1', self.ts)
|
|
|
|
self.assertEqual(result['winning_streak'], 2)
|
|
self.assertEqual(result['unbeaten_streak'], 3)
|
|
self.assertAlmostEqual(result['clean_sheet_rate'], 2 / 4)
|
|
self.assertAlmostEqual(result['scoring_rate'], 3 / 4)
|
|
|
|
def test_all_losses(self):
|
|
rows = [
|
|
{'home_team_id': 'team1', 'away_team_id': 'x', 'score_home': 0, 'score_away': 1},
|
|
{'home_team_id': 'team1', 'away_team_id': 'x', 'score_home': 0, 'score_away': 3},
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_form_streaks(cur, 'team1', self.ts)
|
|
|
|
self.assertEqual(result['winning_streak'], 0)
|
|
self.assertEqual(result['unbeaten_streak'], 0)
|
|
self.assertAlmostEqual(result['scoring_rate'], 0.0)
|
|
|
|
|
|
class TestComputeRefereeStats(unittest.TestCase):
|
|
def setUp(self):
|
|
self.svc = FeatureEnrichmentService()
|
|
self.ts = 1700000000000
|
|
|
|
def test_returns_defaults_when_no_name(self):
|
|
result = self.svc.compute_referee_stats(MagicMock(), None, self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_REFEREE)
|
|
|
|
def test_calculates_referee_tendencies(self):
|
|
match_rows = [
|
|
{'home_team_id': 'h1', 'score_home': 2, 'score_away': 0, 'match_id': 'm1'}, # home win
|
|
{'home_team_id': 'h2', 'score_home': 1, 'score_away': 1, 'match_id': 'm2'}, # draw
|
|
]
|
|
card_row = {'yellows': 6, 'total_cards': 8}
|
|
|
|
cur = MagicMock()
|
|
# First execute (match query) → match_rows
|
|
# Second execute (card query) → card_row
|
|
cur.fetchall.return_value = match_rows
|
|
cur.fetchone.return_value = card_row
|
|
|
|
result = self.svc.compute_referee_stats(cur, 'Ref Name', self.ts)
|
|
|
|
self.assertEqual(result['experience'], 2)
|
|
self.assertAlmostEqual(result['avg_goals'], (2 + 0 + 1 + 1) / 2)
|
|
# home_bias = (1/2) - 0.46 = 0.04
|
|
self.assertAlmostEqual(result['home_bias'], 0.04, places=4)
|
|
self.assertAlmostEqual(result['avg_yellow'], 6 / 2)
|
|
self.assertAlmostEqual(result['cards_total'], 8 / 2)
|
|
|
|
def test_returns_defaults_on_db_error(self):
|
|
cur = _make_cursor(side_effect=Exception('timeout'))
|
|
result = self.svc.compute_referee_stats(cur, 'Some Ref', self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_REFEREE)
|
|
|
|
|
|
class TestComputeLeagueAverages(unittest.TestCase):
|
|
def setUp(self):
|
|
self.svc = FeatureEnrichmentService()
|
|
self.ts = 1700000000000
|
|
|
|
def test_returns_defaults_when_no_league_id(self):
|
|
result = self.svc.compute_league_averages(MagicMock(), None, self.ts)
|
|
self.assertEqual(result, FeatureEnrichmentService._DEFAULT_LEAGUE)
|
|
|
|
def test_calculates_league_averages(self):
|
|
rows = [
|
|
{'score_home': 1, 'score_away': 1}, # 2 goals
|
|
{'score_home': 0, 'score_away': 0}, # 0 goals (zero-goal match)
|
|
{'score_home': 3, 'score_away': 2}, # 5 goals
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_league_averages(cur, 'league1', self.ts)
|
|
|
|
self.assertAlmostEqual(result['avg_goals'], 7 / 3, places=4)
|
|
self.assertAlmostEqual(result['zero_goal_rate'], 1 / 3, places=4)
|
|
|
|
|
|
class TestComputeMomentum(unittest.TestCase):
|
|
def setUp(self):
|
|
self.svc = FeatureEnrichmentService()
|
|
self.ts = 1700000000000
|
|
|
|
def test_returns_zero_when_no_team_id(self):
|
|
result = self.svc.compute_momentum(MagicMock(), '', self.ts)
|
|
self.assertEqual(result, 0.0)
|
|
|
|
def test_returns_zero_when_no_rows(self):
|
|
cur = _make_cursor(rows=[])
|
|
result = self.svc.compute_momentum(cur, 'team1', self.ts)
|
|
self.assertEqual(result, 0.0)
|
|
|
|
def test_all_wins_returns_one(self):
|
|
"""All wins → momentum = 1.0 (max possible)."""
|
|
rows = [
|
|
{'home_team_id': 'team1', 'score_home': 3, 'score_away': 0},
|
|
{'home_team_id': 'team1', 'score_home': 2, 'score_away': 1},
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_momentum(cur, 'team1', self.ts)
|
|
self.assertAlmostEqual(result, 1.0, places=4)
|
|
|
|
def test_all_losses_returns_negative(self):
|
|
"""All losses → negative momentum."""
|
|
rows = [
|
|
{'home_team_id': 'team1', 'score_home': 0, 'score_away': 2},
|
|
{'home_team_id': 'team1', 'score_home': 1, 'score_away': 3},
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_momentum(cur, 'team1', self.ts)
|
|
self.assertLess(result, 0.0)
|
|
|
|
def test_mixed_results(self):
|
|
"""W, D, L → weighted score between -1 and 1."""
|
|
rows = [
|
|
{'home_team_id': 'team1', 'score_home': 1, 'score_away': 0}, # W (weight=3)
|
|
{'home_team_id': 'x', 'away_team_id': 'team1', 'score_home': 0, 'score_away': 0}, # D (weight=2)
|
|
{'home_team_id': 'team1', 'score_home': 0, 'score_away': 1}, # L (weight=1)
|
|
]
|
|
cur = _make_cursor(rows)
|
|
result = self.svc.compute_momentum(cur, 'team1', self.ts)
|
|
|
|
# weighted = 3*3 + 1*2 + (-1)*1 = 9+2-1 = 10
|
|
# max_possible = 3*3 + 3*2 + 3*1 = 18
|
|
# normalised = 10/18 ≈ 0.5556
|
|
self.assertAlmostEqual(result, round(10 / 18, 4), places=4)
|
|
|
|
def test_returns_zero_on_db_error(self):
|
|
cur = _make_cursor(side_effect=Exception('broken pipe'))
|
|
result = self.svc.compute_momentum(cur, 'team1', self.ts)
|
|
self.assertEqual(result, 0.0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|