File size: 4,916 Bytes
d69447e |
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 |
"""
Test script to verify PostgreSQL database operations
"""
import sys
from pathlib import Path
from datetime import datetime, timedelta
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from database.connection import execute_query, execute_write
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def test_insert_order():
"""Test inserting a new order"""
logger.info("Testing order insertion...")
now = datetime.now()
time_window_start = now + timedelta(hours=2)
time_window_end = now + timedelta(hours=6)
query = """
INSERT INTO orders (
order_id, customer_name, customer_phone, customer_email,
delivery_address, delivery_lat, delivery_lng,
time_window_start, time_window_end,
priority, weight_kg, status
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
params = (
"ORD-TEST-001",
"John Doe",
"+1-555-0123",
"[email protected]",
"123 Main Street, San Francisco, CA 94103",
37.7749,
-122.4194,
time_window_start,
time_window_end,
"standard",
5.5,
"pending"
)
try:
result = execute_write(query, params)
logger.info(f"✓ Order inserted successfully (rows affected: {result})")
return True
except Exception as e:
logger.error(f"✗ Failed to insert order: {e}")
return False
def test_query_orders():
"""Test querying orders"""
logger.info("Testing order query...")
query = "SELECT * FROM orders WHERE status = %s"
params = ("pending",)
try:
results = execute_query(query, params)
logger.info(f"✓ Query successful: Found {len(results)} pending orders")
for row in results:
logger.info(f" Order ID: {row['order_id']}")
logger.info(f" Customer: {row['customer_name']}")
logger.info(f" Address: {row['delivery_address']}")
logger.info(f" Priority: {row['priority']}")
logger.info(f" Status: {row['status']}")
logger.info(" ---")
return True
except Exception as e:
logger.error(f"✗ Failed to query orders: {e}")
return False
def test_update_order():
"""Test updating an order"""
logger.info("Testing order update...")
query = "UPDATE orders SET status = %s, assigned_driver_id = %s WHERE order_id = %s"
params = ("assigned", "DRV-001", "ORD-TEST-001")
try:
result = execute_write(query, params)
logger.info(f"✓ Order updated successfully (rows affected: {result})")
# Verify update
verify_query = "SELECT status, assigned_driver_id FROM orders WHERE order_id = %s"
verify_result = execute_query(verify_query, ("ORD-TEST-001",))
if verify_result:
row = verify_result[0]
logger.info(f" New status: {row['status']}")
logger.info(f" Assigned driver: {row['assigned_driver_id']}")
return True
except Exception as e:
logger.error(f"✗ Failed to update order: {e}")
return False
def test_delete_order():
"""Test deleting the test order"""
logger.info("Testing order deletion (cleanup)...")
query = "DELETE FROM orders WHERE order_id = %s"
params = ("ORD-TEST-001",)
try:
result = execute_write(query, params)
logger.info(f"✓ Order deleted successfully (rows affected: {result})")
return True
except Exception as e:
logger.error(f"✗ Failed to delete order: {e}")
return False
def main():
"""Run all database tests"""
logger.info("=" * 50)
logger.info("Starting FleetMind PostgreSQL Database Tests")
logger.info("=" * 50)
tests = [
("Insert Order", test_insert_order),
("Query Orders", test_query_orders),
("Update Order", test_update_order),
("Delete Order", test_delete_order),
]
results = []
for test_name, test_func in tests:
logger.info(f"\n--- {test_name} ---")
success = test_func()
results.append((test_name, success))
# Summary
logger.info("\n" + "=" * 50)
logger.info("Test Summary")
logger.info("=" * 50)
passed = sum(1 for _, success in results if success)
total = len(results)
for test_name, success in results:
status = "✓ PASSED" if success else "✗ FAILED"
logger.info(f"{test_name}: {status}")
logger.info(f"\nTotal: {passed}/{total} tests passed")
if passed == total:
logger.info("\n🎉 All tests passed! Your PostgreSQL database is working correctly!")
return 0
else:
logger.error("\n❌ Some tests failed. Please check the errors above.")
return 1
if __name__ == "__main__":
sys.exit(main())
|