make HTTPResponse.stream use read1 when amt=None

changelog/3216.feature.rst

@@ -0,0 +1,3 @@
+HTTPResponse.stream(amt=None) now yields data from non-chunked responses as it
+is available.  Previously all intermediate data was buffered and only returned
+as a single bytes object at the end of the the response.

src/urllib3/response.py

@@ -1020,10 +1020,18 @@ def stream(
             yield from self.read_chunked(amt, decode_content=decode_content)
         else:
             while not is_fp_closed(self._fp) or len(self._decoded_buffer) > 0:
-                data = self.read(amt=amt, decode_content=decode_content)
-
+                if amt is None:
+                    data = self.read1(amt=amt, decode_content=decode_content)
+                else:
+                    data = self.read(amt=amt, decode_content=decode_content)
                 if data:
                     yield data
+                else:
+                    if self._fp:
+                        self._fp.close()
+                    break
+        if self.length_remaining:
+            raise IncompleteRead(self._fp_bytes_read, self.length_remaining)
 
     # Overrides from io.IOBase
     def readable(self) -> bool:

test/test_response.py

@@ -1021,6 +1021,32 @@ def close(self) -> None:
         with pytest.raises(StopIteration):
             next(stream)
 
+    def test_mock_stream_via_read1(self) -> None:
+        class MockHTTPRequest:
+            def __init__(self, chunkiter: typing.Iterator[bytes]) -> None:
+                self.chunkiter = chunkiter
+                self.closed = False
+
+            def read(self, amt: int) -> bytes:
+                "not called, presence indicates to HTTPResponse that this is a stream"
+                assert False
+
+            def read1(self, amt: int | None = None) -> bytes:
+                assert not self.closed
+                try:
+                    return next(self.chunkiter)
+                except StopIteration:
+                    return b""
+
+            def close(self) -> None:
+                self.closed = True
+
+        chunks = b"foo longer bar".split()
+        fp = MockHTTPRequest(iter(chunks))
+        resp = HTTPResponse(fp, preload_content=False)  # type: ignore[arg-type]
+        assert list(resp.stream(None)) == chunks
+        assert fp.closed
+
     def test_mock_transfer_encoding_chunked(self) -> None:
         stream = [b"fo", b"o", b"bar"]
         fp = MockChunkedEncodingResponse(stream)