Fix two bugs in _Timestamp.to_datetime and _Timestamp.from_datetime

src/betterproto/__init__.py

@@ -1887,16 +1887,28 @@ def delta_to_json(delta: timedelta) -> str:
         return f"{'.'.join(parts)}s"
 
 
+_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
+
+
 class _Timestamp(Timestamp):
     @classmethod
     def from_datetime(cls, dt: datetime) -> "_Timestamp":
-        seconds = int(dt.timestamp())
-        nanos = int(dt.microsecond * 1e3)
-        return cls(seconds, nanos)
+        # manual epoch offset calulation to avoid rounding errors and
+        # to support negative timestamps (before 1970)
+        offset = dt - _EPOCH
+        # 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 * 86400 + offset.seconds
+        ) * 10**6 + offset.microseconds
+        seconds, us = divmod(offset_us, 10**6)
+        return cls(seconds, us * 1000)
 
     def to_datetime(self) -> datetime:
-        ts = self.seconds + (self.nanos / 1e9)
-        return datetime.fromtimestamp(ts, tz=timezone.utc)
+        # datetime.fromtimestamp() expects a timestamp in seconds, not microseconds
+        # if we pass it as a floating point number, we will run into rounding errors
+        offset = timedelta(seconds=self.seconds, microseconds=self.nanos // 1000)
+        return _EPOCH + offset
 
     @staticmethod
     def timestamp_to_json(dt: datetime) -> str:

tests/test_timestamp.py

@@ -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