| # Copyright 2017, Google LLC |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # 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 |
| |
| |
| def test_utcnow(): |
| result = datetime_helpers.utcnow() |
| assert isinstance(result, datetime.datetime) |
| |
| |
| def test_to_milliseconds(): |
| dt = datetime.datetime(1970, 1, 1, 0, 0, 1, tzinfo=pytz.utc) |
| assert datetime_helpers.to_milliseconds(dt) == 1000 |
| |
| |
| def test_to_microseconds(): |
| microseconds = 314159 |
| dt = datetime.datetime(1970, 1, 1, 0, 0, 0, microsecond=microseconds) |
| assert datetime_helpers.to_microseconds(dt) == microseconds |
| |
| |
| def test_to_microseconds_non_utc(): |
| zone = pytz.FixedOffset(-1) |
| dt = datetime.datetime(1970, 1, 1, 0, 0, 0, tzinfo=zone) |
| assert datetime_helpers.to_microseconds(dt) == ONE_MINUTE_IN_MICROSECONDS |
| |
| |
| def test_to_microseconds_naive(): |
| microseconds = 314159 |
| dt = datetime.datetime(1970, 1, 1, 0, 0, 0, microsecond=microseconds, tzinfo=None) |
| assert datetime_helpers.to_microseconds(dt) == microseconds |
| |
| |
| def test_from_microseconds(): |
| five_mins_from_epoch_in_microseconds = 5 * ONE_MINUTE_IN_MICROSECONDS |
| five_mins_from_epoch_datetime = datetime.datetime( |
| 1970, 1, 1, 0, 5, 0, tzinfo=pytz.utc |
| ) |
| |
| result = datetime_helpers.from_microseconds(five_mins_from_epoch_in_microseconds) |
| |
| assert result == five_mins_from_epoch_datetime |
| |
| |
| def test_from_iso8601_date(): |
| today = datetime.date.today() |
| iso_8601_today = today.strftime("%Y-%m-%d") |
| |
| assert datetime_helpers.from_iso8601_date(iso_8601_today) == today |
| |
| |
| def test_from_iso8601_time(): |
| assert datetime_helpers.from_iso8601_time("12:09:42") == datetime.time(12, 9, 42) |
| |
| |
| def test_from_rfc3339(): |
| value = "2009-12-17T12:44:32.123456Z" |
| assert datetime_helpers.from_rfc3339(value) == datetime.datetime( |
| 2009, 12, 17, 12, 44, 32, 123456, pytz.utc |
| ) |
| |
| |
| def test_from_rfc3339_nanos(): |
| value = "2009-12-17T12:44:32.123456Z" |
| assert datetime_helpers.from_rfc3339_nanos(value) == datetime.datetime( |
| 2009, 12, 17, 12, 44, 32, 123456, pytz.utc |
| ) |
| |
| |
| def test_from_rfc3339_without_nanos(): |
| value = "2009-12-17T12:44:32Z" |
| assert datetime_helpers.from_rfc3339(value) == datetime.datetime( |
| 2009, 12, 17, 12, 44, 32, 0, pytz.utc |
| ) |
| |
| |
| def test_from_rfc3339_nanos_without_nanos(): |
| value = "2009-12-17T12:44:32Z" |
| assert datetime_helpers.from_rfc3339_nanos(value) == datetime.datetime( |
| 2009, 12, 17, 12, 44, 32, 0, pytz.utc |
| ) |
| |
| |
| @pytest.mark.parametrize( |
| "truncated, micros", |
| [ |
| ("12345678", 123456), |
| ("1234567", 123456), |
| ("123456", 123456), |
| ("12345", 123450), |
| ("1234", 123400), |
| ("123", 123000), |
| ("12", 120000), |
| ("1", 100000), |
| ], |
| ) |
| def test_from_rfc3339_with_truncated_nanos(truncated, micros): |
| value = "2009-12-17T12:44:32.{}Z".format(truncated) |
| assert datetime_helpers.from_rfc3339(value) == datetime.datetime( |
| 2009, 12, 17, 12, 44, 32, micros, pytz.utc |
| ) |
| |
| |
| def test_from_rfc3339_nanos_is_deprecated(): |
| value = "2009-12-17T12:44:32.123456Z" |
| |
| result = datetime_helpers.from_rfc3339(value) |
| result_nanos = datetime_helpers.from_rfc3339_nanos(value) |
| |
| assert result == result_nanos |
| |
| |
| @pytest.mark.parametrize( |
| "truncated, micros", |
| [ |
| ("12345678", 123456), |
| ("1234567", 123456), |
| ("123456", 123456), |
| ("12345", 123450), |
| ("1234", 123400), |
| ("123", 123000), |
| ("12", 120000), |
| ("1", 100000), |
| ], |
| ) |
| def test_from_rfc3339_nanos_with_truncated_nanos(truncated, micros): |
| value = "2009-12-17T12:44:32.{}Z".format(truncated) |
| assert datetime_helpers.from_rfc3339_nanos(value) == datetime.datetime( |
| 2009, 12, 17, 12, 44, 32, micros, pytz.utc |
| ) |
| |
| |
| def test_from_rfc3339_wo_nanos_raise_exception(): |
| value = "2009-12-17T12:44:32" |
| with pytest.raises(ValueError): |
| datetime_helpers.from_rfc3339(value) |
| |
| |
| def test_from_rfc3339_w_nanos_raise_exception(): |
| value = "2009-12-17T12:44:32.123456" |
| with pytest.raises(ValueError): |
| datetime_helpers.from_rfc3339(value) |
| |
| |
| def test_to_rfc3339(): |
| value = datetime.datetime(2016, 4, 5, 13, 30, 0) |
| expected = "2016-04-05T13:30:00.000000Z" |
| assert datetime_helpers.to_rfc3339(value) == expected |
| |
| |
| def test_to_rfc3339_with_utc(): |
| value = datetime.datetime(2016, 4, 5, 13, 30, 0, tzinfo=pytz.utc) |
| expected = "2016-04-05T13:30:00.000000Z" |
| assert datetime_helpers.to_rfc3339(value, ignore_zone=False) == expected |
| |
| |
| def test_to_rfc3339_with_non_utc(): |
| zone = pytz.FixedOffset(-60) |
| value = datetime.datetime(2016, 4, 5, 13, 30, 0, tzinfo=zone) |
| expected = "2016-04-05T14:30:00.000000Z" |
| assert datetime_helpers.to_rfc3339(value, ignore_zone=False) == expected |
| |
| |
| def test_to_rfc3339_with_non_utc_ignore_zone(): |
| zone = pytz.FixedOffset(-60) |
| value = datetime.datetime(2016, 4, 5, 13, 30, 0, tzinfo=zone) |
| expected = "2016-04-05T13:30:00.000000Z" |
| assert datetime_helpers.to_rfc3339(value, ignore_zone=True) == expected |
| |
| |
| 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 |
| ) |
| 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_wo_nanos_w_leading_zero(): |
| stamp = datetime_helpers.DatetimeWithNanoseconds(2016, 12, 20, 21, 13, 47, 1234) |
| assert stamp.rfc3339() == "2016-12-20T21:13:47.001234Z" |
| |
| @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" |
| |
| @staticmethod |
| def test_rfc3339_w_nanos_w_leading_zero(): |
| stamp = datetime_helpers.DatetimeWithNanoseconds( |
| 2016, 12, 20, 21, 13, 47, nanosecond=1234567 |
| ) |
| assert stamp.rfc3339() == "2016-12-20T21:13:47.001234567Z" |
| |
| @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_rfc3339_w_nanos_w_leading_zero_and_no_trailing_zeros(): |
| stamp = datetime_helpers.DatetimeWithNanoseconds( |
| 2016, 12, 20, 21, 13, 47, nanosecond=1234500 |
| ) |
| assert stamp.rfc3339() == "2016-12-20T21:13:47.0012345Z" |
| |
| @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 |
| @pytest.mark.parametrize( |
| "fractional, nanos", |
| [ |
| ("12345678", 123456780), |
| ("1234567", 123456700), |
| ("123456", 123456000), |
| ("12345", 123450000), |
| ("1234", 123400000), |
| ("123", 123000000), |
| ("12", 120000000), |
| ("1", 100000000), |
| ], |
| ) |
| def test_from_rfc3339_test_nanoseconds(fractional, nanos): |
| value = "2009-12-17T12:44:32.{}Z".format(fractional) |
| assert ( |
| datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(value).nanosecond |
| == nanos |
| ) |
| |
| @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()) |