DX-1086-1096 track package tests

shipengine_sdk/jsonrpc/__init__.py

@@ -42,4 +42,4 @@ def rpc_request_loop(
             else:
                 raise err
         retry += 1
-    return api_response
+        return api_response

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,23 +132,17 @@ 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
         self.carrier_status_code = (
             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)

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):

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"