""" Comprehensive test script for NutriScan Backend API Tests the complete workflow: User profile -> Image upload -> OCR -> AI analysis """ import requests import json import time import os from io import BytesIO from PIL import Image, ImageDraw, ImageFont BASE_URL = "http://localhost:5000" API_BASE = f"{BASE_URL}/api" TEST_USER = { "username": "testuser_" + str(int(time.time())), "email": f"test_{int(time.time())}@example.com", "password": "testpassword123" } TEST_PROFILE = { "age": 30, "gender": "Female", "weight": 65.0, "height": 165.0, "medical_conditions": ["diabetes", "hypertension"], "allergies": ["peanuts", "shellfish"], "medications": ["metformin", "lisinopril"], "dietary_restrictions": ["gluten-free", "low-sodium"], "health_goals": ["weight management", "blood sugar control"] } class NutriScanTester: def __init__(self): self.session = requests.Session() self.access_token = None self.user_id = None def log(self, message, status="INFO"): timestamp = time.strftime("%H:%M:%S") print(f"[{timestamp}] {status}: {message}") def test_health_check(self): """Test if the server is running""" self.log("Testing server health...") try: response = self.session.get(f"{API_BASE}/health") if response.status_code == 200: self.log(" Server is healthy") return True else: self.log(f" Health check failed: {response.status_code}", "ERROR") return False except Exception as e: self.log(f" Cannot connect to server: {e}", "ERROR") return False def test_user_registration(self): """Test user registration""" self.log("Testing user registration...") try: response = self.session.post( f"{API_BASE}/auth/register", json=TEST_USER ) if response.status_code == 201: data = response.json() self.access_token = data['access_token'] self.user_id = data['user']['id'] self.session.headers.update({ 'Authorization': f'Bearer {self.access_token}' }) self.log(f" User registered successfully: {TEST_USER['username']}") return True else: self.log(f" Registration failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" Registration error: {e}", "ERROR") return False def test_profile_update(self): """Test updating user health profile""" self.log("Testing profile update with health information...") try: response = self.session.put( f"{API_BASE}/auth/profile", json=TEST_PROFILE ) if response.status_code == 200: data = response.json() self.log(" Profile updated successfully") self.log(f" Age: {data['user']['age']}") self.log(f" Medical conditions: {data['user']['medical_conditions']}") self.log(f" Allergies: {data['user']['allergies']}") return True else: self.log(f" Profile update failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" Profile update error: {e}", "ERROR") return False def create_test_image(self): """Create a test image with ingredient text""" self.log("Creating test ingredient label image...") img = Image.new('RGB', (400, 300), color='white') draw = ImageDraw.Draw(img) try: font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 20) except: font = ImageFont.load_default() text = """INGREDIENTS: Sugar, Palm Oil, Peanuts, Wheat Flour, Salt, E202 (Potassium Sorbate), Artificial Flavoring, May contain traces of nuts""" draw.text((20, 50), text, fill='black', font=font) img_bytes = BytesIO() img.save(img_bytes, format='PNG') img_bytes.seek(0) return img_bytes def test_image_analysis(self): """Test complete image analysis workflow""" self.log("Testing complete image analysis workflow...") test_image = self.create_test_image() try: files = { 'image': ('test_ingredients.png', test_image, 'image/png') } response = self.session.post( f"{API_BASE}/inference/analyze-image", files=files ) if response.status_code == 200: data = response.json() result = data['result'] self.log(" Image analysis completed successfully") self.log(f" Extracted text: {result['extracted_text'][:100]}...") self.log(f" Ingredients found: {len(result['extracted_ingredients'])}") self.log(f" Ingredients: {result['ingredients_text']}") self.log(f" Personalized analysis: {result['personalized']}") self.log(f" OCR processing time: {result['processing_time']['ocr_processing']:.2f}s") self.log(f" AI analysis time: {result['processing_time']['ai_analysis']:.2f}s") self.log(f" Analysis preview: {result['analysis'][:150]}...") return True, data['request_id'] else: self.log(f" Image analysis failed: {response.text}", "ERROR") return False, None except Exception as e: self.log(f" Image analysis error: {e}", "ERROR") return False, None def test_ocr_methods(self): """Test OCR methods specifically""" self.log("Testing OCR methods (Google Vision + PaddleOCR)...") test_image = self.create_test_image() try: files = { 'image': ('test_ocr.png', test_image, 'image/png') } response = self.session.post( f"{API_BASE}/inference/test-ocr", files=files ) if response.status_code == 200: data = response.json() results = data['results'] summary = data['summary'] self.log(" OCR methods tested successfully") self.log(f" Google Vision available: {summary['google_available']}") self.log(f" PaddleOCR available: {summary['paddle_available']}") self.log(f" Primary method used: {summary['primary_method_used']}") self.log(f" Primary success: {summary['primary_success']}") if results['google_vision']['success']: self.log(f" Google Vision text preview: {results['google_vision']['text'][:100]}...") else: self.log(f" Google Vision error: {results['google_vision']['error']}") if results['paddle_ocr']['success']: self.log(f" PaddleOCR text preview: {results['paddle_ocr']['text'][:100]}...") else: self.log(f" PaddleOCR error: {results['paddle_ocr']['error']}") return True else: self.log(f" OCR test failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" OCR test error: {e}", "ERROR") return False def test_text_analysis(self): """Test direct text ingredient analysis""" self.log("Testing direct ingredient analysis...") test_ingredients = "sugar, palm oil, peanuts, wheat flour, artificial colors" try: response = self.session.post( f"{API_BASE}/inference/analyze-ingredients", json={"ingredients": test_ingredients} ) if response.status_code == 200: data = response.json() result = data['result'] self.log(" Text analysis completed successfully") self.log(f" Ingredients: {result['ingredients']}") self.log(f" Personalized: {result['personalized']}") self.log(f" Processing time: {result['processing_time']:.2f}s") self.log(f" Analysis preview: {result['analysis'][:150]}...") return True else: self.log(f" Text analysis failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" Text analysis error: {e}", "ERROR") return False def test_health_suggestions(self): """Test personalized health suggestions""" self.log("Testing personalized health suggestions...") try: response = self.session.get(f"{API_BASE}/inference/health-suggestions") if response.status_code == 200: data = response.json() result = data['result'] self.log(" Health suggestions generated successfully") self.log(f" Has profile: {result['has_profile']}") self.log(f" Processing time: {result['processing_time']:.2f}s") self.log(f" Suggestions preview: {result['suggestions'][:150]}...") return True else: self.log(f" Health suggestions failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" Health suggestions error: {e}", "ERROR") return False def test_quick_scan(self): """Test quick ingredient scan""" self.log("Testing quick ingredient scan...") try: response = self.session.post( f"{API_BASE}/inference/quick-scan", json={"ingredient": "aspartame"} ) if response.status_code == 200: data = response.json() result = data['result'] self.log(" Quick scan completed successfully") self.log(f" Ingredient: {result['ingredient']}") self.log(f" Personalized: {result['personalized']}") self.log(f" Assessment: {result['safety_assessment']}") return True else: self.log(f" Quick scan failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" Quick scan error: {e}", "ERROR") return False def test_history(self): """Test request history retrieval""" self.log("Testing request history...") try: response = self.session.get(f"{API_BASE}/inference/history") if response.status_code == 200: data = response.json() history = data['history'] self.log(f" History retrieved successfully") self.log(f" Total requests: {len(history)}") if history: self.log(f" Latest request type: {history[0]['analysis_type']}") self.log(f" Latest request time: {history[0]['created_at']}") return True else: self.log(f" History retrieval failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" History retrieval error: {e}", "ERROR") return False def test_model_status(self): """Test model status""" self.log("Testing model status...") try: response = self.session.get(f"{API_BASE}/inference/model-status") if response.status_code == 200: data = response.json() self.log(f" Model status retrieved") self.log(f" Model loaded: {data['model_loaded']}") self.log(f" Device: {data['device']}") self.log(f" Model path: {data['model_path']}") return True else: self.log(f" Model status failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" Model status error: {e}", "ERROR") return False def test_user_login(self): """Test user login (already done in registration)""" self.log(" User login already completed during registration") return True def test_inference_history(self): """Test inference history retrieval""" self.log("Testing inference history...") try: response = self.session.get(f"{API_BASE}/inference/history") if response.status_code == 200: data = response.json() history = data['history'] self.log(f" History retrieved successfully") self.log(f" Total requests: {len(history)}") if history: self.log(f" Latest request type: {history[0]['analysis_type']}") self.log(f" Latest request time: {history[0]['created_at']}") return True else: self.log(f" History retrieval failed: {response.text}", "ERROR") return False except Exception as e: self.log(f" History retrieval error: {e}", "ERROR") return False def run_all_tests(self): """Run all tests in sequence""" self.log("=" * 60, "HEADER") self.log(" STARTING COMPLETE BACKEND WORKFLOW TESTS", "HEADER") self.log("=" * 60, "HEADER") success = self.test_health_check() if not success: self.log(" Server not running - please start the backend first", "ERROR") return False success = self.test_user_registration() if not success: self.log(" Registration failed - stopping tests", "ERROR") return False success = self.test_user_login() if not success: self.log(" Login failed - stopping tests", "ERROR") return False success = self.test_profile_update() if not success: self.log(" Profile update failed", "WARNING") success = self.test_ocr_methods() if not success: self.log(" OCR test failed - but continuing with workflow", "WARNING") success, request_id = self.test_image_analysis() if not success: self.log(" Image analysis failed", "ERROR") return False success = self.test_health_suggestions() if not success: self.log(" Health suggestions failed", "WARNING") success = self.test_inference_history() if not success: self.log(" Inference history failed", "WARNING") self.log("=" * 60, "HEADER") self.log(" ALL TESTS COMPLETED SUCCESSFULLY!", "SUCCESS") self.log("Backend is ready for frontend integration", "SUCCESS") self.log("=" * 60, "HEADER") return True def main(): """Main test execution""" print("NutriScan Backend API Test Suite") print("=" * 60) print("This script tests the complete workflow:") print("1. User registration and profile setup") print("2. Image upload and OCR processing") print("3. AI-powered ingredient analysis") print("4. Personalized recommendations") print("5. Request history tracking") print("=" * 60) tester = NutriScanTester() success = tester.run_all_tests() if success: print(f"\n Backend is ready for frontend integration!") print(f" Base URL: {BASE_URL}") print(f" Main endpoint: POST {API_BASE}/inference/analyze-image") print(f" Documentation: See API_DOCUMENTATION.md") return success if __name__ == "__main__": try: success = main() exit(0 if success else 1) except KeyboardInterrupt: print("\n\n⏹ Tests interrupted by user") exit(1) except Exception as e: print(f"\n\n Test suite error: {e}") exit(1)