test(time): Make nanos assertable (#4562)

For: #4559

Make millies/micros/nanos assertable with our `time_*` assertion
helpers.

Author: Dav1dde

tests/integration/asserts/time.py

@@ -1,24 +1,40 @@
 from datetime import timedelta, datetime, timezone
+from enum import StrEnum
 
 
-def _to_datetime(v):
+class Resolution(StrEnum):
+    Seconds = "s"
+    MilliSeconds = "ms"
+    MicroSeconds = "us"
+    NanoSeconds = "ns"
+
+
+def _to_datetime(v, resolution=Resolution.Seconds):
     if isinstance(v, datetime):
         return v
     elif isinstance(v, (int, float)):
-        return datetime.fromtimestamp(v, timezone.utc)
+        resolution = Resolution(resolution)
+        to_secs_d = {
+            Resolution.Seconds: 1,
+            Resolution.MilliSeconds: 1_000,
+            Resolution.MicroSeconds: 1_000_000,
+            Resolution.NanoSeconds: 1_000_000_000,
+        }[resolution]
+        return datetime.fromtimestamp(v / to_secs_d, timezone.utc)
     elif isinstance(v, str):
         return datetime.fromisoformat(v)
     else:
         assert False, f"cannot convert {v} to datetime"
 
 
 class _WithinBounds:
-    def __init__(self, lower_bound, upper_bound):
+    def __init__(self, lower_bound, upper_bound, expect_resolution=Resolution.Seconds):
         self._lower_bound = lower_bound
         self._upper_bound = upper_bound
+        self._expect_resolution = expect_resolution
 
     def __eq__(self, other):
-        other = _to_datetime(other)
+        other = _to_datetime(other, resolution=self._expect_resolution)
         return self._lower_bound <= other <= self._upper_bound
 
     def __str__(self):
@@ -28,18 +44,18 @@ def __repr__(self) -> str:
         return str(self)
 
 
-def time_after(lower_bound):
+def time_after(lower_bound, **kwargs):
     upper_bound = datetime.now(tz=timezone.utc)
-    return time_within(lower_bound, upper_bound)
+    return time_within(lower_bound, upper_bound, **kwargs)
 
 
-def time_within(lower_bound, upper_bound):
+def time_within(lower_bound, upper_bound, **kwargs):
     lower_bound = _to_datetime(lower_bound)
     upper_bound = _to_datetime(upper_bound)
     assert lower_bound <= upper_bound
-    return _WithinBounds(lower_bound, upper_bound)
+    return _WithinBounds(lower_bound, upper_bound, **kwargs)
 
 
-def time_within_delta(time=None, delta=timedelta(seconds=30)):
+def time_within_delta(time=None, delta=timedelta(seconds=30), **kwargs):
     time = _to_datetime(time) if time is not None else datetime.now(tz=timezone.utc)
-    return _WithinBounds(time - delta, time + delta)
+    return _WithinBounds(time - delta, time + delta, **kwargs)

tests/integration/test_ourlogs.py

@@ -3,6 +3,8 @@
 
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
 
+from .asserts.time import time_within_delta
+
 
 TEST_CONFIG = {
     "aggregator": {
@@ -77,7 +79,9 @@ def test_ourlog_extraction(
         "project_id": 42,
         "retention_days": 90,
         "timestamp_nanos": int(start.timestamp() * 1e9),
-        "observed_timestamp_nanos": int(end.timestamp() * 1e9),
+        "observed_timestamp_nanos": time_within_delta(
+            start.timestamp(), expect_resolution="ns"
+        ),
         "trace_id": "5b8efff798038103d269b633813fc60c",
         "body": "Example log record",
         "trace_flags": 0,