diff options
| -rw-r--r-- | google/api_core/datetime_helpers.py | 38 | ||||
| -rw-r--r-- | tests/unit/test_datetime_helpers.py | 242 |
2 files changed, 193 insertions, 87 deletions
diff --git a/google/api_core/datetime_helpers.py b/google/api_core/datetime_helpers.py index 3f3523b..b0d9105 100644 --- a/google/api_core/datetime_helpers.py +++ b/google/api_core/datetime_helpers.py @@ -20,6 +20,8 @@ import re import pytz +from google.protobuf import timestamp_pb2 + _UTC_EPOCH = datetime.datetime.utcfromtimestamp(0).replace(tzinfo=pytz.utc) _RFC3339_MICROS = "%Y-%m-%dT%H:%M:%S.%fZ" @@ -263,3 +265,39 @@ class DatetimeWithNanoseconds(datetime.datetime): nanosecond=nanos, tzinfo=pytz.UTC, ) + + def timestamp_pb(self): + """Return a timestamp message. + + Returns: + (:class:`~google.protobuf.timestamp_pb2.Timestamp`): Timestamp message + """ + inst = self if self.tzinfo is not None else self.replace(tzinfo=pytz.UTC) + delta = inst - _UTC_EPOCH + seconds = int(delta.total_seconds()) + nanos = self._nanosecond or self.microsecond * 1000 + return timestamp_pb2.Timestamp(seconds=seconds, nanos=nanos) + + @classmethod + def from_timestamp_pb(cls, stamp): + """Parse RFC 3339-compliant timestamp, preserving nanoseconds. + + Args: + stamp (:class:`~google.protobuf.timestamp_pb2.Timestamp`): timestamp message + + Returns: + :class:`DatetimeWithNanoseconds`: + an instance matching the timestamp message + """ + microseconds = int(stamp.seconds * 1e6) + bare = from_microseconds(microseconds) + return cls( + bare.year, + bare.month, + bare.day, + bare.hour, + bare.minute, + bare.second, + nanosecond=stamp.nanos, + tzinfo=pytz.UTC, + ) diff --git a/tests/unit/test_datetime_helpers.py b/tests/unit/test_datetime_helpers.py index 2f99235..e5220ae 100644 --- a/tests/unit/test_datetime_helpers.py +++ b/tests/unit/test_datetime_helpers.py @@ -12,12 +12,14 @@ # See the License for the specific language governing permissions and # limitations under the License. +import calendar import datetime import pytest import pytz from google.api_core import datetime_helpers +from google.protobuf import timestamp_pb2 ONE_MINUTE_IN_MICROSECONDS = 60 * 1e6 @@ -154,93 +156,159 @@ def test_to_rfc3339_with_non_utc_ignore_zone(): assert datetime_helpers.to_rfc3339(value, ignore_zone=True) == expected -def test_datetimewithnanos_ctor_wo_nanos(): - stamp = datetime_helpers.DatetimeWithNanoseconds(2016, 12, 20, 21, 13, 47, 123456) - assert stamp.year == 2016 - assert stamp.month == 12 - assert stamp.day == 20 - assert stamp.hour == 21 - assert stamp.minute == 13 - assert stamp.second == 47 - assert stamp.microsecond == 123456 - assert stamp.nanosecond == 0 - - -def test_datetimewithnanos_ctor_w_nanos(): - stamp = datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, nanosecond=123456789 - ) - assert stamp.year == 2016 - assert stamp.month == 12 - assert stamp.day == 20 - assert stamp.hour == 21 - assert stamp.minute == 13 - assert stamp.second == 47 - assert stamp.microsecond == 123456 - assert stamp.nanosecond == 123456789 - - -def test_datetimewithnanos_ctor_w_micros_positional_and_nanos(): - with pytest.raises(TypeError): - datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, 123456, nanosecond=123456789 +class Test_DateTimeWithNanos(object): + + @staticmethod + def test_ctor_wo_nanos(): + stamp = datetime_helpers.DatetimeWithNanoseconds(2016, 12, 20, 21, 13, 47, 123456) + assert stamp.year == 2016 + assert stamp.month == 12 + assert stamp.day == 20 + assert stamp.hour == 21 + assert stamp.minute == 13 + assert stamp.second == 47 + assert stamp.microsecond == 123456 + assert stamp.nanosecond == 0 + + @staticmethod + def test_ctor_w_nanos(): + stamp = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, nanosecond=123456789 ) - - -def test_datetimewithnanos_ctor_w_micros_keyword_and_nanos(): - with pytest.raises(TypeError): - datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, microsecond=123456, nanosecond=123456789 + assert stamp.year == 2016 + assert stamp.month == 12 + assert stamp.day == 20 + assert stamp.hour == 21 + assert stamp.minute == 13 + assert stamp.second == 47 + assert stamp.microsecond == 123456 + assert stamp.nanosecond == 123456789 + + @staticmethod + def test_ctor_w_micros_positional_and_nanos(): + with pytest.raises(TypeError): + datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, 123456, nanosecond=123456789 + ) + + @staticmethod + def test_ctor_w_micros_keyword_and_nanos(): + with pytest.raises(TypeError): + datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, microsecond=123456, nanosecond=123456789 + ) + + @staticmethod + def test_rfc3339_wo_nanos(): + stamp = datetime_helpers.DatetimeWithNanoseconds(2016, 12, 20, 21, 13, 47, 123456) + assert stamp.rfc3339() == "2016-12-20T21:13:47.123456Z" + + @staticmethod + def test_rfc3339_w_nanos(): + stamp = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, nanosecond=123456789 ) + assert stamp.rfc3339() == "2016-12-20T21:13:47.123456789Z" - -def test_datetimewithnanos_rfc339_wo_nanos(): - stamp = datetime_helpers.DatetimeWithNanoseconds(2016, 12, 20, 21, 13, 47, 123456) - assert stamp.rfc3339() == "2016-12-20T21:13:47.123456Z" - - -def test_datetimewithnanos_rfc339_w_nanos(): - stamp = datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, nanosecond=123456789 - ) - assert stamp.rfc3339() == "2016-12-20T21:13:47.123456789Z" - - -def test_datetimewithnanos_rfc339_w_nanos_no_trailing_zeroes(): - stamp = datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, nanosecond=100000000 - ) - assert stamp.rfc3339() == "2016-12-20T21:13:47.1Z" - - -def test_datetimewithnanos_from_rfc3339_w_invalid(): - stamp = "2016-12-20T21:13:47" - with pytest.raises(ValueError): - datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(stamp) - - -def test_datetimewithnanos_from_rfc3339_wo_fraction(): - timestamp = "2016-12-20T21:13:47Z" - expected = datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, tzinfo=pytz.UTC - ) - stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) - assert stamp == expected - - -def test_datetimewithnanos_from_rfc3339_w_partial_precision(): - timestamp = "2016-12-20T21:13:47.1Z" - expected = datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, microsecond=100000, tzinfo=pytz.UTC - ) - stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) - assert stamp == expected - - -def test_datetimewithnanos_from_rfc3339_w_full_precision(): - timestamp = "2016-12-20T21:13:47.123456789Z" - expected = datetime_helpers.DatetimeWithNanoseconds( - 2016, 12, 20, 21, 13, 47, nanosecond=123456789, tzinfo=pytz.UTC - ) - stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) - assert stamp == expected + @staticmethod + def test_rfc3339_w_nanos_no_trailing_zeroes(): + stamp = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, nanosecond=100000000 + ) + assert stamp.rfc3339() == "2016-12-20T21:13:47.1Z" + + @staticmethod + def test_from_rfc3339_w_invalid(): + stamp = "2016-12-20T21:13:47" + with pytest.raises(ValueError): + datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(stamp) + + @staticmethod + def test_from_rfc3339_wo_fraction(): + timestamp = "2016-12-20T21:13:47Z" + expected = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, tzinfo=pytz.UTC + ) + stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) + assert stamp == expected + + @staticmethod + def test_from_rfc3339_w_partial_precision(): + timestamp = "2016-12-20T21:13:47.1Z" + expected = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, microsecond=100000, tzinfo=pytz.UTC + ) + stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) + assert stamp == expected + + @staticmethod + def test_from_rfc3339_w_full_precision(): + timestamp = "2016-12-20T21:13:47.123456789Z" + expected = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, nanosecond=123456789, tzinfo=pytz.UTC + ) + stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) + assert stamp == expected + + @staticmethod + def test_timestamp_pb_wo_nanos_naive(): + stamp = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, 123456) + delta = stamp.replace(tzinfo=pytz.UTC) - datetime_helpers._UTC_EPOCH + seconds = int(delta.total_seconds()) + nanos = 123456000 + timestamp = timestamp_pb2.Timestamp(seconds=seconds, nanos=nanos) + assert stamp.timestamp_pb() == timestamp + + @staticmethod + def test_timestamp_pb_w_nanos(): + stamp = datetime_helpers.DatetimeWithNanoseconds( + 2016, 12, 20, 21, 13, 47, nanosecond=123456789, tzinfo=pytz.UTC + ) + delta = stamp - datetime_helpers._UTC_EPOCH + timestamp = timestamp_pb2.Timestamp( + seconds=int(delta.total_seconds()), nanos=123456789) + assert stamp.timestamp_pb() == timestamp + + @staticmethod + def test_from_timestamp_pb_wo_nanos(): + when = datetime.datetime(2016, 12, 20, 21, 13, 47, 123456, tzinfo=pytz.UTC) + delta = when - datetime_helpers._UTC_EPOCH + seconds = int(delta.total_seconds()) + timestamp = timestamp_pb2.Timestamp(seconds=seconds) + + stamp = datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb( + timestamp) + + assert _to_seconds(when) == _to_seconds(stamp) + assert stamp.microsecond == 0 + assert stamp.nanosecond == 0 + assert stamp.tzinfo == pytz.UTC + + @staticmethod + def test_from_timestamp_pb_w_nanos(): + when = datetime.datetime(2016, 12, 20, 21, 13, 47, 123456, tzinfo=pytz.UTC) + delta = when - datetime_helpers._UTC_EPOCH + seconds = int(delta.total_seconds()) + timestamp = timestamp_pb2.Timestamp(seconds=seconds, nanos=123456789) + + stamp = datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb( + timestamp) + + assert _to_seconds(when) == _to_seconds(stamp) + assert stamp.microsecond == 123456 + assert stamp.nanosecond == 123456789 + assert stamp.tzinfo == pytz.UTC + + +def _to_seconds(value): + """Convert a datetime to seconds since the unix epoch. + + Args: + value (datetime.datetime): The datetime to covert. + + Returns: + int: Microseconds since the unix epoch. + """ + assert value.tzinfo is pytz.UTC + return calendar.timegm(value.timetuple()) |