1# Copyright 2017, Google LLC 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15import calendar 16import datetime 17 18import pytest 19 20from google.api_core import datetime_helpers 21from google.protobuf import timestamp_pb2 22 23 24ONE_MINUTE_IN_MICROSECONDS = 60 * 1e6 25 26 27def test_utcnow(): 28 result = datetime_helpers.utcnow() 29 assert isinstance(result, datetime.datetime) 30 31 32def test_to_milliseconds(): 33 dt = datetime.datetime(1970, 1, 1, 0, 0, 1, tzinfo=datetime.timezone.utc) 34 assert datetime_helpers.to_milliseconds(dt) == 1000 35 36 37def test_to_microseconds(): 38 microseconds = 314159 39 dt = datetime.datetime(1970, 1, 1, 0, 0, 0, microsecond=microseconds) 40 assert datetime_helpers.to_microseconds(dt) == microseconds 41 42 43def test_to_microseconds_non_utc(): 44 zone = datetime.timezone(datetime.timedelta(minutes=-1)) 45 dt = datetime.datetime(1970, 1, 1, 0, 0, 0, tzinfo=zone) 46 assert datetime_helpers.to_microseconds(dt) == ONE_MINUTE_IN_MICROSECONDS 47 48 49def test_to_microseconds_naive(): 50 microseconds = 314159 51 dt = datetime.datetime(1970, 1, 1, 0, 0, 0, microsecond=microseconds, tzinfo=None) 52 assert datetime_helpers.to_microseconds(dt) == microseconds 53 54 55def test_from_microseconds(): 56 five_mins_from_epoch_in_microseconds = 5 * ONE_MINUTE_IN_MICROSECONDS 57 five_mins_from_epoch_datetime = datetime.datetime( 58 1970, 1, 1, 0, 5, 0, tzinfo=datetime.timezone.utc 59 ) 60 61 result = datetime_helpers.from_microseconds(five_mins_from_epoch_in_microseconds) 62 63 assert result == five_mins_from_epoch_datetime 64 65 66def test_from_iso8601_date(): 67 today = datetime.date.today() 68 iso_8601_today = today.strftime("%Y-%m-%d") 69 70 assert datetime_helpers.from_iso8601_date(iso_8601_today) == today 71 72 73def test_from_iso8601_time(): 74 assert datetime_helpers.from_iso8601_time("12:09:42") == datetime.time(12, 9, 42) 75 76 77def test_from_rfc3339(): 78 value = "2009-12-17T12:44:32.123456Z" 79 assert datetime_helpers.from_rfc3339(value) == datetime.datetime( 80 2009, 12, 17, 12, 44, 32, 123456, datetime.timezone.utc 81 ) 82 83 84def test_from_rfc3339_nanos(): 85 value = "2009-12-17T12:44:32.123456Z" 86 assert datetime_helpers.from_rfc3339_nanos(value) == datetime.datetime( 87 2009, 12, 17, 12, 44, 32, 123456, datetime.timezone.utc 88 ) 89 90 91def test_from_rfc3339_without_nanos(): 92 value = "2009-12-17T12:44:32Z" 93 assert datetime_helpers.from_rfc3339(value) == datetime.datetime( 94 2009, 12, 17, 12, 44, 32, 0, datetime.timezone.utc 95 ) 96 97 98def test_from_rfc3339_nanos_without_nanos(): 99 value = "2009-12-17T12:44:32Z" 100 assert datetime_helpers.from_rfc3339_nanos(value) == datetime.datetime( 101 2009, 12, 17, 12, 44, 32, 0, datetime.timezone.utc 102 ) 103 104 105@pytest.mark.parametrize( 106 "truncated, micros", 107 [ 108 ("12345678", 123456), 109 ("1234567", 123456), 110 ("123456", 123456), 111 ("12345", 123450), 112 ("1234", 123400), 113 ("123", 123000), 114 ("12", 120000), 115 ("1", 100000), 116 ], 117) 118def test_from_rfc3339_with_truncated_nanos(truncated, micros): 119 value = "2009-12-17T12:44:32.{}Z".format(truncated) 120 assert datetime_helpers.from_rfc3339(value) == datetime.datetime( 121 2009, 12, 17, 12, 44, 32, micros, datetime.timezone.utc 122 ) 123 124 125def test_from_rfc3339_nanos_is_deprecated(): 126 value = "2009-12-17T12:44:32.123456Z" 127 128 result = datetime_helpers.from_rfc3339(value) 129 result_nanos = datetime_helpers.from_rfc3339_nanos(value) 130 131 assert result == result_nanos 132 133 134@pytest.mark.parametrize( 135 "truncated, micros", 136 [ 137 ("12345678", 123456), 138 ("1234567", 123456), 139 ("123456", 123456), 140 ("12345", 123450), 141 ("1234", 123400), 142 ("123", 123000), 143 ("12", 120000), 144 ("1", 100000), 145 ], 146) 147def test_from_rfc3339_nanos_with_truncated_nanos(truncated, micros): 148 value = "2009-12-17T12:44:32.{}Z".format(truncated) 149 assert datetime_helpers.from_rfc3339_nanos(value) == datetime.datetime( 150 2009, 12, 17, 12, 44, 32, micros, datetime.timezone.utc 151 ) 152 153 154def test_from_rfc3339_wo_nanos_raise_exception(): 155 value = "2009-12-17T12:44:32" 156 with pytest.raises(ValueError): 157 datetime_helpers.from_rfc3339(value) 158 159 160def test_from_rfc3339_w_nanos_raise_exception(): 161 value = "2009-12-17T12:44:32.123456" 162 with pytest.raises(ValueError): 163 datetime_helpers.from_rfc3339(value) 164 165 166def test_to_rfc3339(): 167 value = datetime.datetime(2016, 4, 5, 13, 30, 0) 168 expected = "2016-04-05T13:30:00.000000Z" 169 assert datetime_helpers.to_rfc3339(value) == expected 170 171 172def test_to_rfc3339_with_utc(): 173 value = datetime.datetime(2016, 4, 5, 13, 30, 0, tzinfo=datetime.timezone.utc) 174 expected = "2016-04-05T13:30:00.000000Z" 175 assert datetime_helpers.to_rfc3339(value, ignore_zone=False) == expected 176 177 178def test_to_rfc3339_with_non_utc(): 179 zone = datetime.timezone(datetime.timedelta(minutes=-60)) 180 value = datetime.datetime(2016, 4, 5, 13, 30, 0, tzinfo=zone) 181 expected = "2016-04-05T14:30:00.000000Z" 182 assert datetime_helpers.to_rfc3339(value, ignore_zone=False) == expected 183 184 185def test_to_rfc3339_with_non_utc_ignore_zone(): 186 zone = datetime.timezone(datetime.timedelta(minutes=-60)) 187 value = datetime.datetime(2016, 4, 5, 13, 30, 0, tzinfo=zone) 188 expected = "2016-04-05T13:30:00.000000Z" 189 assert datetime_helpers.to_rfc3339(value, ignore_zone=True) == expected 190 191 192class Test_DateTimeWithNanos(object): 193 @staticmethod 194 def test_ctor_wo_nanos(): 195 stamp = datetime_helpers.DatetimeWithNanoseconds( 196 2016, 12, 20, 21, 13, 47, 123456 197 ) 198 assert stamp.year == 2016 199 assert stamp.month == 12 200 assert stamp.day == 20 201 assert stamp.hour == 21 202 assert stamp.minute == 13 203 assert stamp.second == 47 204 assert stamp.microsecond == 123456 205 assert stamp.nanosecond == 0 206 207 @staticmethod 208 def test_ctor_w_nanos(): 209 stamp = datetime_helpers.DatetimeWithNanoseconds( 210 2016, 12, 20, 21, 13, 47, nanosecond=123456789 211 ) 212 assert stamp.year == 2016 213 assert stamp.month == 12 214 assert stamp.day == 20 215 assert stamp.hour == 21 216 assert stamp.minute == 13 217 assert stamp.second == 47 218 assert stamp.microsecond == 123456 219 assert stamp.nanosecond == 123456789 220 221 @staticmethod 222 def test_ctor_w_micros_positional_and_nanos(): 223 with pytest.raises(TypeError): 224 datetime_helpers.DatetimeWithNanoseconds( 225 2016, 12, 20, 21, 13, 47, 123456, nanosecond=123456789 226 ) 227 228 @staticmethod 229 def test_ctor_w_micros_keyword_and_nanos(): 230 with pytest.raises(TypeError): 231 datetime_helpers.DatetimeWithNanoseconds( 232 2016, 12, 20, 21, 13, 47, microsecond=123456, nanosecond=123456789 233 ) 234 235 @staticmethod 236 def test_rfc3339_wo_nanos(): 237 stamp = datetime_helpers.DatetimeWithNanoseconds( 238 2016, 12, 20, 21, 13, 47, 123456 239 ) 240 assert stamp.rfc3339() == "2016-12-20T21:13:47.123456Z" 241 242 @staticmethod 243 def test_rfc3339_wo_nanos_w_leading_zero(): 244 stamp = datetime_helpers.DatetimeWithNanoseconds(2016, 12, 20, 21, 13, 47, 1234) 245 assert stamp.rfc3339() == "2016-12-20T21:13:47.001234Z" 246 247 @staticmethod 248 def test_rfc3339_w_nanos(): 249 stamp = datetime_helpers.DatetimeWithNanoseconds( 250 2016, 12, 20, 21, 13, 47, nanosecond=123456789 251 ) 252 assert stamp.rfc3339() == "2016-12-20T21:13:47.123456789Z" 253 254 @staticmethod 255 def test_rfc3339_w_nanos_w_leading_zero(): 256 stamp = datetime_helpers.DatetimeWithNanoseconds( 257 2016, 12, 20, 21, 13, 47, nanosecond=1234567 258 ) 259 assert stamp.rfc3339() == "2016-12-20T21:13:47.001234567Z" 260 261 @staticmethod 262 def test_rfc3339_w_nanos_no_trailing_zeroes(): 263 stamp = datetime_helpers.DatetimeWithNanoseconds( 264 2016, 12, 20, 21, 13, 47, nanosecond=100000000 265 ) 266 assert stamp.rfc3339() == "2016-12-20T21:13:47.1Z" 267 268 @staticmethod 269 def test_rfc3339_w_nanos_w_leading_zero_and_no_trailing_zeros(): 270 stamp = datetime_helpers.DatetimeWithNanoseconds( 271 2016, 12, 20, 21, 13, 47, nanosecond=1234500 272 ) 273 assert stamp.rfc3339() == "2016-12-20T21:13:47.0012345Z" 274 275 @staticmethod 276 def test_from_rfc3339_w_invalid(): 277 stamp = "2016-12-20T21:13:47" 278 with pytest.raises(ValueError): 279 datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(stamp) 280 281 @staticmethod 282 def test_from_rfc3339_wo_fraction(): 283 timestamp = "2016-12-20T21:13:47Z" 284 expected = datetime_helpers.DatetimeWithNanoseconds( 285 2016, 12, 20, 21, 13, 47, tzinfo=datetime.timezone.utc 286 ) 287 stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) 288 assert stamp == expected 289 290 @staticmethod 291 def test_from_rfc3339_w_partial_precision(): 292 timestamp = "2016-12-20T21:13:47.1Z" 293 expected = datetime_helpers.DatetimeWithNanoseconds( 294 2016, 12, 20, 21, 13, 47, microsecond=100000, tzinfo=datetime.timezone.utc 295 ) 296 stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) 297 assert stamp == expected 298 299 @staticmethod 300 def test_from_rfc3339_w_full_precision(): 301 timestamp = "2016-12-20T21:13:47.123456789Z" 302 expected = datetime_helpers.DatetimeWithNanoseconds( 303 2016, 12, 20, 21, 13, 47, nanosecond=123456789, tzinfo=datetime.timezone.utc 304 ) 305 stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp) 306 assert stamp == expected 307 308 @staticmethod 309 @pytest.mark.parametrize( 310 "fractional, nanos", 311 [ 312 ("12345678", 123456780), 313 ("1234567", 123456700), 314 ("123456", 123456000), 315 ("12345", 123450000), 316 ("1234", 123400000), 317 ("123", 123000000), 318 ("12", 120000000), 319 ("1", 100000000), 320 ], 321 ) 322 def test_from_rfc3339_test_nanoseconds(fractional, nanos): 323 value = "2009-12-17T12:44:32.{}Z".format(fractional) 324 assert ( 325 datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(value).nanosecond 326 == nanos 327 ) 328 329 @staticmethod 330 def test_timestamp_pb_wo_nanos_naive(): 331 stamp = datetime_helpers.DatetimeWithNanoseconds( 332 2016, 12, 20, 21, 13, 47, 123456 333 ) 334 delta = ( 335 stamp.replace(tzinfo=datetime.timezone.utc) - datetime_helpers._UTC_EPOCH 336 ) 337 seconds = int(delta.total_seconds()) 338 nanos = 123456000 339 timestamp = timestamp_pb2.Timestamp(seconds=seconds, nanos=nanos) 340 assert stamp.timestamp_pb() == timestamp 341 342 @staticmethod 343 def test_timestamp_pb_w_nanos(): 344 stamp = datetime_helpers.DatetimeWithNanoseconds( 345 2016, 12, 20, 21, 13, 47, nanosecond=123456789, tzinfo=datetime.timezone.utc 346 ) 347 delta = stamp - datetime_helpers._UTC_EPOCH 348 timestamp = timestamp_pb2.Timestamp( 349 seconds=int(delta.total_seconds()), nanos=123456789 350 ) 351 assert stamp.timestamp_pb() == timestamp 352 353 @staticmethod 354 def test_from_timestamp_pb_wo_nanos(): 355 when = datetime.datetime( 356 2016, 12, 20, 21, 13, 47, 123456, tzinfo=datetime.timezone.utc 357 ) 358 delta = when - datetime_helpers._UTC_EPOCH 359 seconds = int(delta.total_seconds()) 360 timestamp = timestamp_pb2.Timestamp(seconds=seconds) 361 362 stamp = datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb(timestamp) 363 364 assert _to_seconds(when) == _to_seconds(stamp) 365 assert stamp.microsecond == 0 366 assert stamp.nanosecond == 0 367 assert stamp.tzinfo == datetime.timezone.utc 368 369 @staticmethod 370 def test_from_timestamp_pb_w_nanos(): 371 when = datetime.datetime( 372 2016, 12, 20, 21, 13, 47, 123456, tzinfo=datetime.timezone.utc 373 ) 374 delta = when - datetime_helpers._UTC_EPOCH 375 seconds = int(delta.total_seconds()) 376 timestamp = timestamp_pb2.Timestamp(seconds=seconds, nanos=123456789) 377 378 stamp = datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb(timestamp) 379 380 assert _to_seconds(when) == _to_seconds(stamp) 381 assert stamp.microsecond == 123456 382 assert stamp.nanosecond == 123456789 383 assert stamp.tzinfo == datetime.timezone.utc 384 385 386def _to_seconds(value): 387 """Convert a datetime to seconds since the unix epoch. 388 389 Args: 390 value (datetime.datetime): The datetime to covert. 391 392 Returns: 393 int: Microseconds since the unix epoch. 394 """ 395 assert value.tzinfo is datetime.timezone.utc 396 return calendar.timegm(value.timetuple()) 397