diff --git a/CHANGELOG.rst b/CHANGELOG.rst index a13405d..7d0003c 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -8,6 +8,11 @@ Versions follow `Semantic Versioning `_ `Unreleased`_ ------------- +Fixed +~~~~~ +* :meth:`.ULID.from_datetime` and :meth:`.ULID.from_timestamp` no longer land one millisecond early + when a value on an exact millisecond multiplies to just below that integer as a float. + `4.1.0`_ - 2026-10-27 --------------------- Changed diff --git a/tests/test_ulid.py b/tests/test_ulid.py index bc26969..1baa740 100644 --- a/tests/test_ulid.py +++ b/tests/test_ulid.py @@ -337,6 +337,19 @@ def test_ulid_from_timestamp() -> None: assert ulid1.timestamp == ulid2.timestamp +@pytest.mark.parametrize("milliseconds", [1078625331860, 1097375394952]) +def test_ulid_from_exact_milliseconds(milliseconds: int) -> None: + # Both values multiply back to just below the integer: x.860 * 1000 == x859.9999... + seconds = milliseconds / constants.MILLISECS_IN_SECS + assert ULID.from_timestamp(seconds).milliseconds == milliseconds + assert ULID.from_datetime(datetime.fromtimestamp(seconds, UTC)).milliseconds == milliseconds + + +def test_ulid_timestamp_keeps_truncating_sub_milliseconds() -> None: + expected = 1000999 + assert ULID.from_timestamp(1000.9999).milliseconds == expected + + def test_ulid_from_timestamp_overflow() -> None: with pytest.raises(ValueError, match="Value exceeds maximum possible timestamp"): ULID.from_timestamp(constants.MAX_TIMESTAMP + 1) diff --git a/ulid/__init__.py b/ulid/__init__.py index 61af2c2..e328a50 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -188,13 +188,18 @@ def _default_clock() -> int: def _default_randomness(_timestamp: int) -> bytes: return os.urandom(constants.RANDOMNESS_LEN) + @staticmethod + def _seconds_to_milliseconds(seconds: float) -> int: + # Round off the float error first, or x.860 s truncates to x859 ms. + return int(round(seconds * constants.MILLISECS_IN_SECS, 3)) + def _normalize_timestamp(self, value: float | datetime | None = None) -> int: if value is None: value = self.clock() elif isinstance(value, datetime): - value = int(value.timestamp() * constants.MILLISECS_IN_SECS) + value = self._seconds_to_milliseconds(value.timestamp()) elif isinstance(value, float): - value = int(value * constants.MILLISECS_IN_SECS) + value = self._seconds_to_milliseconds(value) if value > constants.MAX_TIMESTAMP: raise ValueError("Value exceeds maximum possible timestamp") return value