Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"Connection: Keep-Alive\r\n"
"Content-Length: 0\r\n"
"\r\n"
)
self.connect()
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line = fp.readline() # status line
version, status, reason = (x.strip() for x in line.split(None, 2))
headers = parse_headers(fp)
content_length = int(headers.get("content-length"))
response_body = fp.read(content_length)
self.assertEqual(int(status), 200)
self.assertNotEqual(content_length, len(response_body))
self.assertEqual(len(response_body), content_length - 1)
self.assertEqual(response_body, tobytes("abcdefghi"))
# remote closed connection (despite keepalive header); not sure why
# first send succeeds
self.send_check_error(to_send)
self.assertRaises(ConnectionClosed, read_http, fp)
def test_send_empty_body(self):
to_send = "GET / HTTP/1.0\r\nContent-Length: 0\r\n\r\n"
to_send = tobytes(to_send)
self.connect()
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, echo = self._read_echo(fp)
self.assertline(line, "200", "OK", "HTTP/1.0")
self.assertEqual(echo.content_length, "0")
self.assertEqual(echo.body, b"")
def test_send_with_body(self):
to_send = "GET / HTTP/1.0\r\nContent-Length: 5\r\n\r\n"
to_send += "hello"
to_send = tobytes(to_send)
self.connect()
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, echo = self._read_echo(fp)
self.assertline(line, "200", "OK", "HTTP/1.0")
self.assertEqual(echo.content_length, "5")
self.assertEqual(echo.body, b"hello")
def test_filelike_nocl_http10(self):
to_send = "GET /filelike_nocl HTTP/1.0\r\n\r\n"
to_send = tobytes(to_send)
self.connect()
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, response_body = read_http(fp)
self.assertline(line, "200", "OK", "HTTP/1.0")
cl = int(headers["content-length"])
self.assertEqual(cl, len(response_body))
ct = headers["content-type"]
self.assertEqual(ct, "image/jpeg")
self.assertTrue(b"\377\330\377" in response_body)
# connection has been closed
self.send_check_error(to_send)
self.assertRaises(ConnectionClosed, read_http, fp)
def test_chunking_request_with_content(self):
control_line = b"20;\r\n" # 20 hex = 32 dec
s = b"This string has 32 characters.\r\n"
expected = s * 12
header = tobytes("GET / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n")
self.connect()
self.sock.send(header)
fp = self.sock.makefile("rb", 0)
for n in range(12):
self.sock.send(control_line)
self.sock.send(s)
self.sock.send(b"\r\n") # End the chunk
self.sock.send(b"0\r\n\r\n")
line, headers, echo = self._read_echo(fp)
self.assertline(line, "200", "OK", "HTTP/1.1")
self.assertEqual(echo.body, expected)
self.assertEqual(echo.content_length, str(len(expected)))
self.assertFalse("transfer-encoding" in headers)
def test_after_start_response_http10(self):
to_send = "GET /after_start_response HTTP/1.0\r\n\r\n"
to_send = tobytes(to_send)
self.connect()
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, response_body = read_http(fp)
self.assertline(line, "500", "Internal Server Error", "HTTP/1.0")
cl = int(headers["content-length"])
self.assertEqual(cl, len(response_body))
self.assertTrue(response_body.startswith(b"Internal Server Error"))
self.assertEqual(
sorted(headers.keys()),
["connection", "content-length", "content-type", "date", "server"],
)
self.assertEqual(headers["connection"], "close")
# connection has been closed
self.send_check_error(to_send)
self.assertRaises(ConnectionClosed, read_http, fp)
def test_notfilelike_shortcl_http11(self):
to_send = "GET /notfilelike_shortcl HTTP/1.1\r\n\r\n"
to_send = tobytes(to_send)
self.connect()
for t in range(0, 2):
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, response_body = read_http(fp)
self.assertline(line, "200", "OK", "HTTP/1.1")
cl = int(headers["content-length"])
self.assertEqual(cl, 1)
self.assertEqual(cl, len(response_body))
ct = headers["content-type"]
self.assertEqual(ct, "image/jpeg")
self.assertTrue(b"\377" in response_body)
# connection has not been closed
# check server doesnt close connection when body is too long
# for cl header
to_send = tobytes(
"GET /long_body HTTP/1.0\r\n"
"Connection: Keep-Alive\r\n"
"Content-Length: 0\r\n"
"\r\n"
)
self.connect()
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, response_body = read_http(fp)
content_length = int(headers.get("content-length")) or None
self.assertEqual(content_length, 9)
self.assertEqual(content_length, len(response_body))
self.assertEqual(response_body, tobytes("abcdefghi"))
# remote does not close connection (keepalive header)
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, response_body = read_http(fp)
self.assertline(line, "200", "OK", "HTTP/1.0")
def test_large_body(self):
# 1024 characters.
body = "This string has 32 characters.\r\n" * 32
s = tobytes(
"GET / HTTP/1.0\r\nContent-Length: %d\r\n\r\n%s" % (len(body), body)
)
self.connect()
self.sock.send(s)
fp = self.sock.makefile("rb", 0)
line, headers, echo = self._read_echo(fp)
self.assertline(line, "200", "OK", "HTTP/1.0")
self.assertEqual(echo.content_length, "1024")
self.assertEqual(echo.body, tobytes(body))
def test_filelike_shortcl_http11(self):
to_send = "GET /filelike_shortcl HTTP/1.1\r\n\r\n"
to_send = tobytes(to_send)
self.connect()
for t in range(0, 2):
self.sock.send(to_send)
fp = self.sock.makefile("rb", 0)
line, headers, response_body = read_http(fp)
self.assertline(line, "200", "OK", "HTTP/1.1")
cl = int(headers["content-length"])
self.assertEqual(cl, 1)
self.assertEqual(cl, len(response_body))
ct = headers["content-type"]
self.assertEqual(ct, "image/jpeg")
self.assertTrue(b"\377" in response_body)
# connection has not been closed