(self)
| 942 | self.assertTrue(resp.isclosed()) |
| 943 | |
| 944 | def test_partial_readintos_incomplete_body(self): |
| 945 | # if the server shuts down the connection before the whole |
| 946 | # content-length is delivered, the socket is gracefully closed |
| 947 | body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText" |
| 948 | sock = FakeSocket(body) |
| 949 | resp = client.HTTPResponse(sock) |
| 950 | resp.begin() |
| 951 | b = bytearray(2) |
| 952 | n = resp.readinto(b) |
| 953 | self.assertEqual(n, 2) |
| 954 | self.assertEqual(bytes(b), b'Te') |
| 955 | self.assertFalse(resp.isclosed()) |
| 956 | n = resp.readinto(b) |
| 957 | self.assertEqual(n, 2) |
| 958 | self.assertEqual(bytes(b), b'xt') |
| 959 | n = resp.readinto(b) |
| 960 | self.assertEqual(n, 0) |
| 961 | self.assertTrue(resp.isclosed()) |
| 962 | self.assertFalse(resp.closed) |
| 963 | resp.close() |
| 964 | self.assertTrue(resp.closed) |
| 965 | |
| 966 | def test_host_port(self): |
| 967 | # Check invalid host_port |
nothing calls this directly
no test coverage detected