test(sdk): add tests for force_flush meaningful bool return (#5020)

Author: ravitheja4531-cell

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py

@@ -23,7 +23,7 @@
 from sys import stdout
 from threading import Event, Lock, RLock, Thread
 from time import perf_counter, time_ns
-from typing import IO, Callable, Iterable, Optional
+from typing import IO, Callable, Iterable
 
 from typing_extensions import final
 
@@ -336,7 +336,7 @@ def __init__(
         )
 
     @final
-    def collect(self, timeout_millis: float = 10_000) -> Optional[bool]:
+    def collect(self, timeout_millis: float = 10_000) -> bool | None:
         """Collects the metrics from the internal SDK state and
         invokes the `_receive_metrics` with the collection.
 
@@ -387,7 +387,7 @@ def _receive_metrics(
         metrics_data: MetricsData,
         timeout_millis: float = 10_000,
         **kwargs,
-    ) -> bool:
+    ) -> bool | None:
         """Called by `MetricReader.collect` when it receives a batch of metrics.
 
         Subclasses must return ``True`` on success and ``False`` on failure.
@@ -445,7 +445,7 @@ def __init__(
 
     def get_metrics_data(
         self,
-    ) -> Optional[MetricsData]:
+    ) -> MetricsData | None:
         """Reads and returns current metrics from the SDK"""
         with self._lock:
             self.collect()
@@ -480,8 +480,8 @@ class PeriodicExportingMetricReader(MetricReader):
     def __init__(
         self,
         exporter: MetricExporter,
-        export_interval_millis: Optional[float] = None,
-        export_timeout_millis: Optional[float] = None,
+        export_interval_millis: float | None = None,
+        export_timeout_millis: float | None = None,
     ) -> None:
         # PeriodicExportingMetricReader defers to exporter for configuration
         super().__init__(
@@ -608,6 +608,6 @@ def _shutdown():
         self._exporter.shutdown(timeout=(deadline_ns - time_ns()) / 10**6)
 
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
-        collect_ok = super().force_flush(timeout_millis=timeout_millis)
-        exporter_ok = self._exporter.force_flush(timeout_millis=timeout_millis)
-        return collect_ok and exporter_ok
+        if not super().force_flush(timeout_millis=timeout_millis):
+            return False
+        return self._exporter.force_flush(timeout_millis=timeout_millis)

opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py

@@ -359,4 +359,54 @@ def test_metric_reader_metrics(self):
         assert isinstance(name, str)
         self.assertTrue(name.startswith("periodic_metric_reader/"))
 
-        mp.shutdown()
+        mp.shutdown()                          
+
+    def test_force_flush_returns_true_on_success(self):
+        exporter = FakeMetricsExporter()
+        pmr = self._create_periodic_reader(metrics, exporter)
+        result = pmr.force_flush(timeout_millis=5_000)
+        self.assertTrue(result)
+        pmr.shutdown()
+
+    def test_force_flush_returns_false_on_export_failure(self):
+        exporter = FakeMetricsExporter()
+        exporter.export = Mock(return_value=MetricExportResult.FAILURE)
+        pmr = self._create_periodic_reader(metrics, exporter)
+        result = pmr.force_flush(timeout_millis=5_000)
+        self.assertFalse(result)
+        pmr.shutdown()
+
+    def test_force_flush_skips_exporter_flush_when_collect_fails(self):
+        exporter = FakeMetricsExporter()
+        exporter.force_flush = Mock(return_value=True)
+        pmr = PeriodicExportingMetricReader(
+            exporter, export_interval_millis=math.inf
+        )
+        # No collect callback registered → collect returns None → force_flush
+        # on base treats None as not-False (success), so wire up a failing one
+        exporter.export = Mock(return_value=MetricExportResult.FAILURE)
+
+        def _collect_failure(reader, timeout_millis):
+            return metrics
+
+        pmr._set_collect_callback(_collect_failure)
+        exporter.export = Mock(return_value=MetricExportResult.FAILURE)
+        result = pmr.force_flush(timeout_millis=5_000)
+        self.assertFalse(result)
+        exporter.force_flush.assert_not_called()
+        pmr.shutdown()
+
+    def test_detach_called_on_export_failure(self):
+        """detach(token) must run in finally even when export returns FAILURE."""
+        from unittest.mock import patch
+
+        exporter = FakeMetricsExporter()
+        exporter.export = Mock(return_value=MetricExportResult.FAILURE)
+        pmr = self._create_periodic_reader(metrics, exporter)
+
+        with patch(
+            "opentelemetry.sdk.metrics._internal.export.detach"
+        ) as mock_detach:
+            pmr.force_flush(timeout_millis=5_000)
+            self.assertTrue(mock_detach.called)
+        pmr.shutdown()