Update concurrency tests to use ConcurrencyTestBase and add result validation

Author: chimchim89

opentelemetry-api/tests/metrics/test_instruments.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 # type: ignore
 
-import threading
 from inspect import Signature, isabstract, signature
 from unittest import TestCase
 
@@ -32,6 +31,7 @@
     UpDownCounter,
     _Gauge,
 )
+from opentelemetry.test.concurrency_test import ConcurrencyTestBase
 
 
 class ChildInstrument(Instrument):
@@ -725,42 +725,31 @@ def test_description_check(self):
         )
 
 
-class TestConcurrency(TestCase):
-    def _run_concurrently(self, fn, num_threads=10, calls_per_thread=100):
-        """Helper: run fn concurrently across threads and assert no exceptions."""
-        errors = []
-
-        def worker():
-            try:
-                for _ in range(calls_per_thread):
-                    fn()
-            except Exception as exc:  # pylint: disable=broad-except
-                errors.append(exc)
-
-        threads = [threading.Thread(target=worker) for _ in range(num_threads)]
-        for thread in threads:
-            thread.start()
-        for thread in threads:
-            thread.join()
-
-        self.assertEqual([], errors)
-
+class TestConcurrency(ConcurrencyTestBase):
     def test_counter_add_concurrent(self):
         """Test that Counter.add can be called concurrently safely."""
         counter = NoOpCounter("name")
-        self._run_concurrently(lambda: counter.add(1))
+        results = self.run_with_many_threads(lambda: counter.add(1))
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(result is None for result in results))
 
     def test_up_down_counter_add_concurrent(self):
         """Test that UpDownCounter.add can be called concurrently safely."""
         up_down_counter = NoOpUpDownCounter("name")
-        self._run_concurrently(lambda: up_down_counter.add(1))
+        results = self.run_with_many_threads(lambda: up_down_counter.add(1))
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(result is None for result in results))
 
     def test_histogram_record_concurrent(self):
         """Test that Histogram.record can be called concurrently safely."""
         histogram = NoOpHistogram("name")
-        self._run_concurrently(lambda: histogram.record(1))
+        results = self.run_with_many_threads(lambda: histogram.record(1))
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(result is None for result in results))
 
     def test_gauge_set_concurrent(self):
         """Test that Gauge.set can be called concurrently safely."""
         gauge = NoOpMeter("name").create_gauge("name")
-        self._run_concurrently(lambda: gauge.set(1))
+        results = self.run_with_many_threads(lambda: gauge.set(1))
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(result is None for result in results))

opentelemetry-api/tests/metrics/test_meter.py

@@ -13,12 +13,22 @@
 # limitations under the License.
 # type: ignore
 
-import threading
 from logging import WARNING
 from unittest import TestCase
 from unittest.mock import Mock, patch
 
-from opentelemetry.metrics import Meter, NoOpMeter
+from opentelemetry.metrics import (
+    Counter,
+    Histogram,
+    Meter,
+    NoOpMeter,
+    ObservableCounter,
+    ObservableGauge,
+    ObservableUpDownCounter,
+    UpDownCounter,
+    _Gauge,
+)
+from opentelemetry.test.concurrency_test import ConcurrencyTestBase
 
 
 class ChildMeter(Meter):
@@ -196,71 +206,74 @@ def test_create_observable_up_down_counter(self):
         )
 
 
-class TestConcurrency(TestCase):
-    def _run_concurrently(self, fn, num_threads=10, calls_per_thread=100):
-        """Helper: run fn concurrently across threads and assert no exceptions."""
-        errors = []
-
-        def worker():
-            try:
-                for _ in range(calls_per_thread):
-                    fn()
-            except Exception as exc:  # pylint: disable=broad-except
-                errors.append(exc)
-
-        threads = [threading.Thread(target=worker) for _ in range(num_threads)]
-        for thread in threads:
-            thread.start()
-        for thread in threads:
-            thread.join()
-
-        self.assertEqual([], errors)
-
+class TestConcurrency(ConcurrencyTestBase):
     def test_create_counter_concurrent(self):
         """Test that Meter.create_counter can be called concurrently safely."""
         meter = NoOpMeter("name")
-        self._run_concurrently(lambda: meter.create_counter("counter"))
+        results = self.run_with_many_threads(
+            lambda: meter.create_counter("counter")
+        )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, Counter) for r in results))
 
     def test_create_up_down_counter_concurrent(self):
         """Test that Meter.create_up_down_counter can be called concurrently safely."""
         meter = NoOpMeter("name")
