Copilot review

Author: aclark4life

test/asynchronous/test_periodic_executor.py

@@ -41,7 +41,7 @@ async def target():
     )
 
 
-class AsyncPeriodicExecutorTestBase(AsyncUnitTest):
+class TestAsyncPeriodicExecutor(AsyncUnitTest):
     async def asyncSetUp(self):
         self.executor = None
 
@@ -50,14 +50,6 @@ async def asyncTearDown(self):
             self.executor.close()
             await self.executor.join(timeout=2)
 
-
-class TestAsyncPeriodicExecutor(AsyncPeriodicExecutorTestBase):
-    async def test_repr_contains_class_and_name(self):
-        executor = _make_executor(name="exec")
-        executor_repr = repr(executor)
-        self.assertIn("AsyncPeriodicExecutor", executor_repr)
-        self.assertIn("exec", executor_repr)
-
     async def test_join_without_open_is_safe(self):
         self.executor = _make_executor()
         try:
@@ -81,41 +73,25 @@ async def target():
         self.assertTrue(ran.is_set(), "target never ran")
 
     async def test_target_exception_stops_executor(self):
-        if _IS_SYNC:
-            ran = threading.Event()
-            captured_exc: list = []
-            orig_excepthook = threading.excepthook
-
-            def _capture_excepthook(args):
-                captured_exc.append(args.exc_value)
-
-            threading.excepthook = _capture_excepthook
-            try:
-
-                def target():
-                    ran.set()
-                    raise RuntimeError("error")
-
-                self.executor = _make_executor(target=target)
-                self.executor.open()
-                self.executor.join(timeout=2)
-                self.assertTrue(ran.is_set(), "target never ran")
-            finally:
-                threading.excepthook = orig_excepthook
-            self.assertEqual(len(captured_exc), 1)
-            self.assertIsInstance(captured_exc[0], RuntimeError)
-        else:
-            call_count = 0
+        call_count = 0
 
-            async def target():
-                nonlocal call_count
-                call_count += 1
-                raise RuntimeError("error")
+        async def target():
+            nonlocal call_count
+            call_count += 1
+            raise RuntimeError("error")
 
-            self.executor = _make_executor(target=target)
+        self.executor = _make_executor(target=target)
+        # Suppress threading.excepthook so the intentional RuntimeError raised
+        # in the worker thread is not surfaced by pytest's threadexception
+        # plugin (no-op for the async executor since no thread is involved).
+        orig_excepthook = threading.excepthook
+        threading.excepthook = lambda _args: None
+        try:
             self.executor.open()
             await self.executor.join(timeout=2)
-            self.assertEqual(call_count, 1, "target should stop after exception")
+        finally:
+            threading.excepthook = orig_excepthook
+        self.assertEqual(call_count, 1, "target should stop after exception")
 
     async def test_skip_sleep_flag_skips_interval(self):
         call_times = []

test/test_periodic_executor.py

@@ -39,7 +39,7 @@ def target():
     return PeriodicExecutor(interval=interval, min_interval=min_interval, target=target, name=name)
 
 
-class PeriodicExecutorTestBase(UnitTest):
+class TestPeriodicExecutor(UnitTest):
     def setUp(self):
         self.executor = None
 
@@ -48,14 +48,6 @@ def tearDown(self):
             self.executor.close()
             self.executor.join(timeout=2)
 
-
-class TestPeriodicExecutor(PeriodicExecutorTestBase):
-    def test_repr_contains_class_and_name(self):
-        executor = _make_executor(name="exec")
-        executor_repr = repr(executor)
-        self.assertIn("PeriodicExecutor", executor_repr)
-        self.assertIn("exec", executor_repr)
-
     def test_join_without_open_is_safe(self):
         self.executor = _make_executor()
         try:
@@ -79,41 +71,25 @@ def target():
         self.assertTrue(ran.is_set(), "target never ran")
 
     def test_target_exception_stops_executor(self):
-        if _IS_SYNC:
-            ran = threading.Event()
-            captured_exc: list = []
-            orig_excepthook = threading.excepthook
-
-            def _capture_excepthook(args):
-                captured_exc.append(args.exc_value)
-
-            threading.excepthook = _capture_excepthook
-            try:
-
-                def target():
-                    ran.set()
-                    raise RuntimeError("error")
-
-                self.executor = _make_executor(target=target)
-                self.executor.open()
-                self.executor.join(timeout=2)
-                self.assertTrue(ran.is_set(), "target never ran")
-            finally:
-                threading.excepthook = orig_excepthook
-            self.assertEqual(len(captured_exc), 1)
-            self.assertIsInstance(captured_exc[0], RuntimeError)
-        else:
-            call_count = 0
+        call_count = 0
 
-            def target():
-                nonlocal call_count
-                call_count += 1
-                raise RuntimeError("error")
+        def target():
+            nonlocal call_count
+            call_count += 1
+            raise RuntimeError("error")
 
-            self.executor = _make_executor(target=target)
+        self.executor = _make_executor(target=target)
+        # Suppress threading.excepthook so the intentional RuntimeError raised
+        # in the worker thread is not surfaced by pytest's threadexception
+        # plugin (no-op for the async executor since no thread is involved).
+        orig_excepthook = threading.excepthook
+        threading.excepthook = lambda _args: None
+        try:
             self.executor.open()
             self.executor.join(timeout=2)
-            self.assertEqual(call_count, 1, "target should stop after exception")
+        finally:
+            threading.excepthook = orig_excepthook
+        self.assertEqual(call_count, 1, "target should stop after exception")
 
     def test_skip_sleep_flag_skips_interval(self):
         call_times = []