339 lines
16 KiB
Python
339 lines
16 KiB
Python
import json
|
|
from django.test import TestCase, Client
|
|
from django.urls import reverse
|
|
from shipping.models import ActivityLog
|
|
|
|
class ShippingAPITestCase(TestCase):
|
|
def setUp(self):
|
|
self.client = Client()
|
|
|
|
def test_healthcheck_api(self):
|
|
response = self.client.get(reverse('api_healthcheck'))
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data['status'], 'healthy')
|
|
self.assertEqual(data['service'], 'Shadowfax Shipping Microservice')
|
|
|
|
def test_forward_serviceability(self):
|
|
url = reverse('api_forward_router', kwargs={'action_name': 'serviceability'})
|
|
payload = {"pincodes": "560077,110009", "use_mock": True}
|
|
response = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data['status'], 'SUCCESS')
|
|
|
|
def test_forward_generate_awb(self):
|
|
url = reverse('api_forward_router', kwargs={'action_name': 'generate-awb'})
|
|
payload = {"count": 2, "use_mock": True}
|
|
response = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_forward_create_order(self):
|
|
url = reverse('api_forward_router', kwargs={'action_name': 'create-order'})
|
|
payload = {
|
|
"use_mock": True,
|
|
"payload": {
|
|
"order_details": {"client_order_id": "TEST-ORD-001", "actual_weight": 500, "product_value": 999},
|
|
"customer_details": {"name": "Test User", "phone": "9876543210", "pincode": "560038"}
|
|
}
|
|
}
|
|
response = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_forward_track_and_bulk_track(self):
|
|
url_single = reverse('api_forward_router', kwargs={'action_name': 'track'})
|
|
response = self.client.get(f"{url_single}?awb=SFX9928374102")
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
url_bulk = reverse('api_forward_router', kwargs={'action_name': 'bulk-track'})
|
|
payload = {"awb_numbers": ["SFX9928374102", "SFX8817263541"], "use_mock": True}
|
|
response = self.client.post(url_bulk, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_forward_escalation_pod_qr(self):
|
|
# Support escalation
|
|
url_esc = reverse('api_forward_router', kwargs={'action_name': 'escalation'})
|
|
payload = {"awb_number": "SFX9928374102", "category": "EXPEDITE_PICKUP", "use_mock": True}
|
|
response = self.client.post(url_esc, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# POD details
|
|
url_pod = reverse('api_forward_router', kwargs={'action_name': 'get-pod'})
|
|
payload_pod = {"awb_number": "SFX9928374102", "use_mock": True}
|
|
response = self.client.post(url_pod, data=json.dumps(payload_pod), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# Dynamic QR
|
|
url_qr = reverse('api_forward_router', kwargs={'action_name': 'generate-qr'})
|
|
payload_qr = {"awb_number": "SFX9928374102", "cod_amount": 1499.0, "use_mock": True}
|
|
response = self.client.post(url_qr, data=json.dumps(payload_qr), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_reverse_router_actions(self):
|
|
# Serviceability
|
|
url_serv = reverse('api_reverse_router', kwargs={'action_name': 'serviceability'})
|
|
payload = {"pincodes": "560077", "use_mock": True}
|
|
res = self.client.post(url_serv, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
# Create request
|
|
url_create = reverse('api_reverse_router', kwargs={'action_name': 'create-request'})
|
|
payload_req = {
|
|
"use_mock": True,
|
|
"payload": {"client_request_id": "REV-TEST-101", "actual_weight": 500, "product_value": 1500}
|
|
}
|
|
res = self.client.post(url_create, data=json.dumps(payload_req), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
# Update QC
|
|
url_qc = reverse('api_reverse_router', kwargs={'action_name': 'update-qc'})
|
|
payload_qc = {"client_request_id": "REV-TEST-101", "qc_check": True, "use_mock": True}
|
|
res = self.client.put(url_qc, data=json.dumps(payload_qc), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
def test_ekart_router_gateway(self):
|
|
url_track = reverse('api_ekart_router', kwargs={'action_name': 'track'})
|
|
res = self.client.get(f"{url_track}?tracking_id=500999A3408005")
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
def test_webhook_listener(self):
|
|
url = reverse('webhook_listener_api')
|
|
payload = {"client_order_id": "ORD-101", "status": "delivered"}
|
|
res = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
self.assertEqual(ActivityLog.objects.filter(service_type='WEBHOOK').count(), 1)
|
|
|
|
def test_api_logs_export(self):
|
|
url = reverse('api_logs_export')
|
|
res = self.client.get(url)
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertIn('count', data)
|
|
self.assertIn('logs', data)
|
|
|
|
from unittest.mock import patch
|
|
|
|
@patch('shipping.carrier_allocator.check_shadowfax_availability')
|
|
@patch('shipping.carrier_allocator.check_ekart_availability')
|
|
@patch('shipping.carrier_allocator.check_delhivery_availability')
|
|
def test_carrier_check_availability(self, mock_delhivery, mock_ekart, mock_sfx):
|
|
mock_sfx.return_value = {"carrier": "SHADOWFAX", "serviceable": True, "cod_available": True, "estimated_cost": 60.0}
|
|
mock_ekart.return_value = {"carrier": "EKART", "serviceable": True, "cod_available": True, "estimated_cost": 50.0}
|
|
mock_delhivery.return_value = {"carrier": "DELHIVERY", "serviceable": True, "cod_available": True, "estimated_cost": 46.0}
|
|
|
|
url = reverse('api_carrier_availability')
|
|
payload = {
|
|
"pickup_pincode": "560077",
|
|
"drop_pincode": "110009",
|
|
"payment_type": "Prepaid",
|
|
"weight": 500
|
|
}
|
|
res = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
# Test omitting weight returns 400 Bad Request
|
|
res_no_weight = self.client.post(url, data=json.dumps({
|
|
"pickup_pincode": "560077",
|
|
"drop_pincode": "110009"
|
|
}), content_type='application/json')
|
|
self.assertEqual(res_no_weight.status_code, 400)
|
|
self.assertEqual(res_no_weight.json()['status'], 'error')
|
|
self.assertIn("weight is required", res_no_weight.json()['message'])
|
|
data = res.json()
|
|
self.assertEqual(data['status'], 'success')
|
|
self.assertIn('SHADOWFAX', data['carriers'])
|
|
self.assertIn('EKART', data['carriers'])
|
|
self.assertIn('DELHIVERY', data['carriers'])
|
|
|
|
@patch('shipping.carrier_allocator.check_shadowfax_availability')
|
|
@patch('shipping.carrier_allocator.check_ekart_availability')
|
|
@patch('shipping.carrier_allocator.check_delhivery_availability')
|
|
@patch('delhivery.client.DelhiveryAPIClient.create_shipment')
|
|
def test_carrier_auto_dispatch_cheapest_delhivery(self, mock_create, mock_delhivery, mock_ekart, mock_sfx):
|
|
mock_sfx.return_value = {"carrier": "SHADOWFAX", "serviceable": True, "cod_available": True, "estimated_cost": 60.0}
|
|
mock_ekart.return_value = {"carrier": "EKART", "serviceable": True, "cod_available": True, "estimated_cost": 50.0}
|
|
mock_delhivery.return_value = {"carrier": "DELHIVERY", "serviceable": True, "cod_available": True, "estimated_cost": 46.0}
|
|
mock_create.return_value = {"success": True, "packages": [{"waybill": "123456789"}]}
|
|
|
|
url = reverse('api_carrier_auto_dispatch')
|
|
payload = {
|
|
"selection_strategy": "CHEAPEST",
|
|
"payload": {
|
|
"order_details": {"client_order_id": "ORD-AUTO-DELHIVERY-101", "actual_weight": 500, "payment_mode": "Prepaid"},
|
|
"customer_details": {"name": "Test Consignee", "phone": "9876543210", "pincode": "110009"},
|
|
"pickup_details": {"name": "Test Warehouse", "pincode": "560077"}
|
|
}
|
|
}
|
|
res = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertEqual(data['assigned_carrier'], 'DELHIVERY')
|
|
self.assertEqual(data['estimated_cost'], 46.0)
|
|
|
|
@patch('shipping.carrier_allocator.check_shadowfax_availability')
|
|
@patch('shipping.carrier_allocator.check_ekart_availability')
|
|
@patch('shipping.carrier_allocator.check_delhivery_availability')
|
|
def test_carrier_tie_breaker_hierarchy(self, mock_delhivery, mock_ekart, mock_sfx):
|
|
# All 3 carriers offer the exact same price of ₹50.00
|
|
mock_sfx.return_value = {"carrier": "SHADOWFAX", "serviceable": True, "cod_available": True, "estimated_cost": 50.0}
|
|
mock_ekart.return_value = {"carrier": "EKART", "serviceable": True, "cod_available": True, "estimated_cost": 50.0}
|
|
mock_delhivery.return_value = {"carrier": "DELHIVERY", "serviceable": True, "cod_available": True, "estimated_cost": 50.0}
|
|
|
|
from shipping.carrier_allocator import select_best_carrier
|
|
avail = {
|
|
"carriers": {
|
|
"SHADOWFAX": mock_sfx.return_value,
|
|
"EKART": mock_ekart.return_value,
|
|
"DELHIVERY": mock_delhivery.return_value
|
|
}
|
|
}
|
|
|
|
# Scenario 1: All 3 tie -> Shadowfax selected (#1 priority)
|
|
res1 = select_best_carrier(avail, selection_strategy='CHEAPEST')
|
|
self.assertEqual(res1['carrier'], 'SHADOWFAX')
|
|
|
|
# Scenario 2: Shadowfax unserviceable, Delhivery & Ekart tie -> Delhivery selected (#2 priority)
|
|
avail_no_sfx = {
|
|
"carriers": {
|
|
"SHADOWFAX": {"carrier": "SHADOWFAX", "serviceable": False},
|
|
"EKART": mock_ekart.return_value,
|
|
"DELHIVERY": mock_delhivery.return_value
|
|
}
|
|
}
|
|
res2 = select_best_carrier(avail_no_sfx, selection_strategy='CHEAPEST')
|
|
self.assertEqual(res2['carrier'], 'DELHIVERY')
|
|
|
|
@patch('delhivery.client.DelhiveryAPIClient.track_shipment')
|
|
@patch('delhivery.client.DelhiveryAPIClient.check_serviceability')
|
|
def test_delhivery_router_gateway(self, mock_serv, mock_track):
|
|
mock_track.return_value = {"ShipmentData": [{"Shipment": {"AWB": "1234567890123"}}]}
|
|
mock_serv.return_value = {"delivery_codes": [{"postal_code": {"pin": 560077, "pre_paid": "Y"}}]}
|
|
|
|
url_track = reverse('api_delhivery_router', kwargs={'action_name': 'track'})
|
|
res = self.client.get(f"{url_track}?waybill=1234567890123")
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
url_serv = reverse('api_delhivery_router', kwargs={'action_name': 'serviceability'})
|
|
res_serv = self.client.get(f"{url_serv}?pincode=560077")
|
|
self.assertEqual(res_serv.status_code, 200)
|
|
|
|
def test_delhivery_official_serviceability_matrix(self):
|
|
from shipping.delhivery_serviceability import (
|
|
get_delhivery_pincode_info,
|
|
is_delhivery_serviceable,
|
|
check_delhivery_route_serviceability
|
|
)
|
|
from shipping.carrier_allocator import resolve_city_state_by_pincode, check_delhivery_availability
|
|
|
|
# 1. Pincode metadata
|
|
info = get_delhivery_pincode_info('600001')
|
|
self.assertIsNotNone(info)
|
|
self.assertEqual(info['city'], 'Chennai')
|
|
self.assertEqual(info['state'], 'Tamil Nadu')
|
|
self.assertEqual(info['state_code'], 'TN')
|
|
self.assertTrue(info['prepaid'])
|
|
|
|
# 2. Location resolution
|
|
city, state = resolve_city_state_by_pincode('600001')
|
|
self.assertEqual(city, 'Chennai')
|
|
self.assertEqual(state, 'Tamil Nadu')
|
|
|
|
city_blr, state_blr = resolve_city_state_by_pincode('560001')
|
|
self.assertEqual(city_blr, 'Bangalore')
|
|
self.assertEqual(state_blr, 'Karnataka')
|
|
|
|
# 3. Valid route check
|
|
res_valid = check_delhivery_availability('560077', '600001', payment_type='Prepaid', weight=500.0)
|
|
self.assertTrue(res_valid['serviceable'])
|
|
self.assertEqual(res_valid['city'], 'Chennai')
|
|
|
|
# 4. Invalid pincode check
|
|
res_invalid = check_delhivery_availability('560077', '999999', weight=500.0)
|
|
self.assertFalse(res_invalid['serviceable'])
|
|
|
|
def test_carrier_checker_ui_and_first_last_mile(self):
|
|
# Test UI template rendering
|
|
response = self.client.get(reverse('carrier_checker_view'))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "Multi-Carrier Route & Serviceability Checker")
|
|
self.assertContains(response, "First-Mile (Pickup Hub)")
|
|
self.assertContains(response, "Last-Mile (Delivery)")
|
|
|
|
# Test API response with first-mile and last-mile breakdown
|
|
api_url = reverse('api_carrier_availability')
|
|
payload = {
|
|
"pickup_pincode": "560077",
|
|
"drop_pincode": "600001",
|
|
"payment_type": "COD",
|
|
"weight": 500,
|
|
"use_mock": True
|
|
}
|
|
api_res = self.client.post(api_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(api_res.status_code, 200)
|
|
data = api_res.json()
|
|
self.assertEqual(data['status'], 'success')
|
|
self.assertIn('SHADOWFAX', data['carriers'])
|
|
self.assertIn('DELHIVERY', data['carriers'])
|
|
|
|
sfx = data['carriers']['SHADOWFAX']
|
|
self.assertIn('first_mile', sfx)
|
|
self.assertIn('last_mile', sfx)
|
|
self.assertTrue(sfx['first_mile']['serviceable'])
|
|
self.assertTrue(sfx['last_mile']['serviceable'])
|
|
|
|
delhivery = data['carriers']['DELHIVERY']
|
|
self.assertIn('first_mile', delhivery)
|
|
self.assertIn('last_mile', delhivery)
|
|
self.assertTrue(delhivery['first_mile']['serviceable'])
|
|
self.assertTrue(delhivery['last_mile']['serviceable'])
|
|
|
|
def test_single_pincode_checker_api(self):
|
|
# Test single pincode endpoint with live/mock evaluation across all 3 carriers
|
|
url = reverse('api_carrier_pincode_check')
|
|
res = self.client.get(f"{url}?pincode=560077&weight=500&use_mock=true")
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertEqual(data['status'], 'success')
|
|
self.assertEqual(data['pincode'], '560077')
|
|
self.assertIn('city', data)
|
|
self.assertIn('carriers', data)
|
|
self.assertIn('best_carrier', data)
|
|
|
|
for carrier_name in ['SHADOWFAX', 'DELHIVERY', 'EKART']:
|
|
self.assertIn(carrier_name, data['carriers'])
|
|
c = data['carriers'][carrier_name]
|
|
self.assertIn('first_mile_available', c)
|
|
self.assertIn('last_mile_available', c)
|
|
self.assertIn('delivery_types', c)
|
|
self.assertIn('estimated_charge', c)
|
|
self.assertIsInstance(c['delivery_types'], list)
|
|
self.assertGreater(len(c['delivery_types']), 0)
|
|
self.assertIsNotNone(c['estimated_charge'])
|
|
|
|
self.assertTrue(data['best_carrier']['selected'])
|
|
self.assertIsNotNone(data['best_carrier']['charge'])
|
|
|
|
def test_first_mile_pickup_serviceability_validation(self):
|
|
from shipping.carrier_allocator import check_shadowfax_availability, check_single_pincode_all_carriers
|
|
|
|
# Pincode 641301 (Mettupalayam/Coimbatore) has no Shadowfax seller_pickup hub
|
|
res = check_single_pincode_all_carriers('641301', weight=1000)
|
|
self.assertEqual(res['status'], 'success')
|
|
sfx = res['carriers']['SHADOWFAX']
|
|
self.assertFalse(sfx['first_mile_available'])
|
|
self.assertFalse(sfx['serviceable'])
|
|
self.assertIn("NOT serviceable", sfx['first_mile_details'])
|
|
|
|
# Route check from 641301 to 600059
|
|
route = check_shadowfax_availability('641301', '600059', weight=1000)
|
|
self.assertFalse(route['serviceable'])
|
|
self.assertFalse(route['first_mile_available'])
|
|
self.assertFalse(route['first_mile']['serviceable'])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|