-        self._run_concurrently(
+        results = self.run_with_many_threads(
             lambda: meter.create_up_down_counter("up_down_counter")
         )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, UpDownCounter) for r in results))
 
     def test_create_observable_counter_concurrent(self):
         """Test that Meter.create_observable_counter can be called concurrently safely."""
         meter = NoOpMeter("name")
-        self._run_concurrently(
+        results = self.run_with_many_threads(
             lambda: meter.create_observable_counter(
                 "observable_counter", lambda options: []
             )
         )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, ObservableCounter) for r in results))
 
     def test_create_histogram_concurrent(self):
         """Test that Meter.create_histogram can be called concurrently safely."""
         meter = NoOpMeter("name")
-        self._run_concurrently(lambda: meter.create_histogram("histogram"))
+        results = self.run_with_many_threads(
+            lambda: meter.create_histogram("histogram")
+        )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, Histogram) for r in results))
 
     def test_create_gauge_concurrent(self):
         """Test that Meter.create_gauge can be called concurrently safely."""
         meter = NoOpMeter("name")
-        self._run_concurrently(lambda: meter.create_gauge("gauge"))
+        results = self.run_with_many_threads(
+            lambda: meter.create_gauge("gauge")
+        )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, _Gauge) for r in results))
 
     def test_create_observable_gauge_concurrent(self):
         """Test that Meter.create_observable_gauge can be called concurrently safely."""
         meter = NoOpMeter("name")
-        self._run_concurrently(
+        results = self.run_with_many_threads(
             lambda: meter.create_observable_gauge(
                 "observable_gauge", lambda options: []
             )
         )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, ObservableGauge) for r in results))
 
     def test_create_observable_up_down_counter_concurrent(self):
         """Test that Meter.create_observable_up_down_counter can be called concurrently safely."""
         meter = NoOpMeter("name")
-        self._run_concurrently(
+        results = self.run_with_many_threads(
             lambda: meter.create_observable_up_down_counter(
                 "observable_up_down_counter", lambda options: []
             )
         )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(
+            all(isinstance(r, ObservableUpDownCounter) for r in results)
+        )

opentelemetry-api/tests/metrics/test_meter_provider.py

@@ -15,7 +15,6 @@
 
 # pylint: disable=protected-access
 
-import threading
 from unittest import TestCase
 from unittest.mock import Mock, patch
 
@@ -44,6 +43,7 @@
     _ProxyObservableUpDownCounter,
     _ProxyUpDownCounter,
 )
+from opentelemetry.test.concurrency_test import ConcurrencyTestBase
 from opentelemetry.test.globals_test import (
     MetricsGlobalsTest,
     reset_metrics_globals,
@@ -164,35 +164,24 @@ def test_get_meter_wrapper(self):
         self.assertEqual(meter.schema_url, "schema_url")
 
 
-class TestConcurrency(TestCase):
-    def _run_concurrently(self, fn, num_threads=10, calls_per_thread=100):
-        """Helper: run fn concurrently across threads and assert no exceptions."""
-        errors = []
-
-        def worker():
-            try:
-                for _ in range(calls_per_thread):
-                    fn()
-            except Exception as exc:  # pylint: disable=broad-except
-                errors.append(exc)
-
-        threads = [threading.Thread(target=worker) for _ in range(num_threads)]
-        for thread in threads:
-            thread.start()
-        for thread in threads:
-            thread.join()
-
-        self.assertEqual([], errors)
-
+class TestConcurrency(ConcurrencyTestBase):
     def test_no_op_meter_provider_get_meter_concurrent(self):
         """Test that NoOpMeterProvider.get_meter can be called concurrently safely."""
         meter_provider = NoOpMeterProvider()
-        self._run_concurrently(lambda: meter_provider.get_meter("name"))
+        results = self.run_with_many_threads(
+            lambda: meter_provider.get_meter("name")
+        )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, NoOpMeter) for r in results))
 
     def test_proxy_meter_provider_get_meter_concurrent(self):
         """Test that _ProxyMeterProvider.get_meter can be called concurrently safely."""
         meter_provider = _ProxyMeterProvider()
-        self._run_concurrently(lambda: meter_provider.get_meter("name"))
+        results = self.run_with_many_threads(
+            lambda: meter_provider.get_meter("name")
+        )
+        self.assertEqual(len(results), 100)
+        self.assertTrue(all(isinstance(r, _ProxyMeter) for r in results))
 
 
 class TestProxy(MetricsGlobalsTest, TestCase):