test(parallel): close coverage gaps to 100% branch

Author: wolph

progressbar/_parallel/_async.py

@@ -141,11 +141,8 @@ async def completions(self) -> typing.AsyncIterator[Completion]:
         while self.in_flight:
             self._check_deadline()
             task = await self._next_done()
-            if task is None:
-                continue
-            event: Completion | None = self._handle(task)
-            if event is not None:
-                yield event
+            if task is not None:
+                yield self._handle(task)
 
     def _launch_one(self) -> bool:
         """Create the next task; `False` when the input is exhausted."""
@@ -196,7 +193,7 @@ def _check_deadline(self) -> None:
                 f'parallel execution exceeded timeout={self.timeout}'
             )
 
-    def _handle(self, task: asyncio.Task[typing.Any]) -> Completion | None:
+    def _handle(self, task: asyncio.Task[typing.Any]) -> Completion:
         """Turn one finished task into a completion event."""
         index, args, seq = self.in_flight.pop(task)
         if task.cancelled():

progressbar/_parallel/_display.py

@@ -280,10 +280,8 @@ def tick(self) -> None:
 
     def finish(self, *, success: bool = True) -> None:
         """Finish the overall bar and wind down the render thread."""
-        for key in list(self._keys.values()):
-            if key in self.multibar:
-                del self.multibar[key]
-        self._keys.clear()
+        for seq in list(self._keys):
+            self.task_finished(seq, ok=success)
         self._overall.finish(dirty=not success)
         if self._started_thread:
             # One last frame so the final state is on screen even if

progressbar/_parallel/_sync.py

@@ -109,7 +109,7 @@ def _interpreter_executor(
         )
     except AttributeError:
         raise ValueError('pool="interpreter" requires Python 3.14+') from None
-    return interpreter_pool(
+    return interpreter_pool(  # pragma: no cover - reachable on 3.14+ only
         max_workers=workers,
         initializer=initializer,
         initargs=initargs,

tests/test_parallel_async.py

@@ -347,6 +347,46 @@ async def _run() -> tuple[list[int], list[tuple[int, int]]]:
         assert pairs == [(0, 0), (1, 2), (2, 4)]
 
 
+class TestMultiBarMode:
+    def test_async_workers_see_their_task_bar(self) -> None:
+        from progressbar._parallel import _common
+
+        seen: list[bool] = []
+
+        async def _check(value: int) -> int:
+            seen.append(_common.current_task_bar() is not None)
+            return value
+
+        async def _run() -> list[int]:
+            return await _async.amap(
+                _check, range(3), bar='multi', fd=io.StringIO()
+            )
+
+        assert asyncio.run(_run()) == [0, 1, 2]
+        assert seen == [True, True, True]
+
+
+class TestExternalCancellation:
+    def test_self_cancelling_task_surfaces(self) -> None:
+        async def _self_cancel(value: int) -> int:
+            if value == 1:
+                task = asyncio.current_task()
+                assert task is not None
+                task.cancel()
+                await asyncio.sleep(1)
+            return value
+
+        async def _run() -> list[int]:
+            return await _async.amap(
+                _self_cancel, range(3), concurrency=1, bar=False
+            )
+
+        # A cancellation this run did not initiate must surface, never
+        # silently drop the item.
+        with pytest.raises(asyncio.CancelledError):
+            asyncio.run(_run())
+
+
 class TestCallStrategy:
     def test_detects_coroutine_function(self) -> None:
         assert _async._call_strategy(_async_double) == 'async'

tests/test_parallel_display.py

@@ -152,6 +152,21 @@ def test_render_thread_stopped_after_finish(self) -> None:
         display.finish()
         assert display.multibar._thread is None  # noqa: SLF001
 
+    def test_task_finished_with_unknown_seq_is_a_noop(self) -> None:
+        display = self._multi()
+        display.start(1)
+        display.task_finished(99, ok=True)
+        display.finish()
+
+    def test_finish_removes_live_task_bars(self) -> None:
+        display = self._multi()
+        display.start(3)
+        display.task_started(1, 'still-running')
+        display.task_started(2, 'also-running')
+        display.finish(success=False)
+        assert '1: still-running' not in display.multibar
+        assert '2: also-running' not in display.multibar
+
     def test_adopts_existing_multibar_without_stopping_it(self) -> None:
         multibar = progressbar.MultiBar(fd=io.StringIO())
         multibar.start()

tests/test_parallel_map.py

@@ -100,6 +100,32 @@ def _record(value: int) -> int:
         assert main_thread.name not in seen
 
 
+class TestMultiBarMode:
+    def test_workers_see_their_task_bar(self) -> None:
+        from progressbar._parallel import _common
+
+        seen: list[bool] = []
+
+        def _check(value: int) -> int:
+            seen.append(_common.current_task_bar() is not None)
+            return value
+
+        _sync.map(_check, range(4), workers=2, bar='multi', fd=io.StringIO())
+        assert seen == [True, True, True, True]
+
+    def test_plain_mode_has_no_task_bar(self) -> None:
+        from progressbar._parallel import _common
+
+        seen: list[bool] = []
+
+        def _check(value: int) -> int:
+            seen.append(_common.current_task_bar() is None)
+            return value
+
+        _sync.map(_check, range(2), workers=2, bar=False)
+        assert seen == [True, True]
+
+
 class TestResolveExecutor:
     def test_thread_pool_created_and_owned(self) -> None:
         executor, owned, workers = _sync.resolve_executor(

tests/test_parallel_shell.py

@@ -124,6 +124,10 @@ def test_on_error_return_embeds_the_error(self) -> None:
         assert results[0].returncode == 0
         assert isinstance(results[1], subprocess.CalledProcessError)
 
+    def test_pool_kwarg_rejected(self) -> None:
+        with pytest.raises(TypeError, match=r'Pool\.run'):
+            _shell.run(_EXIT, [1], pool='process', bar=False)
+
     @pytest.mark.no_freezegun
     def test_pool_run_method(self) -> None:
         with _sync.Pool(2) as pool: