File size: 6,011 Bytes
5625fed |
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 |
"""
Test script for complete_delivery tool
Verifies that delivery completion 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_complete_delivery,
handle_get_driver_details,
handle_get_assignment_details
)
print("=" * 70)
print("Testing Delivery Completion Workflow")
print("=" * 70)
# Step 1: Create test order
print("\n[1] Creating test order...")
order_result = handle_create_order({
"customer_name": "Completion Test Customer",
"customer_phone": "+8801712345678",
"delivery_address": "Ahsanullah University, Dhaka",
"delivery_lat": 23.7808,
"delivery_lng": 90.4130,
"priority": "standard",
"weight_kg": 3.0
})
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": "Completion Test Driver",
"phone": "+8801812345678",
"vehicle_type": "motorcycle",
"current_lat": 23.7549, # Different from delivery location
"current_lng": 90.3909
})
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.7549
driver_start_lng = 90.3909
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 completion
print("\n[4] Getting driver location BEFORE delivery completion...")
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: Complete delivery
print("\n[5] Completing delivery...")
completion_result = handle_complete_delivery({
"assignment_id": assignment_id,
"confirm": True,
"actual_distance_meters": 4500,
"notes": "Delivered successfully to security desk"
})
if not completion_result.get("success"):
print(f"FAILED: {completion_result.get('error')}")
sys.exit(1)
print(f"SUCCESS: Delivery completed!")
print(f" Assignment ID: {completion_result['assignment_id']}")
print(f" Order ID: {completion_result['order_id']}")
print(f" Customer: {completion_result['customer_name']}")
print(f" Driver: {completion_result['driver_name']}")
print(f" Completed at: {completion_result['completed_at']}")
# Check driver location update
driver_updated = completion_result.get("driver_updated", {})
print(f"\nDriver location UPDATE:")
print(f" New location: {driver_updated.get('new_location', 'N/A')}")
print(f" Updated at: {driver_updated.get('location_updated_at', 'N/A')}")
# Cascading actions
cascading = completion_result.get("cascading_actions", [])
if cascading:
print(f"\nCascading actions:")
for action in cascading:
print(f" - {action}")
# Step 6: Get driver details AFTER completion to verify location changed
print("\n[6] Verifying driver location AFTER delivery completion...")
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 changed
delivery_lat = 23.7808
delivery_lng = 90.4130
if abs(after_lat - delivery_lat) < 0.0001 and abs(after_lng - delivery_lng) < 0.0001:
print(f"\nSUCCESS: Driver location updated to delivery address!")
print(f" Expected: ({delivery_lat}, {delivery_lng})")
print(f" Got: ({after_lat}, {after_lng})")
else:
print(f"\nFAILED: Driver location NOT updated correctly")
print(f" Expected: ({delivery_lat}, {delivery_lng})")
print(f" Got: ({after_lat}, {after_lng})")
else:
print(f"FAILED to get driver details after completion")
# Step 7: Verify assignment status
print("\n[7] Verifying assignment status...")
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"Actual arrival: {assignment.get('actual_arrival')}")
order = assignment.get("order", {})
print(f"Order status: {order.get('status')}")
if assignment.get('status') == 'completed' and order.get('status') == 'delivered':
print(f"\nSUCCESS: Assignment and order statuses updated correctly!")
else:
print(f"\nFAILED: Statuses not updated correctly")
else:
print(f"FAILED to get assignment details")
print("\n" + "=" * 70)
print("Delivery Completion 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!")
|