""" 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()