""" Test script for fail_delivery tool Verifies that delivery failure requires location and reason, and updates driver location correctly """ import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from chat.tools import ( handle_create_order, handle_create_driver, handle_create_assignment, handle_fail_delivery, handle_get_driver_details, handle_get_assignment_details ) print("=" * 70) print("Testing Delivery Failure Workflow") print("=" * 70) # Step 1: Create test order print("\n[1] Creating test order...") order_result = handle_create_order({ "customer_name": "Failure Test Customer", "customer_phone": "+8801712345678", "delivery_address": "Dhanmondi 32, Dhaka", "delivery_lat": 23.7465, "delivery_lng": 90.3760, "priority": "standard", "weight_kg": 2.5 }) if not order_result.get("success"): print(f"FAILED: {order_result.get('error')}") sys.exit(1) order_id = order_result["order_id"] print(f"SUCCESS: Order created: {order_id}") # Step 2: Create test driver at different location print("\n[2] Creating test driver at starting location...") driver_result = handle_create_driver({ "name": "Failure Test Driver", "phone": "+8801812345678", "vehicle_type": "motorcycle", "current_lat": 23.8103, # Mirpur area "current_lng": 90.4125 }) if not driver_result.get("success"): print(f"FAILED: {driver_result.get('error')}") sys.exit(1) driver_id = driver_result["driver_id"] driver_start_lat = 23.8103 driver_start_lng = 90.4125 print(f"SUCCESS: Driver created: {driver_id}") print(f" Driver starting location: ({driver_start_lat}, {driver_start_lng})") # Step 3: Create assignment print("\n[3] Creating assignment...") assignment_result = handle_create_assignment({ "order_id": order_id, "driver_id": driver_id }) if not assignment_result.get("success"): print(f"FAILED: {assignment_result.get('error')}") sys.exit(1) assignment_id = assignment_result["assignment_id"] print(f"SUCCESS: Assignment created: {assignment_id}") # Step 4: Get driver details BEFORE failure print("\n[4] Getting driver location BEFORE failure...") driver_before = handle_get_driver_details({"driver_id": driver_id}) if driver_before.get("success"): driver_data = driver_before["driver"] location = driver_data["location"] print(f"Driver location BEFORE: ({location['latitude']}, {location['longitude']})") print(f" Status: {driver_data['status']}") else: print(f"FAILED to get driver details") # Step 5: Test validation - attempt to fail without location print("\n[5] Testing validation - attempting to fail without GPS location...") fail_without_location = handle_fail_delivery({ "assignment_id": assignment_id, "failure_reason": "customer_not_available", "confirm": True }) if not fail_without_location.get("success"): print(f"EXPECTED FAILURE: {fail_without_location.get('error')}") else: print(f"UNEXPECTED: Should have failed without location!") # Step 6: Test validation - attempt to fail without reason print("\n[6] Testing validation - attempting to fail without reason...") fail_without_reason = handle_fail_delivery({ "assignment_id": assignment_id, "current_lat": 23.7500, "current_lng": 90.3800, "confirm": True }) if not fail_without_reason.get("success"): print(f"EXPECTED FAILURE: {fail_without_reason.get('error')}") else: print(f"UNEXPECTED: Should have failed without reason!") # Step 7: Test validation - attempt with invalid reason print("\n[7] Testing validation - attempting with invalid failure reason...") fail_invalid_reason = handle_fail_delivery({ "assignment_id": assignment_id, "current_lat": 23.7500, "current_lng": 90.3800, "failure_reason": "invalid_reason", "confirm": True }) if not fail_invalid_reason.get("success"): print(f"EXPECTED FAILURE: {fail_invalid_reason.get('error')}") else: print(f"UNEXPECTED: Should have failed with invalid reason!") # Step 8: Fail delivery properly with location and reason print("\n[8] Failing delivery with proper location and reason...") # Driver reports failure from a location along the route (between start and delivery) failure_lat = 23.7750 # Somewhere between Mirpur and Dhanmondi failure_lng = 90.3950 failure_reason = "customer_not_available" failure_result = handle_fail_delivery({ "assignment_id": assignment_id, "current_lat": failure_lat, "current_lng": failure_lng, "failure_reason": failure_reason, "confirm": True, "notes": "Customer phone was switched off. Attempted contact 3 times." }) if not failure_result.get("success"): print(f"FAILED: {failure_result.get('error')}") sys.exit(1) print(f"SUCCESS: Delivery marked as failed!") print(f" Assignment ID: {failure_result['assignment_id']}") print(f" Order ID: {failure_result['order_id']}") print(f" Customer: {failure_result['customer_name']}") print(f" Driver: {failure_result['driver_name']}") print(f" Failed at: {failure_result['failed_at']}") print(f" Failure reason: {failure_result['failure_reason_display']}") # Check driver location update driver_loc = failure_result.get("driver_location", {}) print(f"\nDriver location UPDATE:") print(f" New location: ({driver_loc.get('lat')}, {driver_loc.get('lng')})") print(f" Updated at: {driver_loc.get('updated_at')}") # Cascading actions cascading = failure_result.get("cascading_actions", []) if cascading: print(f"\nCascading actions:") for action in cascading: print(f" - {action}") # Step 9: Get driver details AFTER failure to verify location changed print("\n[9] Verifying driver location AFTER failure...") driver_after = handle_get_driver_details({"driver_id": driver_id}) if driver_after.get("success"): driver_data = driver_after["driver"] location = driver_data["location"] after_lat = location['latitude'] after_lng = location['longitude'] print(f"Driver location AFTER: ({after_lat}, {after_lng})") print(f" Status: {driver_data['status']}") # Verify location matches reported failure location if abs(after_lat - failure_lat) < 0.0001 and abs(after_lng - failure_lng) < 0.0001: print(f"\nSUCCESS: Driver location updated to reported failure position!") print(f" Expected: ({failure_lat}, {failure_lng})") print(f" Got: ({after_lat}, {after_lng})") else: print(f"\nFAILED: Driver location NOT updated correctly") print(f" Expected: ({failure_lat}, {failure_lng})") print(f" Got: ({after_lat}, {after_lng})") else: print(f"FAILED to get driver details after failure") # Step 10: Verify assignment status and failure reason print("\n[10] Verifying assignment status and failure reason...") assignment_details = handle_get_assignment_details({"assignment_id": assignment_id}) if assignment_details.get("success"): assignment = assignment_details.get("assignment", {}) print(f"Assignment status: {assignment.get('status')}") print(f"Failure reason: {assignment.get('failure_reason')}") print(f"Notes: {assignment.get('notes')}") print(f"Actual arrival: {assignment.get('actual_arrival')}") order = assignment.get("order", {}) print(f"Order status: {order.get('status')}") if (assignment.get('status') == 'failed' and order.get('status') == 'failed' and assignment.get('failure_reason') == failure_reason): print(f"\nSUCCESS: Assignment and order statuses updated correctly!") print(f"SUCCESS: Failure reason stored correctly!") else: print(f"\nFAILED: Statuses or failure reason not updated correctly") else: print(f"FAILED to get assignment details") print("\n" + "=" * 70) print("Delivery Failure Test Complete!") print("=" * 70) # Cleanup print("\nCleaning up test data...") from chat.tools import handle_delete_order, handle_delete_driver handle_delete_order({"order_id": order_id, "confirm": True}) handle_delete_driver({"driver_id": driver_id, "confirm": True}) print("Cleanup complete!")