Dive into secure and efficient coding practices with our curated list of the top 10 examples showcasing 'waitress' in functional components in Python. Our advanced machine learning engine meticulously scans each line of code, cross-referencing millions of open source libraries to ensure your implementation is not just functional, but also robust and secure. Elevate your React applications to new heights by mastering the art of handling side effects, API calls, and asynchronous operations with confidence and precision.
"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