""" Test script to verify duplicate assignment prevention Ensures that an order cannot be assigned to multiple drivers simultaneously """ 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 ) print("=" * 70) print("Testing Duplicate Assignment Prevention") print("=" * 70) # Step 1: Create test order print("\n[1] Creating test order...") order_result = handle_create_order({ "customer_name": "Duplicate Test Customer", "customer_phone": "+8801712345678", "delivery_address": "Uttara, Dhaka", "delivery_lat": 23.8759, "delivery_lng": 90.3795, "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 first driver print("\n[2] Creating first driver...") driver1_result = handle_create_driver({ "name": "Driver One", "phone": "+8801812345678", "vehicle_type": "motorcycle", "current_lat": 23.8103, "current_lng": 90.4125 }) if not driver1_result.get("success"): print(f"FAILED: {driver1_result.get('error')}") sys.exit(1) driver1_id = driver1_result["driver_id"] print(f"SUCCESS: Driver 1 created: {driver1_id} (Driver One)") # Step 3: Create second driver (add small delay to avoid ID collision) print("\n[3] Creating second driver...") import time time.sleep(0.1) # Small delay to ensure different timestamp driver2_result = handle_create_driver({ "name": "Driver Two", "phone": "+8801912345678", "vehicle_type": "car", "current_lat": 23.7465, "current_lng": 90.3760 }) if not driver2_result.get("success"): print(f"FAILED: {driver2_result.get('error')}") sys.exit(1) driver2_id = driver2_result["driver_id"] print(f"SUCCESS: Driver 2 created: {driver2_id} (Driver Two)") # Step 4: Assign order to first driver print(f"\n[4] Assigning order {order_id} to Driver One...") assignment1_result = handle_create_assignment({ "order_id": order_id, "driver_id": driver1_id }) if not assignment1_result.get("success"): print(f"FAILED: {assignment1_result.get('error')}") sys.exit(1) assignment1_id = assignment1_result["assignment_id"] print(f"SUCCESS: Assignment created: {assignment1_id}") print(f" Order {order_id} assigned to Driver One") # Step 5: Attempt to assign the same order to second driver (should fail) print(f"\n[5] Attempting to assign same order to Driver Two (should fail)...") assignment2_result = handle_create_assignment({ "order_id": order_id, "driver_id": driver2_id }) if not assignment2_result.get("success"): error_msg = assignment2_result.get('error', '') print(f"EXPECTED FAILURE: {error_msg}") # Verify error message contains expected information if "already assigned" in error_msg.lower() and "Driver One" in error_msg: print(f"SUCCESS: Error message correctly identifies existing assignment!") print(f" - Mentions order is already assigned") print(f" - Shows driver name (Driver One)") print(f" - Shows assignment ID ({assignment1_id})") else: print(f"WARNING: Error message could be more descriptive") else: print(f"FAILED: Should have prevented duplicate assignment!") print(f" Unexpected assignment created: {assignment2_result.get('assignment_id')}") sys.exit(1) # Step 6: Attempt to assign same order to SAME driver again (should also fail) print(f"\n[6] Attempting to assign same order to Driver One again (should also fail)...") assignment3_result = handle_create_assignment({ "order_id": order_id, "driver_id": driver1_id }) if not assignment3_result.get("success"): error_msg = assignment3_result.get('error', '') print(f"EXPECTED FAILURE: {error_msg}") print(f"SUCCESS: Correctly prevents reassigning to same driver!") else: print(f"FAILED: Should have prevented duplicate assignment to same driver!") sys.exit(1) print("\n" + "=" * 70) print("Duplicate Assignment Prevention Test Complete!") print("=" * 70) print("\nSummary:") print(" - Order can be assigned to a driver: YES") print(" - Same order can be assigned to another driver: NO (prevented)") print(" - Same order can be reassigned to same driver: NO (prevented)") print(" - Error message is informative: YES") # 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": driver1_id, "confirm": True}) handle_delete_driver({"driver_id": driver2_id, "confirm": True}) print("Cleanup complete!")