diff --git a/shipengine_sdk/jsonrpc/__init__.py b/shipengine_sdk/jsonrpc/__init__.py index 39dee0e..902dd94 100644 --- a/shipengine_sdk/jsonrpc/__init__.py +++ b/shipengine_sdk/jsonrpc/__init__.py @@ -42,4 +42,4 @@ def rpc_request_loop( else: raise err retry += 1 - return api_response + return api_response diff --git a/shipengine_sdk/models/track_pacakge/__init__.py b/shipengine_sdk/models/track_pacakge/__init__.py index db95781..9aab008 100644 --- a/shipengine_sdk/models/track_pacakge/__init__.py +++ b/shipengine_sdk/models/track_pacakge/__init__.py @@ -1,7 +1,7 @@ """Data objects to be used in the `track_package` and `track` methods.""" import json from dataclasses import dataclass -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union from dataclasses_json import LetterCase, dataclass_json # type: ignore @@ -18,11 +18,11 @@ class Shipment: account_id: Optional[str] = None carrier_account: Optional[CarrierAccount] = None carrier: Carrier - estimated_delivery_date: str - actual_delivery_date: str + estimated_delivery_date: Union[IsoString, str] + actual_delivery_date: Union[IsoString, str] def __init__( - self, shipment: Dict[str, Any], actual_delivery_date: str, config: ShipEngineConfig + self, shipment: Dict[str, Any], actual_delivery_date: IsoString, config: ShipEngineConfig ) -> None: """This object represents a given Shipment.""" self.config = config @@ -43,9 +43,7 @@ def __init__( else ShipEngineError("The carrierCode field was null from api response.") ) - self.estimated_delivery_date = IsoString( - iso_string=shipment["estimatedDelivery"] - ).to_string() + self.estimated_delivery_date = IsoString(iso_string=shipment["estimatedDelivery"]) self.actual_delivery_date = actual_delivery_date def _get_carrier_account(self, carrier: str, account_id: str) -> CarrierAccount: @@ -58,11 +56,11 @@ def _get_carrier_account(self, carrier: str, account_id: str) -> CarrierAccount: for account in carrier_accounts: if account_id == account.account_id: target_carrier.append(account) + return target_carrier[0] else: raise ShipEngineError( message=f"accountID [{account_id}] doesn't match any of the accounts connected to your ShipEngine Account." # noqa ) - return target_carrier[0] def to_dict(self) -> Dict[str, Any]: if hasattr(self, "config"): @@ -123,8 +121,8 @@ class Location: class TrackingEvent: - date_time: str - carrier_date_time: str + date_time: Union[IsoString, str] + carrier_date_time: Union[IsoString, str] status: str description: Optional[str] carrier_status_code: Optional[str] @@ -134,15 +132,9 @@ class TrackingEvent: def __init__(self, event: Dict[str, Any]) -> None: """Tracking event object.""" - if IsoString(event["timestamp"]).has_timezone(): - self.date_time = IsoString(iso_string=event["timestamp"]).to_string() - else: - self.date_time = IsoString(event["timestamp"]).to_string() + self.date_time = IsoString(iso_string=event["timestamp"]) - if IsoString(iso_string=event["carrierTimestamp"]).has_timezone(): - self.carrier_date_time = IsoString(iso_string=event["carrierTimestamp"]).to_string() - else: - self.carrier_date_time = IsoString(event["carrierTimestamp"]).to_string() + self.carrier_date_time = IsoString(iso_string=event["carrierTimestamp"]) self.status = event["status"] self.description = event["description"] if "description" in event else None @@ -150,7 +142,7 @@ def __init__(self, event: Dict[str, Any]) -> None: event["carrierStatusCode"] if "carrierStatusCode" in event else None ) self.signer = event["signer"] if "signer" in event else None - self.location = Location.from_dict(event["location"]) if "location" in event else None # type: ignore + self.location = Location.from_dict(event["location"]) if "location" in event else None def to_dict(self): return (lambda o: o.__dict__)(self) diff --git a/shipengine_sdk/util/iso_string.py b/shipengine_sdk/util/iso_string.py index da5623d..4966d20 100644 --- a/shipengine_sdk/util/iso_string.py +++ b/shipengine_sdk/util/iso_string.py @@ -25,7 +25,10 @@ def to_string(self) -> str: return self.iso_string def to_datetime_object(self) -> datetime: - return datetime.strptime(self.iso_string, "%Y-%m-%dT%H:%M:%S.%fZ") + if self.has_timezone(): + return datetime.strptime(self.iso_string, "%Y-%m-%dT%H:%M:%S.%fZ") + elif self.is_valid_iso_string_no_tz(self.iso_string): + return datetime.fromisoformat(self.iso_string) def has_timezone(self) -> bool: if self.is_valid_iso_string(self.iso_string): diff --git a/tests/services/test_track_package.py b/tests/services/test_track_package.py index 7f9c886..32b13fe 100644 --- a/tests/services/test_track_package.py +++ b/tests/services/test_track_package.py @@ -1,9 +1,60 @@ """Testing the `track_package` method of the ShipEngine SDK.""" -from shipengine_sdk.models import TrackingQuery +from typing import List + +from shipengine_sdk.models import TrackingQuery, TrackPackageResult from tests.util.test_helpers import configurable_stub_shipengine_instance, stub_config +def assertions_on_delivered_after_exception_or_multiple_attempts( + tracking_result: TrackPackageResult, +) -> None: + track_package_assertions(tracking_result=tracking_result) + does_delivery_date_match(tracking_result) + assert_events_in_order(tracking_result.events) + assert len(tracking_result.events) == 8 + assert tracking_result.events[0].status == "accepted" + assert tracking_result.events[1].status == "in_transit" + assert tracking_result.events[2].status == "unknown" + assert tracking_result.events[3].status == "in_transit" + assert tracking_result.events[4].status == "exception" + assert tracking_result.events[5].status == "exception" + assert tracking_result.events[6].status == "in_transit" + assert tracking_result.events[7].status == "delivered" + assert tracking_result.events[-1].status == "delivered" + + +def does_delivery_date_match(tracking_result: TrackPackageResult) -> None: + """Check that the delivery dates for a given tracking response match.""" + assert ( + tracking_result.shipment.actual_delivery_date.to_datetime_object() + == tracking_result.events[-1].date_time.to_datetime_object() + ) + + +def assert_events_in_order(events: List) -> None: + """""" + previous_date_time = events[0].date_time + for event in events: + assert event.date_time.to_datetime_object() >= previous_date_time.to_datetime_object() + previous_date_time = event.date_time + + +def track_package_assertions(tracking_result: TrackPackageResult) -> None: + """""" + carrier_account_carrier_code = tracking_result.shipment.carrier_account.carrier["code"] + carrier_code = tracking_result.shipment.carrier["code"] + estimated_delivery = tracking_result.shipment.estimated_delivery_date + + assert carrier_account_carrier_code is not None + assert type(carrier_account_carrier_code) is str + assert carrier_code is not None + assert type(carrier_code) is str + assert estimated_delivery.has_timezone() is True + + class TestTrackPackage: + _PACKAGE_ID_FEDEX_ACCEPTED: str = "pkg_1FedExAccepted" + def test_track_by_tracking_number_and_carrier_code(self) -> None: """DX-1084 - Test track by tracking number and carrier code.""" shipengine = configurable_stub_shipengine_instance(stub_config()) @@ -14,3 +65,135 @@ def test_track_by_tracking_number_and_carrier_code(self) -> None: assert tracking_data.tracking_number == tracking_result.package.tracking_number assert tracking_result.package.tracking_url is not None assert type(tracking_result.package.tracking_url) is str + + def test_track_by_package_id(self) -> None: + """DX-1086 - Test track by package ID.""" + package_id = self._PACKAGE_ID_FEDEX_ACCEPTED + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + assert tracking_result.package.package_id == package_id + assert tracking_result.package.tracking_number is not None + assert tracking_result.package.tracking_url is not None + assert tracking_result.shipment.shipment_id is not None + assert tracking_result.shipment.account_id is not None + + def test_initial_scan_tracking_event(self) -> None: + """DX-1088 - Test initial scan tracking event.""" + package_id = self._PACKAGE_ID_FEDEX_ACCEPTED + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + track_package_assertions(tracking_result=tracking_result) + assert len(tracking_result.events) == 1 + assert tracking_result.events[0].status == "accepted" + + def test_out_for_delivery_tracking_event(self) -> None: + """DX-1089 - Test out for delivery tracking event.""" + package_id = "pkg_1FedExAttempted" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + track_package_assertions(tracking_result=tracking_result) + assert len(tracking_result.events) == 5 + assert tracking_result.events[0].status == "accepted" + assert tracking_result.events[1].status == "in_transit" + + def test_multiple_delivery_attempts(self) -> None: + """DX-1090 - Test multiple delivery attempt events.""" + package_id = "pkg_1FedexDeLiveredAttempted" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + track_package_assertions(tracking_result=tracking_result) + assert len(tracking_result.events) == 9 + assert_events_in_order(tracking_result.events) + assert tracking_result.events[0].status == "accepted" + assert tracking_result.events[1].status == "in_transit" + assert tracking_result.events[2].status == "unknown" + assert tracking_result.events[3].status == "in_transit" + assert tracking_result.events[4].status == "attempted_delivery" + assert tracking_result.events[5].status == "in_transit" + assert tracking_result.events[6].status == "attempted_delivery" + assert tracking_result.events[7].status == "in_transit" + assert tracking_result.events[8].status == "delivered" + + def test_delivered_on_first_try(self) -> None: + """DX-1091 - Test delivered on first try tracking event.""" + package_id = "pkg_1FedExDeLivered" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + track_package_assertions(tracking_result=tracking_result) + assert ( + tracking_result.shipment.actual_delivery_date.to_datetime_object() + == tracking_result.events[4].date_time.to_datetime_object() + ) + does_delivery_date_match(tracking_result) + assert_events_in_order(tracking_result.events) + assert len(tracking_result.events) == 5 + assert tracking_result.events[0].status == "accepted" + assert tracking_result.events[1].status == "in_transit" + assert tracking_result.events[4].status == "delivered" + assert tracking_result.events[-1].status == "delivered" + + def test_delivered_with_signature(self) -> None: + """DX-1092 - Test track delivered with signature event.""" + package_id = "pkg_1FedExDeLivered" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + track_package_assertions(tracking_result=tracking_result) + does_delivery_date_match(tracking_result) + assert_events_in_order(tracking_result.events) + assert len(tracking_result.events) == 5 + assert tracking_result.events[0].status == "accepted" + assert tracking_result.events[1].status == "in_transit" + assert tracking_result.events[2].status == "unknown" + assert tracking_result.events[3].status == "in_transit" + assert tracking_result.events[4].status == "delivered" + assert tracking_result.events[-1].status == "delivered" + assert tracking_result.events[-1].signer is not None + assert type(tracking_result.events[-1].signer) is str + + def test_delivered_after_multiple_attempts(self) -> None: + """DX-1093 - Test delivered after multiple attempts tracking event.""" + package_id = "pkg_1FedexDeLiveredException" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + assertions_on_delivered_after_exception_or_multiple_attempts(tracking_result) + + def test_delivered_after_exception(self) -> None: + """DX-1094 - Test delivered after exception tracking event.""" + package_id = "pkg_1FedexDeLiveredException" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + assertions_on_delivered_after_exception_or_multiple_attempts(tracking_result) + + def test_single_exception_tracking_event(self) -> None: + """DX-1095 - Test single exception tracking event.""" + package_id = "pkg_1FedexException" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + track_package_assertions(tracking_result=tracking_result) + assert_events_in_order(tracking_result.events) + assert len(tracking_result.events) == 3 + assert tracking_result.events[0].status == "accepted" + assert tracking_result.events[1].status == "in_transit" + assert tracking_result.events[2].status == "exception" + + def test_track_with_multiple_exceptions(self) -> None: + """DX-1096 - Test track with multiple exceptions.""" + package_id = "pkg_1FedexDeLiveredException" + shipengine = configurable_stub_shipengine_instance(stub_config()) + tracking_result = shipengine.track_package(tracking_data=package_id) + + track_package_assertions(tracking_result=tracking_result) + assert_events_in_order(tracking_result.events) + assert len(tracking_result.events) == 8 + assert tracking_result.events[0].status == "accepted" + assert tracking_result.events[4].status == "exception" + assert tracking_result.events[5].status == "exception" + assert tracking_result.events[7].status == "delivered" + assert tracking_result.events[-1].status == "delivered"