PYTHON-5784 Cover threaded PeriodicExecutor exception path via synchro

Consolidate the target-exception test into the synchro'd source so both
AsyncPeriodicExecutor and the threaded PeriodicExecutor are covered from a
single source, and delete the async-only test file.

Why this reverses the async-only split from 3340959: that commit carved the
exception test out into test_async_periodic_executor.py to dodge the noisy
traceback an uncaught exception prints via threading.excepthook on the
background thread. But the two _run loops handle the exception differently --
the threaded version additionally takes the lock and sets _thread_will_exit,
which feeds the join-on-reopen path in open(). The async test therefore did
not cover that sync-only branch, leaving a real gap (also flagged by Copilot).

Rather than maintain two near-identical hand-written tests (which cuts against
the synchro direction this PR has otherwise followed), the test now lives in
the async source with an `if _IS_SYNC:` branch that swaps threading.excepthook
to a no-op for the threaded case. Verified the generated sync test passes
15/15 with no flakiness or stderr noise, which confirms the path is testable
and there was no real reason to keep it async-only.

Author: aclark4life

test/asynchronous/test_async_periodic_executor.py

@@ -142,6 +142,41 @@ async def target():
         await executor.join(timeout=2)
         self.assertGreaterEqual(called, 2)
 
+    async def test_target_exception_stops_executor(self):
+        call_count = 0
+
+        async def target():
+            nonlocal call_count
+            call_count += 1
+            raise RuntimeError("error")
+
+        executor = self._make_executor(target=target)
+
+        if _IS_SYNC:
+            # The exception re-raises on the executor's background thread,
+            # which would otherwise trigger threading.excepthook and print a
+            # noisy traceback. Swap it for a no-op for the duration of the test.
+            original_excepthook = threading.excepthook
+            threading.excepthook = lambda args: None
+            self.addCleanup(setattr, threading, "excepthook", original_excepthook)
+
+        executor.open()
+        await executor.join(timeout=2)
+        if not _IS_SYNC and executor._task is not None and executor._task.done():
+            # Retrieve the exception to avoid "Task exception was never
+            # retrieved" warnings when the task is garbage collected.
+            executor._task.exception()
+        self.assertEqual(call_count, 1, "target should stop after raising")
+
+        # Re-opening after an exception restarts the executor. For the threaded
+        # PeriodicExecutor this also exercises the _thread_will_exit join path
+        # in open().
+        executor.open()
+        await executor.join(timeout=2)
+        if not _IS_SYNC and executor._task is not None and executor._task.done():
+            executor._task.exception()
+        self.assertEqual(call_count, 2, "executor should run again after re-open")
+
 
 if __name__ == "__main__":
     unittest.main()

test/asynchronous/test_periodic_executor.py

@@ -142,6 +142,41 @@ def target():
         executor.join(timeout=2)
         self.assertGreaterEqual(called, 2)
 
+    def test_target_exception_stops_executor(self):
+        call_count = 0
+
+        def target():
+            nonlocal call_count
+            call_count += 1
+            raise RuntimeError("error")
+
+        executor = self._make_executor(target=target)
+
+        if _IS_SYNC:
+            # The exception re-raises on the executor's background thread,
+            # which would otherwise trigger threading.excepthook and print a
+            # noisy traceback. Swap it for a no-op for the duration of the test.
+            original_excepthook = threading.excepthook
+            threading.excepthook = lambda args: None
+            self.addCleanup(setattr, threading, "excepthook", original_excepthook)
+
+        executor.open()
+        executor.join(timeout=2)
+        if not _IS_SYNC and executor._task is not None and executor._task.done():
+            # Retrieve the exception to avoid "Task exception was never
+            # retrieved" warnings when the task is garbage collected.
+            executor._task.exception()
+        self.assertEqual(call_count, 1, "target should stop after raising")
+
+        # Re-opening after an exception restarts the executor. For the threaded
+        # PeriodicExecutor this also exercises the _thread_will_exit join path
+        # in open().
+        executor.open()
+        executor.join(timeout=2)
+        if not _IS_SYNC and executor._task is not None and executor._task.done():
+            executor._task.exception()
+        self.assertEqual(call_count, 2, "executor should run again after re-open")
+
 
 if __name__ == "__main__":
     unittest.main()

test/test_periodic_executor.py

@@ -190,7 +190,6 @@ def async_only_test(f: str) -> bool:
         "test_async_loop_safety.py",
         "test_async_contextvars_reset.py",
         "test_async_loop_unblocked.py",
-        "test_async_periodic_executor.py",
     ]