Use io.IOBase for readable, closed, and writeable
MODIFY
src/urllib3/response.py
src/urllib3/response.py
@@ -305,10 +305,6 @@ class BaseHTTPResponse(io.IOBase):
def url(self, url: Optional[str]) -> None:
raise NotImplementedError()
- @property
- def closed(self) -> bool:
- raise NotImplementedError()
-
@property
def connection(self) -> Optional[HTTPConnection]:
raise NotImplementedError()
@@ -406,9 +402,6 @@ class BaseHTTPResponse(io.IOBase):
return b""
# Compatibility methods for `io` module
- def readable(self) -> bool:
- return True
-
def readinto(self, b: bytearray) -> int:
temp = self.read(len(b))
if len(temp) == 0:
@@ -832,6 +825,9 @@ class HTTPResponse(BaseHTTPResponse):
yield data
# Overrides from io.IOBase
+ def readable(self) -> bool:
+ return True
+
def close(self) -> None:
if not self.closed and self._fp:
self._fp.close()

MODIFY
test/test_response.py
test/test_response.py
@@ -22,7 +22,12 @@ from urllib3.exceptions import (
ResponseNotChunked,
SSLError,
)
-from urllib3.response import HTTPResponse, brotli, zstd # type: ignore[attr-defined]
+from urllib3.response import ( # type: ignore[attr-defined]
+ BaseHTTPResponse,
+ HTTPResponse,
+ brotli,
+ zstd,
+)
from urllib3.util.response import is_fp_closed
from urllib3.util.retry import RequestHistory, Retry
@@ -324,6 +329,24 @@ class TestResponse:
assert resp.data == b"foo"
assert resp.closed
+ def test_base_io(self) -> None:
+ resp = BaseHTTPResponse(
+ status=200,
+ version=11,
+ reason=None,
+ decode_content=False,
+ request_url=None,
+ )
+
+ assert not resp.closed
+ assert not resp.readable()
+ assert not resp.writable()
+
+ with pytest.raises(NotImplementedError):
+ resp.read()
+ with pytest.raises(NotImplementedError):
+ resp.close()
+
def test_io(self, sock: socket.socket) -> None:
fp = BytesIO(b"foo")
resp = HTTPResponse(fp, preload_content=False)

Use io.IOBase for readable, closed, and writeable