use time mocks

Author: daniel-sanche

google/cloud/bigtable/data/_metrics/data_model.py

@@ -128,7 +128,7 @@ class ActiveAttemptMetric:
     """
 
     # keep monotonic timestamps for active attempts
-    start_time_ns: int = field(default_factory=time.monotonic_ns)
+    start_time_ns: int = field(default_factory=lambda: time.monotonic_ns())
     # the time taken by the backend, in nanoseconds. Taken from response header
     gfe_latency_ns: int | None = None
     # time waiting on user to process the response, in nanoseconds
@@ -157,7 +157,7 @@ class ActiveOperationMetric:
         )
     )
     # keep monotonic timestamps for active operations
-    start_time_ns: int = field(default_factory=time.monotonic_ns)
+    start_time_ns: int = field(default_factory=lambda: time.monotonic_ns())
     active_attempt: ActiveAttemptMetric | None = None
     cluster_id: str | None = None
     zone: str | None = None

tests/unit/data/_metrics/test_data_model.py

@@ -26,16 +26,17 @@ def _make_one(self, *args, **kwargs):
 
         return ActiveOperationMetric(*args, **kwargs)
 
-    def test_ctor_defaults(self):
+    @mock.patch("time.monotonic_ns")
+    def test_ctor_defaults(self, mock_monotonic_ns):
         """
         create an instance with default values
         """
+        expected_timestamp = 123456789
+        mock_monotonic_ns.return_value = expected_timestamp
         mock_type = mock.Mock()
         metric = self._make_one(mock_type)
         assert metric.op_type == mock_type
-        assert metric.start_time_ns == pytest.approx(
-            time.monotonic_ns(), abs=50e6
-        )  # 50 ms buffer
+        assert metric.start_time_ns == expected_timestamp
         assert metric.active_attempt is None
         assert metric.cluster_id is None
         assert metric.zone is None
@@ -176,33 +177,39 @@ def test_error_invalid_states(self, method, args, valid_states, error_method_nam
                     == f"Invalid state for {error_method_name}: {state}"
                 )
 
-    def test_start(self):
+    @mock.patch("time.monotonic_ns")
+    def test_start(self, mock_monotonic_ns):
         """
         calling start op operation should reset start_time
         """
+        expected_timestamp = 123456789
+        mock_monotonic_ns.return_value = expected_timestamp
         orig_time = 0
         metric = self._make_one(mock.Mock(), start_time_ns=orig_time)
         assert metric.start_time_ns == 0
         metric.start()
         assert metric.start_time_ns != orig_time
-        assert metric.start_time_ns == pytest.approx(time.monotonic_ns(), abs=50e6)
+        assert metric.start_time_ns == expected_timestamp
         # should remain in CREATED state after completing
         assert metric.state == State.CREATED
 
-    def test_start_attempt(self):
+    @mock.patch("time.monotonic_ns")
+    def test_start_attempt(self, mock_monotonic_ns):
         """
         calling start_attempt should create a new emptu atempt metric
         """
         from google.cloud.bigtable.data._metrics.data_model import ActiveAttemptMetric
 
+        from google.cloud.bigtable.data._metrics.data_model import ActiveAttemptMetric
+
+        expected_timestamp = 123456789
+        mock_monotonic_ns.return_value = expected_timestamp
         metric = self._make_one(mock.Mock())
         assert metric.active_attempt is None
         metric.start_attempt()
         assert isinstance(metric.active_attempt, ActiveAttemptMetric)
         # make sure it was initialized with the correct values
-        assert metric.active_attempt.start_time_ns == pytest.approx(
-            time.monotonic_ns(), abs=50e6
-        )
+        assert metric.active_attempt.start_time_ns == expected_timestamp
         assert metric.active_attempt.gfe_latency_ns is None
         assert metric.active_attempt.grpc_throttling_time_ns == 0
         # should be in ACTIVE_ATTEMPT state after completing
@@ -387,14 +394,17 @@ def test_add_response_metadata_server_timing_header(
             assert metric.cluster_id is None
             assert metric.zone is None
 
-    def test_end_attempt_with_status(self):
+    @mock.patch("time.monotonic_ns")
+    def test_end_attempt_with_status(self, mock_monotonic_ns):
         """
         ending the attempt should:
         - add one to completed_attempts
         - reset active_attempt to None
         - update state
         - notify handlers
         """
+        expected_mock_time = 123456789
+        mock_monotonic_ns.return_value = expected_mock_time
         expected_start_time = 1
         expected_status = object()
         expected_gfe_latency_ns = 5
@@ -415,8 +425,8 @@ def test_end_attempt_with_status(self):
         metric.end_attempt_with_status(expected_status)
         assert len(metric.completed_attempts) == 1
         got_attempt = metric.completed_attempts[0]
-        expected_duration = time.monotonic_ns() - expected_start_time
-        assert got_attempt.duration_ns == pytest.approx(expected_duration, abs=50e6)
+        expected_duration = expected_mock_time - expected_start_time
+        assert got_attempt.duration_ns == expected_duration
         assert got_attempt.grpc_throttling_time_ns == expected_grpc_throttle
         assert got_attempt.end_status == expected_status
         assert got_attempt.gfe_latency_ns == expected_gfe_latency_ns
@@ -447,7 +457,8 @@ def test_end_attempt_with_status_w_exception(self):
             assert mock_exc_to_status.call_args[0][0] == input_status
             assert metric.completed_attempts[0].end_status == expected_status
 
-    def test_end_with_status(self):
+    @mock.patch("time.monotonic_ns")
+    def test_end_with_status(self, mock_monotonic_ns):
         """
         ending the operation should:
         - end active attempt
@@ -456,6 +467,8 @@ def test_end_with_status(self):
         """
         from google.cloud.bigtable.data._metrics.data_model import ActiveAttemptMetric
 
+        expected_mock_time = 123456789
+        mock_monotonic_ns.return_value = expected_mock_time
         expected_attempt_start_time = 0
         expected_attempt_gfe_latency_ns = 5
         expected_flow_time = 16
@@ -494,8 +507,8 @@ def test_end_with_status(self):
             assert len(h.on_operation_complete.call_args[0]) == 1
             called_with = h.on_operation_complete.call_args[0][0]
             assert called_with.op_type == expected_type
-            expected_duration = time.monotonic_ns() - expected_start_time
-            assert called_with.duration_ns == pytest.approx(expected_duration, abs=50e6)
+            expected_duration = expected_mock_time - expected_start_time
+            assert called_with.duration_ns == expected_duration
             assert called_with.final_status == expected_status
             assert called_with.cluster_id == expected_cluster
             assert called_with.zone == expected_zone
@@ -510,10 +523,8 @@ def test_end_with_status(self):
             final_attempt = called_with.completed_attempts[0]
             assert final_attempt.gfe_latency_ns == expected_attempt_gfe_latency_ns
             assert final_attempt.end_status == expected_status
-            expected_duration = time.monotonic_ns() - expected_attempt_start_time
-            assert final_attempt.duration_ns == pytest.approx(
-                expected_duration, abs=50e6
-            )
+            expected_duration = expected_mock_time - expected_attempt_start_time
+            assert final_attempt.duration_ns == expected_duration
 
     def test_end_with_status_w_exception(self):
         """