#!/usr/bin/env python3 """ Comprehensive API Data Integrity Test Suite Compares data between localhost PostgreSQL API and production SQLite API for all routers except battingstats, custom_commands, fieldingstats, pitchingstats. Usage: python comprehensive_api_integrity_tests.py python comprehensive_api_integrity_tests.py --verbose python comprehensive_api_integrity_tests.py --router teams """ import requests import json import sys import argparse from typing import Dict, List, Any, Tuple, Optional from dataclasses import dataclass from datetime import datetime import logging # API Configuration LOCALHOST_API = "https://api.sba.manticorum.com/v3" PRODUCTION_API = "https://sba.manticorum.com/api/v3" # Test Configuration TEST_SEASON = 10 SAMPLE_PLAYER_IDS = [9916, 9958, 9525, 9349, 9892] SAMPLE_TEAM_IDS = [404, 428, 443, 422, 425] SAMPLE_GAME_IDS = [1571, 1458, 1710] SAMPLE_MANAGER_IDS = [1, 2, 3, 4, 5] @dataclass class TestResult: """Container for test results""" test_name: str passed: bool localhost_data: Any production_data: Any error_message: str = "" details: Dict[str, Any] = None class ComprehensiveAPITester: """Comprehensive test suite for all API routers""" def __init__(self, verbose: bool = False): self.verbose = verbose self.results: List[TestResult] = [] self.setup_logging() def setup_logging(self): """Setup logging configuration""" level = logging.DEBUG if self.verbose else logging.INFO log_filename = f'logs/comprehensive_api_test_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log' logging.basicConfig( level=level, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_filename), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) def make_request(self, base_url: str, endpoint: str, params: Dict = None) -> Tuple[bool, Any]: """Make API request and return success status and data""" try: url = f"{base_url}{endpoint}" response = requests.get(url, params=params, timeout=30) response.raise_for_status() return True, response.json() except requests.exceptions.RequestException as e: self.logger.debug(f"Request failed for {url}: {e}") return False, str(e) def compare_basic_data(self, endpoint: str, params: Dict = None, fields_to_compare: List[str] = None) -> TestResult: """Generic comparison for basic endpoint data""" test_name = f"{endpoint}: {params or 'no params'}" if fields_to_compare is None: fields_to_compare = ['count'] localhost_success, localhost_data = self.make_request(LOCALHOST_API, endpoint, params) production_success, production_data = self.make_request(PRODUCTION_API, endpoint, params) if not localhost_success or not production_success: return TestResult( test_name=test_name, passed=False, localhost_data=localhost_data if localhost_success else None, production_data=production_data if production_success else None, error_message="API request failed" ) # Compare specified fields differences = {} for field in fields_to_compare: localhost_val = localhost_data.get(field) production_val = production_data.get(field) if localhost_val != production_val: differences[field] = { 'localhost': localhost_val, 'production': production_val } passed = len(differences) == 0 error_msg = f"Field differences: {differences}" if differences else "" return TestResult( test_name=test_name, passed=passed, localhost_data=localhost_data, production_data=production_data, error_message=error_msg, details={'differences': differences} ) def compare_list_data(self, endpoint: str, params: Dict = None, compare_top_n: int = 3) -> TestResult: """Compare list-based endpoints (teams, players, etc.)""" test_name = f"{endpoint}: {params or 'no params'}" localhost_success, localhost_data = self.make_request(LOCALHOST_API, endpoint, params) production_success, production_data = self.make_request(PRODUCTION_API, endpoint, params) if not localhost_success or not production_success: return TestResult( test_name=test_name, passed=False, localhost_data=localhost_data if localhost_success else None, production_data=production_data if production_success else None, error_message="API request failed" ) differences = {} # Compare counts localhost_count = localhost_data.get('count', len(localhost_data) if isinstance(localhost_data, list) else 0) production_count = production_data.get('count', len(production_data) if isinstance(production_data, list) else 0) if localhost_count != production_count: differences['count'] = { 'localhost': localhost_count, 'production': production_count } # Get list data (handle both formats: {'results': []} and direct list) localhost_list = localhost_data.get('results', localhost_data) if isinstance(localhost_data, dict) else localhost_data production_list = production_data.get('results', production_data) if isinstance(production_data, dict) else production_data if isinstance(localhost_list, list) and isinstance(production_list, list): # Compare top N items top_n = min(compare_top_n, len(localhost_list), len(production_list)) if top_n > 0: for i in range(top_n): local_item = localhost_list[i] prod_item = production_list[i] # Compare key identifying fields local_id = local_item.get('id') prod_id = prod_item.get('id') if local_id != prod_id: differences[f'top_{i+1}_id'] = { 'localhost': local_id, 'production': prod_id } passed = len(differences) == 0 error_msg = f"Data differences: {differences}" if differences else "" return TestResult( test_name=test_name, passed=passed, localhost_data=localhost_data, production_data=production_data, error_message=error_msg, details={'differences': differences} ) # =============================== # ROUTER-SPECIFIC TEST METHODS # =============================== def test_awards_router(self) -> List[TestResult]: """Test awards router endpoints""" self.logger.info("Testing awards router...") results = [] test_cases = [ ("/awards", {"season": TEST_SEASON}), ("/awards", {"season": TEST_SEASON, "limit": 10}), ("/awards", {"team_id": SAMPLE_TEAM_IDS[0], "season": TEST_SEASON}), ] for endpoint, params in test_cases: result = self.compare_list_data(endpoint, params) results.append(result) self.logger.info(f"Awards {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_current_router(self) -> List[TestResult]: """Test current router endpoints""" self.logger.info("Testing current router...") results = [] test_cases = [ ("/current", {}), ("/current", {"league": "SBa"}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['season', 'week']) results.append(result) self.logger.info(f"Current {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_decisions_router(self) -> List[TestResult]: """Test decisions router endpoints""" self.logger.info("Testing decisions router...") results = [] test_cases = [ ("/decisions", {"season": TEST_SEASON, "limit": 10}), ("/decisions", {"season": TEST_SEASON, "player_id": SAMPLE_PLAYER_IDS[0]}), ("/decisions", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Decisions {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_divisions_router(self) -> List[TestResult]: """Test divisions router endpoints""" self.logger.info("Testing divisions router...") results = [] test_cases = [ ("/divisions", {"season": TEST_SEASON}), ("/divisions", {"season": TEST_SEASON, "league": "SBa"}), ] for endpoint, params in test_cases: result = self.compare_list_data(endpoint, params) results.append(result) self.logger.info(f"Divisions {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_draftdata_router(self) -> List[TestResult]: """Test draftdata router endpoints""" self.logger.info("Testing draftdata router...") results = [] test_cases = [ ("/draftdata", {"season": TEST_SEASON}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['current_pick', 'current_round']) results.append(result) self.logger.info(f"Draft data {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_draftlist_router(self) -> List[TestResult]: """Test draftlist router endpoints - REQUIRES AUTHENTICATION""" self.logger.info("Testing draftlist router (authentication required)...") results = [] # Note: This endpoint requires authentication, which test suite doesn't provide # This is expected behavior and not a migration issue self.logger.info("Skipping draftlist tests - authentication required (expected)") return results def test_draftpicks_router(self) -> List[TestResult]: """Test draftpicks router endpoints""" self.logger.info("Testing draftpicks router...") results = [] test_cases = [ ("/draftpicks", {"season": TEST_SEASON, "limit": 10}), ("/draftpicks", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ("/draftpicks", {"season": TEST_SEASON, "round": 1}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Draft picks {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_injuries_router(self) -> List[TestResult]: """Test injuries router endpoints""" self.logger.info("Testing injuries router...") results = [] test_cases = [ ("/injuries", {"season": TEST_SEASON, "limit": 10}), ("/injuries", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ("/injuries", {"season": TEST_SEASON, "active": True}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Injuries {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_keepers_router(self) -> List[TestResult]: """Test keepers router endpoints""" self.logger.info("Testing keepers router...") results = [] test_cases = [ ("/keepers", {"season": TEST_SEASON, "limit": 10}), ("/keepers", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Keepers {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_managers_router(self) -> List[TestResult]: """Test managers router endpoints""" self.logger.info("Testing managers router...") results = [] test_cases = [ ("/managers", {}), ("/managers", {"limit": 10}), ] for endpoint, params in test_cases: result = self.compare_list_data(endpoint, params) results.append(result) self.logger.info(f"Managers {params}: {'PASS' if result.passed else 'FAIL'}") # Test individual manager if SAMPLE_MANAGER_IDS: for manager_id in SAMPLE_MANAGER_IDS[:2]: # Test first 2 result = self.compare_basic_data(f"/managers/{manager_id}", {}) results.append(result) self.logger.info(f"Manager {manager_id}: {'PASS' if result.passed else 'FAIL'}") return results def test_players_router(self) -> List[TestResult]: """Test players router endpoints""" self.logger.info("Testing players router...") results = [] test_cases = [ ("/players", {"season": TEST_SEASON, "limit": 10}), ("/players", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ("/players", {"season": TEST_SEASON, "pos": "OF", "limit": 5}), ("/players", {"season": TEST_SEASON, "active": True, "limit": 10}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Players {params}: {'PASS' if result.passed else 'FAIL'}") # Test individual players for player_id in SAMPLE_PLAYER_IDS[:3]: # Test first 3 result = self.compare_basic_data(f"/players/{player_id}", {}) results.append(result) self.logger.info(f"Player {player_id}: {'PASS' if result.passed else 'FAIL'}") return results def test_results_router(self) -> List[TestResult]: """Test results router endpoints""" self.logger.info("Testing results router...") results = [] test_cases = [ ("/results", {"season": TEST_SEASON, "limit": 10}), ("/results", {"season": TEST_SEASON, "week": 1}), ("/results", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Results {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_sbaplayers_router(self) -> List[TestResult]: """Test sbaplayers router endpoints""" self.logger.info("Testing sbaplayers router...") results = [] test_cases = [ ("/sbaplayers", {"limit": 10}), ("/sbaplayers", {"active": True}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"SBA players {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_schedules_router(self) -> List[TestResult]: """Test schedules router endpoints""" self.logger.info("Testing schedules router...") results = [] test_cases = [ ("/schedules", {"season": TEST_SEASON, "limit": 10}), ("/schedules", {"season": TEST_SEASON, "week": 1}), ("/schedules", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Schedules {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_standings_router(self) -> List[TestResult]: """Test standings router endpoints""" self.logger.info("Testing standings router...") results = [] test_cases = [ ("/standings", {"season": TEST_SEASON}), ("/standings", {"season": TEST_SEASON, "league": "SBa"}), ("/standings", {"season": TEST_SEASON, "division": "Milkshake"}), ] for endpoint, params in test_cases: result = self.compare_list_data(endpoint, params) results.append(result) self.logger.info(f"Standings {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_stratgame_router(self) -> List[TestResult]: """Test games router endpoints (stratgame was renamed to games)""" self.logger.info("Testing games router...") results = [] test_cases = [ ("/games", {"season": TEST_SEASON, "limit": 10}), ("/games", {"season": TEST_SEASON, "week": 1}), ("/games", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Games {params}: {'PASS' if result.passed else 'FAIL'}") # Test individual games for game_id in SAMPLE_GAME_IDS[:2]: # Test first 2 result = self.compare_basic_data(f"/games/{game_id}", {}) results.append(result) self.logger.info(f"Game {game_id}: {'PASS' if result.passed else 'FAIL'}") return results def test_teams_router(self) -> List[TestResult]: """Test teams router endpoints""" self.logger.info("Testing teams router...") results = [] test_cases = [ ("/teams", {"season": TEST_SEASON}), ("/teams", {"season": TEST_SEASON, "division": "Milkshake"}), ("/teams", {"season": TEST_SEASON, "league": "SBa"}), ] for endpoint, params in test_cases: result = self.compare_list_data(endpoint, params) results.append(result) self.logger.info(f"Teams {params}: {'PASS' if result.passed else 'FAIL'}") # Test individual teams for team_id in SAMPLE_TEAM_IDS[:3]: # Test first 3 result = self.compare_basic_data(f"/teams/{team_id}", {}) results.append(result) self.logger.info(f"Team {team_id}: {'PASS' if result.passed else 'FAIL'}") return results def test_transactions_router(self) -> List[TestResult]: """Test transactions router endpoints""" self.logger.info("Testing transactions router...") results = [] test_cases = [ ("/transactions", {"season": TEST_SEASON, "limit": 10}), ("/transactions", {"season": TEST_SEASON, "team_id": SAMPLE_TEAM_IDS[0]}), ("/transactions", {"season": TEST_SEASON, "trans_type": "trade"}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Transactions {params}: {'PASS' if result.passed else 'FAIL'}") return results def test_stratplay_router(self) -> List[TestResult]: """Test stratplay router endpoints (comprehensive)""" self.logger.info("Testing stratplay router...") results = [] # Basic plays endpoint test_cases = [ ("/plays", {"season": TEST_SEASON, "limit": 10}), ("/plays", {"season": TEST_SEASON, "game_id": SAMPLE_GAME_IDS[0]}), ("/plays", {"season": TEST_SEASON, "batter_id": SAMPLE_PLAYER_IDS[0], "limit": 5}), ] for endpoint, params in test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Plays {params}: {'PASS' if result.passed else 'FAIL'}") # Batting stats (already tested in PostgreSQL fixes) batting_test_cases = [ ("/plays/batting", {"season": TEST_SEASON, "group_by": "player", "limit": 5}), ("/plays/batting", {"season": TEST_SEASON, "group_by": "team", "limit": 5}), ("/plays/batting", {"season": TEST_SEASON, "group_by": "playerteam", "limit": 5}), ] for endpoint, params in batting_test_cases: result = self.compare_basic_data(endpoint, params, ['count']) results.append(result) self.logger.info(f"Batting stats {params}: {'PASS' if result.passed else 'FAIL'}") return results # =============================== # TEST RUNNER METHODS # =============================== def run_all_tests(self) -> None: """Run the complete test suite for all routers""" self.logger.info("Starting Comprehensive API Data Integrity Test Suite") self.logger.info(f"Localhost API: {LOCALHOST_API}") self.logger.info(f"Production API: {PRODUCTION_API}") self.logger.info(f"Test Season: {TEST_SEASON}") self.logger.info("=" * 60) # Run all router tests router_tests = [ self.test_awards_router, self.test_current_router, self.test_decisions_router, self.test_divisions_router, self.test_draftdata_router, self.test_draftlist_router, self.test_draftpicks_router, self.test_injuries_router, self.test_keepers_router, self.test_managers_router, self.test_players_router, self.test_results_router, self.test_sbaplayers_router, self.test_schedules_router, self.test_standings_router, self.test_stratgame_router, self.test_teams_router, self.test_transactions_router, self.test_stratplay_router, ] for test_func in router_tests: try: self.results.extend(test_func()) except Exception as e: self.logger.error(f"Error in {test_func.__name__}: {e}") # Generate summary self.generate_summary() def run_router_tests(self, router_name: str) -> None: """Run tests for a specific router""" router_map = { 'awards': self.test_awards_router, 'current': self.test_current_router, 'decisions': self.test_decisions_router, 'divisions': self.test_divisions_router, 'draftdata': self.test_draftdata_router, 'draftlist': self.test_draftlist_router, 'draftpicks': self.test_draftpicks_router, 'injuries': self.test_injuries_router, 'keepers': self.test_keepers_router, 'managers': self.test_managers_router, 'players': self.test_players_router, 'results': self.test_results_router, 'sbaplayers': self.test_sbaplayers_router, 'schedules': self.test_schedules_router, 'standings': self.test_standings_router, 'stratgame': self.test_stratgame_router, 'teams': self.test_teams_router, 'transactions': self.test_transactions_router, 'stratplay': self.test_stratplay_router, } if router_name not in router_map: self.logger.error(f"Unknown router: {router_name}") self.logger.info(f"Available routers: {', '.join(router_map.keys())}") return self.logger.info(f"Running tests for {router_name} router only") self.results.extend(router_map[router_name]()) self.generate_summary() def generate_summary(self) -> None: """Generate and display test summary""" total_tests = len(self.results) passed_tests = sum(1 for r in self.results if r.passed) failed_tests = total_tests - passed_tests success_rate = (passed_tests / total_tests * 100) if total_tests > 0 else 0 self.logger.info("=" * 60) self.logger.info("TEST SUMMARY") self.logger.info("=" * 60) self.logger.info(f"Total Tests: {total_tests}") self.logger.info(f"Passed: {passed_tests}") self.logger.info(f"Failed: {failed_tests}") self.logger.info(f"Success Rate: {success_rate:.1f}%") if failed_tests > 0: self.logger.info("\nFAILED TESTS:") self.logger.info("-" * 40) for result in self.results: if not result.passed: self.logger.info(f"❌ {result.test_name}") self.logger.info(f" Error: {result.error_message}") # Save detailed results timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") results_file = f"logs/comprehensive_api_results_{timestamp}.json" results_data = { 'timestamp': timestamp, 'total_tests': total_tests, 'passed_tests': passed_tests, 'failed_tests': failed_tests, 'success_rate': success_rate, 'results': [ { 'test_name': r.test_name, 'passed': r.passed, 'error_message': r.error_message, 'details': r.details } for r in self.results ] } with open(results_file, 'w') as f: json.dump(results_data, f, indent=2) self.logger.info(f"\nDetailed results saved to: {results_file}") def main(): """Main entry point""" parser = argparse.ArgumentParser(description='Comprehensive API Data Integrity Test Suite') parser.add_argument('--verbose', '-v', action='store_true', help='Enable verbose logging') parser.add_argument('--router', '-r', type=str, help='Test specific router only') args = parser.parse_args() tester = ComprehensiveAPITester(verbose=args.verbose) try: if args.router: tester.run_router_tests(args.router) else: tester.run_all_tests() except KeyboardInterrupt: tester.logger.info("\nTest suite interrupted by user") sys.exit(1) except Exception as e: tester.logger.error(f"Test suite failed with error: {e}") sys.exit(1) if __name__ == "__main__": main()