File size: 8,080 Bytes
5b558e5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 |
"""
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!")
|