fix: handle timestamps with no fractional seconds (#149)

Co-authored-by: Daniel Lee <[email protected]>

Author: DKbyo

src/firebase_functions/firestore_fn.py

@@ -112,12 +112,7 @@ def _firestore_endpoint_handler(
     event_database = event_attributes["database"]
 
     time = event_attributes["time"]
-    is_nanoseconds = _util.is_precision_timestamp(time)
-
-    if is_nanoseconds:
-        event_time = _util.nanoseconds_timestamp_conversion(time)
-    else:
-        event_time = _util.microsecond_timestamp_conversion(time)
+    event_time = _util.timestamp_conversion(time)
 
     if _DEFAULT_APP_NAME not in _apps:
         initialize_app()

src/firebase_functions/private/_alerts_fn.py

@@ -15,8 +15,8 @@
 
 # pylint: disable=protected-access,cyclic-import
 import typing as _typing
-import datetime as _dt
 import cloudevents.http as _ce
+import util as _util
 from firebase_functions.alerts import FirebaseAlertData
 
 from functions_framework import logging as _logging
@@ -102,14 +102,10 @@ def new_nonfatal_issue_payload_from_ce_payload(payload: dict):
 
 def regression_alert_payload_from_ce_payload(payload: dict):
     from firebase_functions.alerts.crashlytics_fn import RegressionAlertPayload
-    return RegressionAlertPayload(
-        type=payload["type"],
-        issue=issue_from_ce_payload(payload["issue"]),
-        resolve_time=_dt.datetime.strptime(
-            payload["resolveTime"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ),
-    )
+    return RegressionAlertPayload(type=payload["type"],
+                                  issue=issue_from_ce_payload(payload["issue"]),
+                                  resolve_time=_util.timestamp_conversion(
+                                      payload["resolveTime"]))
 
 
 def trending_issue_details_from_ce_payload(payload: dict):
@@ -125,10 +121,7 @@ def trending_issue_details_from_ce_payload(payload: dict):
 def stability_digest_payload_from_ce_payload(payload: dict):
     from firebase_functions.alerts.crashlytics_fn import StabilityDigestPayload
     return StabilityDigestPayload(
-        digest_date=_dt.datetime.strptime(
-            payload["digestDate"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ),
+        digest_date=_util.timestamp_conversion(payload["digestDate"]),
         trending_issues=[
             trending_issue_details_from_ce_payload(issue)
             for issue in payload["trendingIssues"]
@@ -139,10 +132,7 @@ def velocity_alert_payload_from_ce_payload(payload: dict):
     from firebase_functions.alerts.crashlytics_fn import VelocityAlertPayload
     return VelocityAlertPayload(
         issue=issue_from_ce_payload(payload["issue"]),
-        create_time=_dt.datetime.strptime(
-            payload["createTime"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ),
+        create_time=_util.timestamp_conversion(payload["createTime"]),
         crash_count=payload["crashCount"],
         crash_percentage=payload["crashPercentage"],
         first_version=payload["firstVersion"],
@@ -186,14 +176,9 @@ def firebase_alert_data_from_ce(event_dict: dict,) -> FirebaseAlertData:
         _logging.warning(f"Unhandled Firebase Alerts alert type: {alert_type}")
 
     return FirebaseAlertData(
-        create_time=_dt.datetime.strptime(
-            event_dict["createTime"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ),
-        end_time=_dt.datetime.strptime(
-            event_dict["endTime"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ) if "endTime" in event_dict else None,
+        create_time=_util.timestamp_conversion(event_dict["createTime"]),
+        end_time=_util.timestamp_conversion(event_dict["endTime"])
+        if "endTime" in event_dict else None,
         payload=alert_payload,
     )
 
@@ -204,25 +189,14 @@ def event_from_ce_helper(raw: _ce.CloudEvent, cls, app_id=True):
     event_dict = {**event_data, **event_attributes}
     alert_type: str = event_dict["alerttype"]
     event_kwargs = {
-        "alert_type":
-            alert_type,
-        "data":
-            firebase_alert_data_from_ce(event_dict),
-        "id":
-            event_dict["id"],
-        "source":
-            event_dict["source"],
-        "specversion":
-            event_dict["specversion"],
-        "subject":
-            event_dict["subject"] if "subject" in event_dict else None,
-        "time":
-            _dt.datetime.strptime(
-                event_dict["time"],
-                "%Y-%m-%dT%H:%M:%S.%f%z",
-            ),
-        "type":
-            event_dict["type"],
+        "alert_type": alert_type,
+        "data": firebase_alert_data_from_ce(event_dict),
+        "id": event_dict["id"],
+        "source": event_dict["source"],
+        "specversion": event_dict["specversion"],
+        "subject": event_dict["subject"] if "subject" in event_dict else None,
+        "time": _util.timestamp_conversion(event_dict["time"]),
+        "type": event_dict["type"],
     }
     if app_id:
         event_kwargs["app_id"] = event_dict.get("appid")

src/firebase_functions/private/util.py

@@ -337,20 +337,55 @@ def nanoseconds_timestamp_conversion(time: str) -> _dt.datetime:
     return event_time
 
 
-def is_precision_timestamp(time: str) -> bool:
+def second_timestamp_conversion(time: str) -> _dt.datetime:
+    """Converts a second timestamp and returns a datetime object of the current time in UTC"""
+    return _dt.datetime.strptime(
+        time,
+        "%Y-%m-%dT%H:%M:%S%z",
+    )
+
+
+class PrecisionTimestamp(_enum.Enum):
+    """
+    The status of a token.
+    """
+
+    NANOSECONDS = "NANOSECONDS"
+
+    MICROSECONDS = "MICROSECONDS"
+
+    SECONDS = "SECONDS"
+
+
+def get_precision_timestamp(time: str) -> PrecisionTimestamp:
     """Return a bool which indicates if the timestamp is in nanoseconds"""
     # Split the string into date-time and fraction of second
     try:
         _, s_fraction = time.split(".")
     except ValueError:
-        return False  # If there's no decimal, it's not a nanosecond timestamp.
+        return PrecisionTimestamp.SECONDS
 
     # Split the fraction from the timezone specifier ('Z' or 'z')
     s_fraction, _ = s_fraction.split(
         "Z") if "Z" in s_fraction else s_fraction.split("z")
 
     # If the fraction is more than 6 digits long, it's a nanosecond timestamp
-    return len(s_fraction) > 6
+    if len(s_fraction) > 6:
+        return PrecisionTimestamp.NANOSECONDS
+    else:
+        return PrecisionTimestamp.MICROSECONDS
+
+
+def timestamp_conversion(time: str) -> _dt.datetime:
+    """Converts a timestamp and returns a datetime object of the current time in UTC"""
+    precision_timestamp = get_precision_timestamp(time)
+
+    if precision_timestamp == PrecisionTimestamp.NANOSECONDS:
+        return nanoseconds_timestamp_conversion(time)
+    elif precision_timestamp == PrecisionTimestamp.MICROSECONDS:
+        return microsecond_timestamp_conversion(time)
+    elif precision_timestamp == PrecisionTimestamp.SECONDS:
+        return second_timestamp_conversion(time)
 
 
 def microsecond_timestamp_conversion(time: str) -> _dt.datetime:

tests/test_util.py

@@ -15,7 +15,7 @@
 Internal utils tests.
 """
 from os import environ, path
-from firebase_functions.private.util import firebase_config, microsecond_timestamp_conversion, nanoseconds_timestamp_conversion, is_precision_timestamp, normalize_path, deep_merge
+from firebase_functions.private.util import firebase_config, microsecond_timestamp_conversion, nanoseconds_timestamp_conversion, get_precision_timestamp, normalize_path, deep_merge, PrecisionTimestamp, second_timestamp_conversion
 import datetime as _dt
 
 test_bucket = "python-functions-testing.appspot.com"
@@ -81,6 +81,24 @@ def test_nanosecond_conversion():
             input_timestamp) == expected_datetime
 
 
+def test_second_conversion():
+    """
+    Testing seconds_timestamp_conversion works as intended
+    """
+    timestamps = [
+        ("2023-01-01T12:34:56Z", "2023-01-01T12:34:56Z"),
+        ("2023-02-14T14:37:52Z", "2023-02-14T14:37:52Z"),
+        ("2023-03-21T06:43:58Z", "2023-03-21T06:43:58Z"),
+        ("2023-10-06T07:00:00Z", "2023-10-06T07:00:00Z"),
+    ]
+
+    for input_timestamp, expected_output in timestamps:
+        expected_datetime = _dt.datetime.strptime(expected_output,
+                                                  "%Y-%m-%dT%H:%M:%SZ")
+        expected_datetime = expected_datetime.replace(tzinfo=_dt.timezone.utc)
+        assert second_timestamp_conversion(input_timestamp) == expected_datetime
+
+
 def test_is_nanoseconds_timestamp():
     """
     Testing is_nanoseconds_timestamp works as intended
@@ -95,19 +113,40 @@ def test_is_nanoseconds_timestamp():
     nanosecond_timestamp3 = "2023-03-21T06:43:58.564738291Z"
     nanosecond_timestamp4 = "2023-08-15T22:22:22.222222222Z"
 
-    assert is_precision_timestamp(microsecond_timestamp1) is False
-    assert is_precision_timestamp(microsecond_timestamp2) is False
-    assert is_precision_timestamp(microsecond_timestamp3) is False
-    assert is_precision_timestamp(microsecond_timestamp4) is False
-    assert is_precision_timestamp(nanosecond_timestamp1) is True
-    assert is_precision_timestamp(nanosecond_timestamp2) is True
-    assert is_precision_timestamp(nanosecond_timestamp3) is True
-    assert is_precision_timestamp(nanosecond_timestamp4) is True
+    second_timestamp1 = "2023-01-01T12:34:56Z"
+    second_timestamp2 = "2023-02-14T14:37:52Z"
+    second_timestamp3 = "2023-03-21T06:43:58Z"
+    second_timestamp4 = "2023-08-15T22:22:22Z"
+
+    assert get_precision_timestamp(
+        microsecond_timestamp1) is PrecisionTimestamp.MICROSECONDS
+    assert get_precision_timestamp(
+        microsecond_timestamp2) is PrecisionTimestamp.MICROSECONDS
+    assert get_precision_timestamp(
+        microsecond_timestamp3) is PrecisionTimestamp.MICROSECONDS
+    assert get_precision_timestamp(
+        microsecond_timestamp4) is PrecisionTimestamp.MICROSECONDS
+    assert get_precision_timestamp(
+        nanosecond_timestamp1) is PrecisionTimestamp.NANOSECONDS
+    assert get_precision_timestamp(
+        nanosecond_timestamp2) is PrecisionTimestamp.NANOSECONDS
+    assert get_precision_timestamp(
+        nanosecond_timestamp3) is PrecisionTimestamp.NANOSECONDS
+    assert get_precision_timestamp(
+        nanosecond_timestamp4) is PrecisionTimestamp.NANOSECONDS
+    assert get_precision_timestamp(
+        second_timestamp1) is PrecisionTimestamp.SECONDS
+    assert get_precision_timestamp(
+        second_timestamp2) is PrecisionTimestamp.SECONDS
+    assert get_precision_timestamp(
+        second_timestamp3) is PrecisionTimestamp.SECONDS
+    assert get_precision_timestamp(
+        second_timestamp4) is PrecisionTimestamp.SECONDS
 
 
 def test_normalize_document_path():
     """
-    Testing "document" path passed to Firestore event listener 
+    Testing "document" path passed to Firestore event listener
     is normalized.
     """
     test_path = "/test/document/"