(
self, sock: socket.socket, length_known: bool, read_amt: int | None
)
| 1073 | @pytest.mark.parametrize("read_amt", (None, 3)) |
| 1074 | @pytest.mark.parametrize("length_known", (True, False)) |
| 1075 | def test_io_closed_consistently_by_read1( |
| 1076 | self, sock: socket.socket, length_known: bool, read_amt: int | None |
| 1077 | ) -> None: |
| 1078 | with httplib.HTTPResponse(sock) as hlr: |
| 1079 | hlr.fp = BytesIO(b"foo") # type: ignore[assignment] |
| 1080 | hlr.chunked = 0 # type: ignore[assignment] |
| 1081 | hlr.length = 3 if length_known else None |
| 1082 | with HTTPResponse(hlr, preload_content=False) as resp: |
| 1083 | if length_known: |
| 1084 | resp.length_remaining = 3 |
| 1085 | assert not resp.closed |
| 1086 | assert resp._fp is not None |
| 1087 | assert not resp._fp.isclosed() |
| 1088 | assert not is_fp_closed(resp._fp) |
| 1089 | assert not resp.isclosed() |
| 1090 | resp.read1(read_amt) |
| 1091 | # If content length is unknown, IO is not closed until |
| 1092 | # the next read returning zero bytes. |
| 1093 | if not length_known: |
| 1094 | assert not resp.closed |
| 1095 | assert resp._fp is not None |
| 1096 | assert not resp._fp.isclosed() |
| 1097 | assert not is_fp_closed(resp._fp) |
| 1098 | assert not resp.isclosed() |
| 1099 | resp.read1(read_amt) |
| 1100 | assert resp.closed |
| 1101 | assert resp._fp.isclosed() |
| 1102 | assert is_fp_closed(resp._fp) |
| 1103 | assert resp.isclosed() |
| 1104 | |
| 1105 | @pytest.mark.parametrize("length_known", (True, False)) |
| 1106 | def test_io_not_closed_until_all_data_is_read( |
nothing calls this directly
no test coverage detected