Fix _Timestamp edge cases (#534)

* Add failing test cases for timestamp conversion

* Fix timestamp to datetime conversion

* Fix formatting

* Move timestamp tests outside of inputs folder

---------

Co-authored-by: Lukas Bindreiter <lukas.bindreiter@tilebox.io>
Co-authored-by: James Hilton-Balfe <gobot1234yt@gmail.com>
This commit is contained in:
Lukas Bindreiter 2023-10-18 01:41:09 +02:00 committed by GitHub
parent 1dd001b6d3
commit 02aa4e88b7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 43 additions and 7 deletions

View File

@ -1908,15 +1908,24 @@ class _Duration(Duration):
class _Timestamp(Timestamp): class _Timestamp(Timestamp):
@classmethod @classmethod
def from_datetime(cls, dt: datetime) -> "_Timestamp": def from_datetime(cls, dt: datetime) -> "_Timestamp":
# apparently 0 isn't a year in [0, 9999]?? # manual epoch offset calulation to avoid rounding errors,
seconds = int((dt - DATETIME_ZERO).total_seconds()) # to support negative timestamps (before 1970) and skirt
nanos = int(dt.microsecond * 1e3) # around datetime bugs (apparently 0 isn't a year in [0, 9999]??)
return cls(seconds, nanos) offset = dt - DATETIME_ZERO
# below is the same as timedelta.total_seconds() but without dividing by 1e6
# so we end up with microseconds as integers instead of seconds as float
offset_us = (
offset.days * 24 * 60 * 60 + offset.seconds
) * 10**6 + offset.microseconds
seconds, us = divmod(offset_us, 10**6)
return cls(seconds, us * 1000)
def to_datetime(self) -> datetime: def to_datetime(self) -> datetime:
ts = self.seconds + (self.nanos / 1e9) # datetime.fromtimestamp() expects a timestamp in seconds, not microseconds
# if datetime.fromtimestamp ever supports -62135596800 use that instead see #407 # if we pass it as a floating point number, we will run into rounding errors
return DATETIME_ZERO + timedelta(seconds=ts) # see also #407
offset = timedelta(seconds=self.seconds, microseconds=self.nanos // 1000)
return DATETIME_ZERO + offset
@staticmethod @staticmethod
def timestamp_to_json(dt: datetime) -> str: def timestamp_to_json(dt: datetime) -> str:

27
tests/test_timestamp.py Normal file
View File

@ -0,0 +1,27 @@
from datetime import (
datetime,
timezone,
)
import pytest
from betterproto import _Timestamp
@pytest.mark.parametrize(
"dt",
[
datetime(2023, 10, 11, 9, 41, 12, tzinfo=timezone.utc),
datetime.now(timezone.utc),
# potential issue with floating point precision:
datetime(2242, 12, 31, 23, 0, 0, 1, tzinfo=timezone.utc),
# potential issue with negative timestamps:
datetime(1969, 12, 31, 23, 0, 0, 1, tzinfo=timezone.utc),
],
)
def test_timestamp_to_datetime_and_back(dt: datetime):
"""
Make sure converting a datetime to a protobuf timestamp message
and then back again ends up with the same datetime.
"""
assert _Timestamp.from_datetime(dt).to_datetime() == dt