#!/usr/bin/env python3 """ Abbas FIT Backend API Testing Suite Tests all new backend endpoints that need testing """ import requests import json import base64 from datetime import datetime, timedelta import uuid # Configuration BASE_URL = "https://fitcoach-108.preview.emergentagent.com/api" ADMIN_EMAIL = "admin@abbasfit.com" ADMIN_PASSWORD = "admin" class AbbasFitTester: def __init__(self): self.admin_token = None self.user_token = None self.test_user_id = None self.session = requests.Session() self.session.headers.update({ 'Content-Type': 'application/json', 'Accept': 'application/json' }) def log(self, message, level="INFO"): timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] {level}: {message}") def login_admin(self): """Login as admin and get JWT token""" try: response = self.session.post(f"{BASE_URL}/auth/login", json={ "email": ADMIN_EMAIL, "password": ADMIN_PASSWORD }) if response.status_code == 200: data = response.json() self.admin_token = data["access_token"] self.session.headers.update({ 'Authorization': f'Bearer {self.admin_token}' }) self.log(f"✅ Admin login successful - Token: {self.admin_token[:20]}...") return True else: self.log(f"❌ Admin login failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Admin login error: {str(e)}", "ERROR") return False def get_existing_user(self): """Get an existing user for user-specific endpoint testing""" try: # Get existing users response = self.session.get(f"{BASE_URL}/admin/users") if response.status_code == 200: users = response.json() if users: # Use the first user user = users[0] self.test_user_id = user["id"] self.log(f"✅ Using existing user - ID: {self.test_user_id}") # For user-specific tests, we'll use admin token since we can't login as user # This is a limitation but allows us to test the endpoints self.user_token = self.admin_token return True else: self.log("❌ No existing users found", "ERROR") return False else: self.log(f"❌ Failed to get existing users: {response.status_code}", "ERROR") return False except Exception as e: self.log(f"❌ Get existing user error: {str(e)}", "ERROR") return False def test_water_intake_logging(self): """Test water intake logging endpoints""" self.log("🧪 Testing Water Intake Logging...") # Switch to user token self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) try: # Test POST /user/water - Log water intake water_data = {"amount_ml": 500} response = self.session.post(f"{BASE_URL}/user/water", json=water_data) if response.status_code == 200: self.log("✅ Water logging successful") # Log another entry self.session.post(f"{BASE_URL}/user/water", json={"amount_ml": 300}) # Test GET /user/water/today - Get today's water intake today_response = self.session.get(f"{BASE_URL}/user/water/today") if today_response.status_code == 200: data = today_response.json() if "total_ml" in data and "progress_percent" in data: self.log(f"✅ Today's water intake: {data['total_ml']}ml ({data['progress_percent']}%)") return True else: self.log("❌ Water today response missing required fields", "ERROR") return False else: self.log(f"❌ Get today's water failed: {today_response.status_code}", "ERROR") return False else: self.log(f"❌ Water logging failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Water intake testing error: {str(e)}", "ERROR") return False def test_step_tracking(self): """Test step tracking endpoints""" self.log("🧪 Testing Step Tracking...") try: # Test POST /user/steps - Log steps steps_data = {"steps": 5000} response = self.session.post(f"{BASE_URL}/user/steps", json=steps_data) if response.status_code == 200: self.log("✅ Step logging successful") # Test GET /user/steps/today - Get today's steps today_response = self.session.get(f"{BASE_URL}/user/steps/today") if today_response.status_code == 200: data = today_response.json() if "steps" in data and "progress_percent" in data: self.log(f"✅ Today's steps: {data['steps']} ({data['progress_percent']}%)") # Test GET /user/steps/history - Get step history history_response = self.session.get(f"{BASE_URL}/user/steps/history") if history_response.status_code == 200: history = history_response.json() self.log(f"✅ Step history retrieved: {len(history)} entries") return True else: self.log(f"❌ Step history failed: {history_response.status_code}", "ERROR") return False else: self.log("❌ Steps today response missing required fields", "ERROR") return False else: self.log(f"❌ Get today's steps failed: {today_response.status_code}", "ERROR") return False else: self.log(f"❌ Step logging failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Step tracking testing error: {str(e)}", "ERROR") return False def test_supplements_management(self): """Test supplements management endpoints""" self.log("🧪 Testing Supplements Management...") # Switch to admin token self.session.headers.update({'Authorization': f'Bearer {self.admin_token}'}) try: # Test POST /admin/supplements - Create supplement supplement_data = { "name": "Whey Protein", "description": "High quality whey protein powder", "dosage": "30g", "timing": "Post workout" } response = self.session.post(f"{BASE_URL}/admin/supplements", json=supplement_data) if response.status_code == 200: supplement = response.json() supplement_id = supplement["id"] self.log("✅ Supplement created successfully") # Test GET /admin/supplements - List all supplements list_response = self.session.get(f"{BASE_URL}/admin/supplements") if list_response.status_code == 200: supplements = list_response.json() self.log(f"✅ Supplements list retrieved: {len(supplements)} supplements") # Test POST /admin/supplements/assign - Assign supplement to user assign_data = { "user_id": self.test_user_id, "supplement_id": supplement_id, "dosage": "25g", "timing": "After breakfast", "notes": "Start with smaller dose" } assign_response = self.session.post(f"{BASE_URL}/admin/supplements/assign", json=assign_data) if assign_response.status_code == 200: self.log("✅ Supplement assigned to user") # Switch to user token for user endpoints self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) # Test GET /user/supplements - Get user's assigned supplements user_supplements_response = self.session.get(f"{BASE_URL}/user/supplements") if user_supplements_response.status_code == 200: user_supplements = user_supplements_response.json() self.log(f"✅ User supplements retrieved: {len(user_supplements)} assigned") # Test POST /user/supplements/log - Log supplement taken log_data = { "supplement_id": supplement_id, "taken": True } log_response = self.session.post(f"{BASE_URL}/user/supplements/log", json=log_data) if log_response.status_code == 200: self.log("✅ Supplement intake logged") # Test GET /user/supplements/today - Get today's supplement logs today_response = self.session.get(f"{BASE_URL}/user/supplements/today") if today_response.status_code == 200: today_data = today_response.json() if "assigned" in today_data and "taken_today" in today_data: self.log(f"✅ Today's supplement status retrieved") return True else: self.log("❌ Today's supplements response missing fields", "ERROR") return False else: self.log(f"❌ Today's supplements failed: {today_response.status_code}", "ERROR") return False else: self.log(f"❌ Supplement logging failed: {log_response.status_code}", "ERROR") return False else: self.log(f"❌ User supplements failed: {user_supplements_response.status_code}", "ERROR") return False else: self.log(f"❌ Supplement assignment failed: {assign_response.status_code}", "ERROR") return False else: self.log(f"❌ Supplements list failed: {list_response.status_code}", "ERROR") return False else: self.log(f"❌ Supplement creation failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Supplements management testing error: {str(e)}", "ERROR") return False def test_progress_photos(self): """Test progress photos endpoints""" self.log("🧪 Testing Progress Photos...") # Switch to user token self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) try: # Create a simple base64 image (1x1 pixel PNG) sample_image = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==" # Test POST /user/progress-photos - Upload progress photo photo_data = { "photo_base64": sample_image, "photo_type": "front", "week_number": 1, "notes": "Starting progress photo" } response = self.session.post(f"{BASE_URL}/user/progress-photos", json=photo_data) if response.status_code == 200: self.log("✅ Progress photo uploaded successfully") # Test GET /user/progress-photos - Get user's progress photos get_response = self.session.get(f"{BASE_URL}/user/progress-photos") if get_response.status_code == 200: photos = get_response.json() self.log(f"✅ Progress photos retrieved: {len(photos)} photos") return True else: self.log(f"❌ Get progress photos failed: {get_response.status_code}", "ERROR") return False else: self.log(f"❌ Progress photo upload failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Progress photos testing error: {str(e)}", "ERROR") return False def test_session_booking(self): """Test session booking endpoints""" self.log("🧪 Testing Session Booking...") # Switch to user token self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) try: # Test POST /user/sessions - Book session tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d") session_data = { "date": tomorrow, "time_slot": "10:00", "notes": "First training session" } response = self.session.post(f"{BASE_URL}/user/sessions", json=session_data) if response.status_code == 200: session = response.json() session_id = session["id"] self.log("✅ Session booked successfully") # Test GET /user/sessions - Get user's sessions get_response = self.session.get(f"{BASE_URL}/user/sessions") if get_response.status_code == 200: sessions = get_response.json() self.log(f"✅ User sessions retrieved: {len(sessions)} sessions") # Switch to admin token for admin endpoints self.session.headers.update({'Authorization': f'Bearer {self.admin_token}'}) # Test GET /admin/sessions - Get all sessions (admin only) admin_sessions_response = self.session.get(f"{BASE_URL}/admin/sessions") if admin_sessions_response.status_code == 200: admin_sessions = admin_sessions_response.json() self.log(f"✅ Admin sessions retrieved: {len(admin_sessions)} sessions") # Test PUT /admin/sessions/{session_id}/status - Update session status status_response = self.session.put(f"{BASE_URL}/admin/sessions/{session_id}/status?status=confirmed") if status_response.status_code == 200: self.log("✅ Session status updated") # Switch back to user token self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) # Test DELETE /user/sessions/{session_id} - Cancel session cancel_response = self.session.delete(f"{BASE_URL}/user/sessions/{session_id}") if cancel_response.status_code == 200: self.log("✅ Session cancelled successfully") return True else: self.log(f"❌ Session cancellation failed: {cancel_response.status_code}", "ERROR") return False else: self.log(f"❌ Session status update failed: {status_response.status_code}", "ERROR") return False else: self.log(f"❌ Admin sessions failed: {admin_sessions_response.status_code}", "ERROR") return False else: self.log(f"❌ Get user sessions failed: {get_response.status_code}", "ERROR") return False else: self.log(f"❌ Session booking failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Session booking testing error: {str(e)}", "ERROR") return False def test_notifications_system(self): """Test notifications system endpoints""" self.log("🧪 Testing Notifications System...") # Switch to admin token self.session.headers.update({'Authorization': f'Bearer {self.admin_token}'}) try: # Test POST /admin/notifications/send - Send notification to user send_data = { "user_id": self.test_user_id, "title": "Welcome!", "message": "Welcome to Abbas FIT! Let's start your fitness journey.", "notification_type": "general" } response = self.session.post(f"{BASE_URL}/admin/notifications/send", params=send_data) if response.status_code == 200: self.log("✅ Notification sent to user") # Test POST /admin/notifications/broadcast - Broadcast to all users broadcast_data = { "title": "System Update", "message": "New features have been added to the app!" } broadcast_response = self.session.post(f"{BASE_URL}/admin/notifications/broadcast", params=broadcast_data) if broadcast_response.status_code == 200: self.log("✅ Notification broadcasted to all users") # Switch to user token for user endpoints self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) # Test GET /user/notifications - Get user notifications notifications_response = self.session.get(f"{BASE_URL}/user/notifications") if notifications_response.status_code == 200: notifications = notifications_response.json() self.log(f"✅ User notifications retrieved: {len(notifications)} notifications") if notifications: notification_id = notifications[0]["id"] # Test GET /user/notifications/unread-count - Get unread count unread_response = self.session.get(f"{BASE_URL}/user/notifications/unread-count") if unread_response.status_code == 200: unread_data = unread_response.json() self.log(f"✅ Unread notifications count: {unread_data['count']}") # Test PUT /user/notifications/{id}/read - Mark as read read_response = self.session.put(f"{BASE_URL}/user/notifications/{notification_id}/read") if read_response.status_code == 200: self.log("✅ Notification marked as read") # Test PUT /user/notifications/read-all - Mark all as read read_all_response = self.session.put(f"{BASE_URL}/user/notifications/read-all") if read_all_response.status_code == 200: self.log("✅ All notifications marked as read") return True else: self.log(f"❌ Mark all read failed: {read_all_response.status_code}", "ERROR") return False else: self.log(f"❌ Mark notification read failed: {read_response.status_code}", "ERROR") return False else: self.log(f"❌ Unread count failed: {unread_response.status_code}", "ERROR") return False else: self.log("✅ No notifications found (expected for new user)") return True else: self.log(f"❌ Get user notifications failed: {notifications_response.status_code}", "ERROR") return False else: self.log(f"❌ Notification broadcast failed: {broadcast_response.status_code}", "ERROR") return False else: self.log(f"❌ Send notification failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Notifications system testing error: {str(e)}", "ERROR") return False def test_weekly_comparison(self): """Test weekly comparison endpoint""" self.log("🧪 Testing Weekly Comparison...") # Switch to user token self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) try: # Test GET /user/weekly-comparison - Get this vs last week comparison response = self.session.get(f"{BASE_URL}/user/weekly-comparison") if response.status_code == 200: data = response.json() if "this_week" in data and "last_week" in data and "changes" in data: self.log("✅ Weekly comparison retrieved successfully") self.log(f" This week calories: {data['this_week']['meals']['total_calories']}") self.log(f" Last week calories: {data['last_week']['meals']['total_calories']}") return True else: self.log("❌ Weekly comparison response missing required fields", "ERROR") return False else: self.log(f"❌ Weekly comparison failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Weekly comparison testing error: {str(e)}", "ERROR") return False def test_admin_insights(self): """Test admin insights endpoints""" self.log("🧪 Testing Admin Insights...") # Switch to admin token self.session.headers.update({'Authorization': f'Bearer {self.admin_token}'}) try: # Test GET /admin/insights - Get admin dashboard stats response = self.session.get(f"{BASE_URL}/admin/insights") if response.status_code == 200: data = response.json() if "total_users" in data and "adherence_rate" in data: self.log(f"✅ Admin insights retrieved: {data['total_users']} users, {data['adherence_rate']}% adherence") # Test GET /admin/user-insights/{user_id} - Get detailed user insights user_insights_response = self.session.get(f"{BASE_URL}/admin/user-insights/{self.test_user_id}") if user_insights_response.status_code == 200: user_data = user_insights_response.json() if "user" in user_data and "weekly_comparison" in user_data: self.log("✅ User insights retrieved successfully") return True else: self.log("❌ User insights response missing required fields", "ERROR") return False else: self.log(f"❌ User insights failed: {user_insights_response.status_code}", "ERROR") return False else: self.log("❌ Admin insights response missing required fields", "ERROR") return False else: self.log(f"❌ Admin insights failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Admin insights testing error: {str(e)}", "ERROR") return False def test_enhanced_diet_workout_plans(self): """Test enhanced diet and workout plans with 7 meals""" self.log("🧪 Testing Enhanced Diet/Workout Plans...") # Switch to admin token self.session.headers.update({'Authorization': f'Bearer {self.admin_token}'}) try: # Test POST /admin/diet-plans - Create diet plan with 7 meals diet_plan_data = { "name": "Weight Loss Plan", "description": "Comprehensive 7-meal diet plan for weight loss", "meals": [ { "meal_type": "breakfast", "name": "Oats with Fruits", "recipe": "50g oats + 1 banana + 10 almonds", "calories": 350, "protein_g": 12, "carbs_g": 45, "fat_g": 8, "time": "07:00 AM" }, { "meal_type": "mid_morning_snack", "name": "Green Tea + Nuts", "recipe": "1 cup green tea + 5 walnuts", "calories": 150, "protein_g": 4, "carbs_g": 5, "fat_g": 12, "time": "10:00 AM" } ], "total_calories": 2000, "assigned_to": [self.test_user_id] } response = self.session.post(f"{BASE_URL}/admin/diet-plans", json=diet_plan_data) if response.status_code == 200: diet_plan = response.json() diet_plan_id = diet_plan["id"] self.log("✅ Enhanced diet plan created successfully") # Test GET /admin/diet-plans - List all diet plans list_response = self.session.get(f"{BASE_URL}/admin/diet-plans") if list_response.status_code == 200: plans = list_response.json() self.log(f"✅ Diet plans list retrieved: {len(plans)} plans") # Test PUT /admin/diet-plans/{id} - Update diet plan update_data = { "name": "Updated Weight Loss Plan", "description": "Updated comprehensive diet plan", "meals": diet_plan_data["meals"], "total_calories": 1800, "assigned_to": [self.test_user_id] } update_response = self.session.put(f"{BASE_URL}/admin/diet-plans/{diet_plan_id}", json=update_data) if update_response.status_code == 200: self.log("✅ Diet plan updated successfully") # Test workout plans workout_plan_data = { "name": "Beginner Strength Training", "description": "3-day strength training program", "exercises": [ { "name": "Push-ups", "sets": 3, "reps": 10, "rest_seconds": 60, "instructions": "Keep body straight" }, { "name": "Squats", "sets": 3, "reps": 15, "rest_seconds": 90, "instructions": "Go down until thighs parallel" } ], "duration_minutes": 45, "assigned_to": [self.test_user_id] } workout_response = self.session.post(f"{BASE_URL}/admin/workout-plans", json=workout_plan_data) if workout_response.status_code == 200: self.log("✅ Workout plan created successfully") # Switch to user token to test user endpoints self.session.headers.update({'Authorization': f'Bearer {self.user_token}'}) # Test GET /user/diet-plans - Get assigned diet plans user_diet_response = self.session.get(f"{BASE_URL}/user/diet-plans") if user_diet_response.status_code == 200: user_diet_plans = user_diet_response.json() self.log(f"✅ User diet plans retrieved: {len(user_diet_plans)} plans") # Test GET /user/workout-plans - Get assigned workout plans user_workout_response = self.session.get(f"{BASE_URL}/user/workout-plans") if user_workout_response.status_code == 200: user_workout_plans = user_workout_response.json() self.log(f"✅ User workout plans retrieved: {len(user_workout_plans)} plans") return True else: self.log(f"❌ User workout plans failed: {user_workout_response.status_code}", "ERROR") return False else: self.log(f"❌ User diet plans failed: {user_diet_response.status_code}", "ERROR") return False else: self.log(f"❌ Workout plan creation failed: {workout_response.status_code}", "ERROR") return False else: self.log(f"❌ Diet plan update failed: {update_response.status_code}", "ERROR") return False else: self.log(f"❌ Diet plans list failed: {list_response.status_code}", "ERROR") return False else: self.log(f"❌ Diet plan creation failed: {response.status_code} - {response.text}", "ERROR") return False except Exception as e: self.log(f"❌ Enhanced diet/workout plans testing error: {str(e)}", "ERROR") return False def run_all_tests(self): """Run all backend tests in priority order""" self.log("🚀 Starting Abbas FIT Backend API Testing Suite") self.log(f"Testing against: {BASE_URL}") results = {} # Initialize if not self.login_admin(): self.log("❌ Cannot proceed without admin login", "ERROR") return results if not self.get_existing_user(): self.log("❌ Cannot proceed without test user", "ERROR") return results # High Priority Tests self.log("\n=== HIGH PRIORITY TESTS ===") results["water_intake_logging"] = self.test_water_intake_logging() results["step_tracking"] = self.test_step_tracking() results["supplements_management"] = self.test_supplements_management() results["progress_photos"] = self.test_progress_photos() # Medium Priority Tests self.log("\n=== MEDIUM PRIORITY TESTS ===") results["session_booking"] = self.test_session_booking() results["notifications_system"] = self.test_notifications_system() results["weekly_comparison"] = self.test_weekly_comparison() results["admin_insights"] = self.test_admin_insights() results["enhanced_diet_workout_plans"] = self.test_enhanced_diet_workout_plans() # Summary self.log("\n=== TEST RESULTS SUMMARY ===") passed = sum(1 for result in results.values() if result) total = len(results) for test_name, result in results.items(): status = "✅ PASS" if result else "❌ FAIL" self.log(f"{test_name}: {status}") self.log(f"\nOverall: {passed}/{total} tests passed ({(passed/total)*100:.1f}%)") return results if __name__ == "__main__": tester = AbbasFitTester() results = tester.run_all_tests()