#!/usr/bin/env python3 """ Test Suite for Full Update Pipeline Tests each step individually and the complete pipeline """ import json import subprocess import time import requests from datetime import datetime from pathlib import Path class Colors: HEADER = '\033[95m' OKBLUE = '\033[94m' OKCYAN = '\033[96m' OKGREEN = '\033[92m' WARNING = '\033[93m' FAIL = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' def print_header(text): print(f"\n{Colors.HEADER}{Colors.BOLD}{'='*70}{Colors.ENDC}") print(f"{Colors.HEADER}{Colors.BOLD}{text:^70}{Colors.ENDC}") print(f"{Colors.HEADER}{Colors.BOLD}{'='*70}{Colors.ENDC}\n") def print_success(text): print(f"{Colors.OKGREEN}✅ {text}{Colors.ENDC}") def print_fail(text): print(f"{Colors.FAIL}❌ {text}{Colors.ENDC}") def print_warning(text): print(f"{Colors.WARNING}⚠️ {text}{Colors.ENDC}") def print_info(text): print(f"{Colors.OKCYAN}ℹ️ {text}{Colors.ENDC}") # ============================================================================ # TEST 1: API Server Status # ============================================================================ def test_api_server(): """Test if API server is running""" print_header("TEST 1: API Server Status") try: response = requests.get('http://localhost:5001/api/system-status', timeout=5) data = response.json() if data.get('success'): print_success("API server is running") print_info(f" Data freshness: {data.get('data_freshness', 'Unknown')}") print_info(f" Active jobs: {data.get('active_jobs', 0)}") return True else: print_fail("API server returned error") return False except requests.exceptions.ConnectionError: print_fail("Cannot connect to API server") print_warning(" Start it with: python3 criteria_api.py") return False except Exception as e: print_fail(f"API server test failed: {e}") return False # ============================================================================ # TEST 2: Data Files Exist # ============================================================================ def test_data_files(): """Test if required data files exist""" print_header("TEST 2: Required Data Files") script_dir = Path(__file__).parent required_files = { 'enriched_data.json': 'Property data with analysis', 'auth.json': 'Properstar authentication', } all_exist = True for filename, description in required_files.items(): filepath = script_dir / filename if filepath.exists(): size = filepath.stat().st_size print_success(f"{filename} exists ({size:,} bytes) - {description}") else: print_fail(f"{filename} missing - {description}") all_exist = False return all_exist # ============================================================================ # TEST 3: Current Data State # ============================================================================ def test_current_data(): """Analyze current property data""" print_header("TEST 3: Current Data State") script_dir = Path(__file__).parent data_file = script_dir / 'enriched_data.json' if not data_file.exists(): print_fail("enriched_data.json not found") return False try: with open(data_file, 'r', encoding='utf-8') as f: properties = json.load(f) total = len(properties) active = sum(1 for p in properties if p.get('status') != 'Removed') removed = total - active with_coords = sum(1 for p in properties if p.get('lat') and p.get('lon')) with_analysis = sum(1 for p in properties if p.get('analysis')) print_success(f"Loaded {total} properties") print_info(f" Active: {active} ({active/total*100:.1f}%)") print_info(f" Removed: {removed} ({removed/total*100:.1f}%)") print_info(f" With coordinates: {with_coords} ({with_coords/total*100:.1f}%)") print_info(f" With GPT analysis: {with_analysis} ({with_analysis/total*100:.1f}%)") # Check data freshness last_modified = data_file.stat().st_mtime hours_old = (time.time() - last_modified) / 3600 if hours_old < 24: print_success(f"Data is fresh ({hours_old:.1f} hours old)") elif hours_old < 168: # 7 days print_warning(f"Data is {hours_old/24:.1f} days old") else: print_fail(f"Data is stale ({hours_old/24:.1f} days old)") return True except Exception as e: print_fail(f"Failed to analyze data: {e}") return False # ============================================================================ # TEST 4: Scraping Capability # ============================================================================ def test_scraping(): """Test if scraping can run (auth check)""" print_header("TEST 4: Scraping Capability") script_dir = Path(__file__).parent auth_file = script_dir / 'auth.json' if not auth_file.exists(): print_fail("auth.json not found - cannot scrape") print_warning(" Create auth.json with your Properstar credentials") return False try: with open(auth_file, 'r') as f: auth = json.load(f) required_keys = ['email', 'password'] missing = [k for k in required_keys if k not in auth] if missing: print_fail(f"auth.json missing keys: {missing}") return False print_success("auth.json configured correctly") print_info(f" Email: {auth['email']}") return True except Exception as e: print_fail(f"auth.json is invalid: {e}") return False # ============================================================================ # TEST 5: Availability Check # ============================================================================ def test_availability_check(): """Test availability checking on one property""" print_header("TEST 5: Availability Check (Single Property)") script_dir = Path(__file__).parent data_file = script_dir / 'enriched_data.json' if not data_file.exists(): print_fail("No data to check availability") return False try: with open(data_file, 'r', encoding='utf-8') as f: properties = json.load(f) if not properties: print_fail("No properties in data") return False # Test on first property test_prop = properties[0] url = test_prop.get('url') print_info(f"Testing: {url}") # Import check function import sys sys.path.insert(0, str(script_dir)) from check_availability import check_property_availability result = check_property_availability(url, timeout=10) if result: status = "✅ Active" if result['available'] else "❌ Removed" print_success(f"Availability check successful: {status}") print_info(f" Reason: {result['reason']}") print_info(f" Status code: {result['status_code']}") return True else: print_fail("Availability check returned None") return False except Exception as e: print_fail(f"Availability check failed: {e}") return False # ============================================================================ # TEST 6: API Endpoints # ============================================================================ def test_api_endpoints(): """Test all API endpoints""" print_header("TEST 6: API Endpoints") endpoints = { 'GET /api/system-status': 'http://localhost:5001/api/system-status', 'GET /api/get-api-key': 'http://localhost:5001/api/get-api-key', } all_pass = True for name, url in endpoints.items(): try: response = requests.get(url, timeout=5) data = response.json() if data.get('success'): print_success(f"{name} ✓") else: print_warning(f"{name} returned success=false") all_pass = False except Exception as e: print_fail(f"{name} failed: {e}") all_pass = False return all_pass # ============================================================================ # TEST 7: Progress Tracking # ============================================================================ def test_progress_tracking(): """Test progress tracking system""" print_header("TEST 7: Progress Tracking System") # Create a test progress file test_job_id = 'test1234' progress_file = Path(f'/tmp/farmmatch_progress_{test_job_id}.json') try: # Create test progress file test_progress = { 'job_id': test_job_id, 'status': 'running', 'progress': 2, 'current_step': 'Testing progress tracking...', 'total_steps': 4, 'started_at': datetime.now().isoformat() } with open(progress_file, 'w') as f: json.dump(test_progress, f) print_success("Created test progress file") # Test API endpoint response = requests.get(f'http://localhost:5001/api/job-status/{test_job_id}', timeout=5) data = response.json() if data.get('success'): print_success("Progress API endpoint working") print_info(f" Status: {data.get('status')}") print_info(f" Progress: {data.get('progress')}/{data.get('total_steps')}") print_info(f" Current step: {data.get('current_step')}") # Cleanup progress_file.unlink() return True else: print_fail("Progress API returned error") return False except Exception as e: print_fail(f"Progress tracking test failed: {e}") # Cleanup if progress_file.exists(): progress_file.unlink() return False # ============================================================================ # TEST 8: Dry Run (Scrape Only) # ============================================================================ def test_scrape_only_dry_run(): """Test scrape-only mode (dry run - don't actually run)""" print_header("TEST 8: Scrape-Only Command Check") script_dir = Path(__file__).parent scraper = script_dir / 'auto_scrape_favorites.py' if not scraper.exists(): print_fail("auto_scrape_favorites.py not found") return False print_success("auto_scrape_favorites.py exists") print_info(" Command to run: python3 auto_scrape_favorites.py scrape-only") print_warning(" Not running actual scrape in test (would take 5-10 min)") return True # ============================================================================ # TEST 9: Full Pipeline Command Check # ============================================================================ def test_full_pipeline_dry_run(): """Test full pipeline command (dry run - don't actually run)""" print_header("TEST 9: Full Pipeline Command Check") script_dir = Path(__file__).parent required_scripts = [ 'auto_scrape_favorites.py', 'check_availability.py', 'geocode_properties.py', 'custom_criteria.py' ] all_exist = True for script in required_scripts: filepath = script_dir / script if filepath.exists(): print_success(f"{script} ✓") else: print_fail(f"{script} missing") all_exist = False if all_exist: print_success("All pipeline scripts present") print_info(" Command: python3 auto_scrape_favorites.py now") print_info(" Pipeline order:") print_info(" 1. Scrape favorites") print_info(" 2. Check availability") print_info(" 3. Geocode active properties") print_info(" 4. Analyze active properties") print_warning(" Not running actual pipeline in test (would take 20-30 min)") return all_exist # ============================================================================ # TEST 10: UI Button Simulation # ============================================================================ def test_ui_button_simulation(): """Simulate clicking the Full Update button""" print_header("TEST 10: UI Button Simulation") try: # Simulate POST request from UI print_info("Simulating 'Full Update' button click...") response = requests.post( 'http://localhost:5001/api/scrape-favorites', json={'full_pipeline': True}, timeout=5 ) data = response.json() if data.get('success'): job_id = data.get('job_id') print_success(f"Full Update initiated successfully") print_info(f" Job ID: {job_id}") print_info(f" Estimated time: {data.get('estimated_time')}") # Check job status print_info("\n Checking job status after 2 seconds...") time.sleep(2) status_response = requests.get( f'http://localhost:5001/api/job-status/{job_id}', timeout=5 ) status_data = status_response.json() if status_data.get('success'): print_success(f" Job status: {status_data.get('status')}") print_info(f" Current step: {status_data.get('current_step')}") # Note: The job is now running in background! print_warning("\n ⚠️ Job is now running in background!") print_warning(" It will take 20-30 minutes to complete.") print_warning(" Check progress at: http://localhost:8000/criteria_manager.html") return True else: print_fail("Failed to get job status") return False else: print_fail(f"Full Update failed: {data.get('message')}") return False except Exception as e: print_fail(f"UI button simulation failed: {e}") return False # ============================================================================ # RUN ALL TESTS # ============================================================================ def run_all_tests(): """Run all tests""" print(f"\n{Colors.BOLD}{Colors.OKBLUE}") print("╔════════════════════════════════════════════════════════════════════╗") print("║ FarmMatch Update Pipeline Test Suite ║") print("╚════════════════════════════════════════════════════════════════════╝") print(f"{Colors.ENDC}") tests = [ ("API Server Status", test_api_server), ("Required Data Files", test_data_files), ("Current Data State", test_current_data), ("Scraping Capability", test_scraping), ("Availability Check", test_availability_check), ("API Endpoints", test_api_endpoints), ("Progress Tracking", test_progress_tracking), ("Scrape-Only Command", test_scrape_only_dry_run), ("Full Pipeline Command", test_full_pipeline_dry_run), ] results = [] for name, test_func in tests: try: result = test_func() results.append((name, result)) except Exception as e: print_fail(f"Test '{name}' crashed: {e}") results.append((name, False)) # Summary print_header("TEST SUMMARY") passed = sum(1 for _, result in results if result) total = len(results) for name, result in results: status = "✅ PASS" if result else "❌ FAIL" color = Colors.OKGREEN if result else Colors.FAIL print(f"{color}{status:>10}{Colors.ENDC} | {name}") print(f"\n{Colors.BOLD}Results: {passed}/{total} tests passed ({passed/total*100:.0f}%){Colors.ENDC}\n") if passed == total: print_success("🎉 All tests passed! System is ready.") print_info("\nYou can now:") print_info(" 1. Open http://localhost:8000/criteria_manager.html") print_info(" 2. Click 'Full Update' button") print_info(" 3. Watch real-time progress") else: print_warning(f"⚠️ {total - passed} test(s) failed. Fix issues before running update.") # Optional: Ask if user wants to run actual Full Update print(f"\n{Colors.BOLD}Would you like to test the actual Full Update now? (y/N){Colors.ENDC}") print(f"{Colors.WARNING}⚠️ This will take 20-30 minutes and trigger actual scraping/analysis{Colors.ENDC}") if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == 'full-update-test': # Run actual full update test test_ui_button_simulation() else: # Run dry tests only run_all_tests()