Recover yield-then-raise coverage via stream-write spy

kdeldycke · kdeldycke · commit d4ed88f87510 · 2026-05-22T13:39:14.000+02:00
diff --git a/tests/test_utils.py b/tests/test_utils.py
@@ -289,6 +289,11 @@ def _test_gen_func_fails():
     yield  # unreachable, keeps this a generator function
 
 
+def _test_gen_func_yields_then_fails():
+    yield "test"
+    raise RuntimeError("This is a test.")
+
+
 def _test_gen_func_echo(file=None):
     yield "test"
     click.echo("hello", file=file)
@@ -451,6 +456,41 @@ def test_echo_via_pager(monkeypatch, capfd, pager_cmd, test, tmp_path):
     )
 
 
+@pytest.mark.skipif(WIN, reason="Different behavior on windows.")
+def test_echo_via_pager_yields_before_exception(monkeypatch, tmp_path):
+    """A generator that yields then raises: click writes the partial output to
+    the pager stream before propagating the exception.
+
+    The pager file content is intentionally NOT asserted: pipe-drain timing
+    between click and the pager subprocess is outside click's control
+    (#2899, #3470). Spying on ``MaybeStripAnsi.write`` records what click sent
+    to the pager, which is deterministic regardless of scheduling.
+    """
+    monkeypatch.setitem(os.environ, "PAGER", "cat")
+    monkeypatch.setattr(click._termui_impl, "isatty", lambda x: True)
+
+    writes: list[str] = []
+    real_write = click._termui_impl.MaybeStripAnsi.write
+
+    def spy(self, text):
+        writes.append(text)
+        return real_write(self, text)
+
+    monkeypatch.setattr(click._termui_impl.MaybeStripAnsi, "write", spy)
+
+    pager_out_tmp = tmp_path / "pager_out.txt"
+    with (
+        pager_out_tmp.open("w") as f,
+        patch.object(subprocess, "Popen", partial(subprocess.Popen, stdout=f)),
+        pytest.raises(RuntimeError, match="This is a test."),
+    ):
+        click.echo_via_pager(_test_gen_func_yields_then_fails())
+
+    assert "".join(writes) == "test", (
+        f"click should have written the yielded chunk before exception, got {writes!r}"
+    )
+
+
 @pytest.mark.stress
 @pytest.mark.skipif(WIN, reason="Different behavior on windows.")
 @pytest.mark.parametrize("_", range(1000))
