FM / test

/opt/imh-python/lib/python3.9/site-packages/cherrypy/test/

/opt/imh-python/lib/python3.9/site-packages/cherrypy/test

Name	Size	Mode	Actions
static/	-	0755	rm
__pycache__/	-	0755	rm
benchmark.py	12563	0644	edit dl rm
checkerdemo.py	1861	0644	edit dl rm
fastcgi.conf	686	0644	edit dl rm
fcgi.conf	486	0644	edit dl rm
helper.py	16369	0644	edit dl rm
logtest.py	8151	0644	edit dl rm
modfastcgi.py	4652	0644	edit dl rm
modfcgid.py	4237	0644	edit dl rm
modpy.py	4978	0644	edit dl rm
modwsgi.py	4834	0644	edit dl rm
sessiondemo.py	5425	0644	edit dl rm
style.css	17	0644	edit dl rm
test.pem	2254	0644	edit dl rm
test_auth_basic.py	4499	0644	edit dl rm
test_auth_digest.py	4454	0644	edit dl rm
test_bus.py	9960	0644	edit dl rm
test_caching.py	14386	0644	edit dl rm
test_config.py	8836	0644	edit dl rm
test_config_server.py	4037	0644	edit dl rm
test_conn.py	30744	0644	edit dl rm
test_core.py	30408	0644	edit dl rm
test_dynamicobjectmapping.py	12331	0644	edit dl rm
test_encoding.py	17535	0644	edit dl rm
test_etags.py	3093	0644	edit dl rm
test_http.py	10939	0644	edit dl rm
test_httputil.py	2412	0644	edit dl rm
test_iterator.py	5724	0644	edit dl rm
test_json.py	2860	0644	edit dl rm
test_logging.py	8315	0644	edit dl rm
test_mime.py	4538	0644	edit dl rm
test_misc_tools.py	7094	0644	edit dl rm
test_native.py	971	0644	edit dl rm
test_objectmapping.py	14504	0644	edit dl rm
test_params.py	1862	0644	edit dl rm
test_plugins.py	334	0644	edit dl rm
test_proxy.py	5630	0644	edit dl rm
test_refleaks.py	1555	0644	edit dl rm
test_request_obj.py	37097	0644	edit dl rm
test_routes.py	2583	0644	edit dl rm
test_session.py	17949	0644	edit dl rm
test_sessionauthenticate.py	2013	0644	edit dl rm
test_states.py	16703	0644	edit dl rm
test_static.py	16702	0644	edit dl rm
test_tools.py	17851	0644	edit dl rm
test_tutorials.py	6964	0644	edit dl rm
test_virtualhost.py	4021	0644	edit dl rm
test_wsgiapps.py	3997	0644	edit dl rm
test_wsgi_ns.py	2812	0644	edit dl rm
test_wsgi_unix_socket.py	2228	0644	edit dl rm
test_wsgi_vhost.py	1034	0644	edit dl rm
test_xmlrpc.py	4584	0644	edit dl rm
webtest.py	262	0644	edit dl rm
_test_decorators.py	951	0644	edit dl rm
_test_states_demo.py	1876	0644	edit dl rm
__init__.py	396	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/cherrypy/test/test_conn.py (30744B)

"""Tests for TCP connection handling, including proper and timely close."""

import errno
import socket
import sys
import time
import urllib.parse
from http.client import BadStatusLine, HTTPConnection, NotConnected

from cheroot.test import webtest

import cherrypy
from cherrypy._cpcompat import HTTPSConnection, ntob, tonative
from cherrypy.test import helper

timeout = 1
pov = 'pPeErRsSiIsStTeEnNcCeE oOfF vViIsSiIoOnN'

def setup_server():

def raise500():
        raise cherrypy.HTTPError(500)

class Root:

@cherrypy.expose
        def index(self):
            return pov
        page1 = index
        page2 = index
        page3 = index

@cherrypy.expose
        def hello(self):
            return 'Hello, world!'

@cherrypy.expose
        def timeout(self, t):
            return str(cherrypy.server.httpserver.timeout)

@cherrypy.expose
        @cherrypy.config(**{'response.stream': True})
        def stream(self, set_cl=False):
            if set_cl:
                cherrypy.response.headers['Content-Length'] = 10

def content():
                for x in range(10):
                    yield str(x)

return content()

@cherrypy.expose
        def error(self, code=500):
            raise cherrypy.HTTPError(code)

@cherrypy.expose
        def upload(self):
            if not cherrypy.request.method == 'POST':
                raise AssertionError("'POST' != request.method %r" %
                                     cherrypy.request.method)
            return "thanks for '%s'" % cherrypy.request.body.read()

@cherrypy.expose
        def custom(self, response_code):
            cherrypy.response.status = response_code
            return 'Code = %s' % response_code

@cherrypy.expose
        @cherrypy.config(**{'hooks.on_start_resource': raise500})
        def err_before_read(self):
            return 'ok'

@cherrypy.expose
        def one_megabyte_of_a(self):
            return ['a' * 1024] * 1024

@cherrypy.expose
        # Turn off the encoding tool so it doens't collapse
        # our response body and reclaculate the Content-Length.
        @cherrypy.config(**{'tools.encode.on': False})
        def custom_cl(self, body, cl):
            cherrypy.response.headers['Content-Length'] = cl
            if not isinstance(body, list):
                body = [body]
            newbody = []
            for chunk in body:
                if isinstance(chunk, str):
                    chunk = chunk.encode('ISO-8859-1')
                newbody.append(chunk)
            return newbody

cherrypy.tree.mount(Root())
    cherrypy.config.update({
        'server.max_request_body_size': 1001,
        'server.socket_timeout': timeout,
    })

class ConnectionCloseTests(helper.CPWebCase):
    setup_server = staticmethod(setup_server)

def test_HTTP11(self):
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

self.persistent = True

# Make the first request and assert there's no "Connection: close".
        self.getPage('/')
        self.assertStatus('200 OK')
        self.assertBody(pov)
        self.assertNoHeader('Connection')

# Make another request on the same connection.
        self.getPage('/page1')
        self.assertStatus('200 OK')
        self.assertBody(pov)
        self.assertNoHeader('Connection')

# Test client-side close.
        self.getPage('/page2', headers=[('Connection', 'close')])
        self.assertStatus('200 OK')
        self.assertBody(pov)
        self.assertHeader('Connection', 'close')

# Make another request on the same connection, which should error.
        self.assertRaises(NotConnected, self.getPage, '/')

def test_Streaming_no_len(self):
        try:
            self._streaming(set_cl=False)
        finally:
            try:
                self.HTTP_CONN.close()
            except (TypeError, AttributeError):
                pass

def test_Streaming_with_len(self):
        try:
            self._streaming(set_cl=True)
        finally:
            try:
                self.HTTP_CONN.close()
            except (TypeError, AttributeError):
                pass

def _streaming(self, set_cl):
        if cherrypy.server.protocol_version == 'HTTP/1.1':
            self.PROTOCOL = 'HTTP/1.1'

self.persistent = True

# Make the first request and assert there's no "Connection: close".
            self.getPage('/')
            self.assertStatus('200 OK')
            self.assertBody(pov)
            self.assertNoHeader('Connection')

# Make another, streamed request on the same connection.
            if set_cl:
                # When a Content-Length is provided, the content should stream
                # without closing the connection.
                self.getPage('/stream?set_cl=Yes')
                self.assertHeader('Content-Length')
                self.assertNoHeader('Connection', 'close')
                self.assertNoHeader('Transfer-Encoding')

self.assertStatus('200 OK')
                self.assertBody('0123456789')
            else:
                # When no Content-Length response header is provided,
                # streamed output will either close the connection, or use
                # chunked encoding, to determine transfer-length.
                self.getPage('/stream')
                self.assertNoHeader('Content-Length')
                self.assertStatus('200 OK')
                self.assertBody('0123456789')

chunked_response = False
                for k, v in self.headers:
                    if k.lower() == 'transfer-encoding':
                        if str(v) == 'chunked':
                            chunked_response = True

if chunked_response:
                    self.assertNoHeader('Connection', 'close')
                else:
                    self.assertHeader('Connection', 'close')

# Make another request on the same connection, which should
                    # error.
                    self.assertRaises(NotConnected, self.getPage, '/')

# Try HEAD. See
                # https://github.com/cherrypy/cherrypy/issues/864.
                self.getPage('/stream', method='HEAD')
                self.assertStatus('200 OK')
                self.assertBody('')
                self.assertNoHeader('Transfer-Encoding')
        else:
            self.PROTOCOL = 'HTTP/1.0'

self.persistent = True

# Make the first request and assert Keep-Alive.
            self.getPage('/', headers=[('Connection', 'Keep-Alive')])
            self.assertStatus('200 OK')
            self.assertBody(pov)
            self.assertHeader('Connection', 'Keep-Alive')

# Make another, streamed request on the same connection.
            if set_cl:
                # When a Content-Length is provided, the content should
                # stream without closing the connection.
                self.getPage('/stream?set_cl=Yes',
                             headers=[('Connection', 'Keep-Alive')])
                self.assertHeader('Content-Length')
                self.assertHeader('Connection', 'Keep-Alive')
                self.assertNoHeader('Transfer-Encoding')
                self.assertStatus('200 OK')
                self.assertBody('0123456789')
            else:
                # When a Content-Length is not provided,
                # the server should close the connection.
                self.getPage('/stream', headers=[('Connection', 'Keep-Alive')])
                self.assertStatus('200 OK')
                self.assertBody('0123456789')

self.assertNoHeader('Content-Length')
                self.assertNoHeader('Connection', 'Keep-Alive')
                self.assertNoHeader('Transfer-Encoding')

# Make another request on the same connection, which should
                # error.
                self.assertRaises(NotConnected, self.getPage, '/')

def test_HTTP10_KeepAlive(self):
        self.PROTOCOL = 'HTTP/1.0'
        if self.scheme == 'https':
            self.HTTP_CONN = HTTPSConnection
        else:
            self.HTTP_CONN = HTTPConnection

# Test a normal HTTP/1.0 request.
        self.getPage('/page2')
        self.assertStatus('200 OK')
        self.assertBody(pov)
        # Apache, for example, may emit a Connection header even for HTTP/1.0
        # self.assertNoHeader("Connection")

# Test a keep-alive HTTP/1.0 request.
        self.persistent = True

self.getPage('/page3', headers=[('Connection', 'Keep-Alive')])
        self.assertStatus('200 OK')
        self.assertBody(pov)
        self.assertHeader('Connection', 'Keep-Alive')

# Remove the keep-alive header again.
        self.getPage('/page3')
        self.assertStatus('200 OK')
        self.assertBody(pov)
        # Apache, for example, may emit a Connection header even for HTTP/1.0
        # self.assertNoHeader("Connection")

class PipelineTests(helper.CPWebCase):
    setup_server = staticmethod(setup_server)

def test_HTTP11_Timeout(self):
        # If we timeout without sending any data,
        # the server will close the conn with a 408.
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

# Connect but send nothing.
        self.persistent = True
        conn = self.HTTP_CONN
        conn.auto_open = False
        conn.connect()

# Wait for our socket timeout
        time.sleep(timeout * 2)

# The request should have returned 408 already.
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.assertEqual(response.status, 408)
        conn.close()

# Connect but send half the headers only.
        self.persistent = True
        conn = self.HTTP_CONN
        conn.auto_open = False
        conn.connect()
        conn.send(b'GET /hello HTTP/1.1')
        conn.send(('Host: %s' % self.HOST).encode('ascii'))

# Wait for our socket timeout
        time.sleep(timeout * 2)

# The conn should have already sent 408.
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.assertEqual(response.status, 408)
        conn.close()

def test_HTTP11_Timeout_after_request(self):
        # If we timeout after at least one request has succeeded,
        # the server will close the conn without 408.
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

# Make an initial request
        self.persistent = True
        conn = self.HTTP_CONN
        conn.putrequest('GET', '/timeout?t=%s' % timeout, skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.endheaders()
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.assertEqual(response.status, 200)
        self.body = response.read()
        self.assertBody(str(timeout))

# Make a second request on the same socket
        conn._output(b'GET /hello HTTP/1.1')
        conn._output(ntob('Host: %s' % self.HOST, 'ascii'))
        conn._send_output()
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.assertEqual(response.status, 200)
        self.body = response.read()
        self.assertBody('Hello, world!')

# Wait for our socket timeout
        time.sleep(timeout * 2)

# Make another request on the same socket, which should error
        conn._output(b'GET /hello HTTP/1.1')
        conn._output(ntob('Host: %s' % self.HOST, 'ascii'))
        conn._send_output()
        response = conn.response_class(conn.sock, method='GET')
        msg = (
            "Writing to timed out socket didn't fail as it should have: %s")
        try:
            response.begin()
        except Exception:
            if not isinstance(sys.exc_info()[1],
                              (socket.error, BadStatusLine)):
                self.fail(msg % sys.exc_info()[1])
        else:
            if response.status != 408:
                self.fail(msg % response.read())

conn.close()

# Make another request on a new socket, which should work
        self.persistent = True
        conn = self.HTTP_CONN
        conn.putrequest('GET', '/', skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.endheaders()
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.assertEqual(response.status, 200)
        self.body = response.read()
        self.assertBody(pov)

# Make another request on the same socket,
        # but timeout on the headers
        conn.send(b'GET /hello HTTP/1.1')
        # Wait for our socket timeout
        time.sleep(timeout * 2)
        response = conn.response_class(conn.sock, method='GET')
        try:
            response.begin()
        except Exception:
            if not isinstance(sys.exc_info()[1],
                              (socket.error, BadStatusLine)):
                self.fail(msg % sys.exc_info()[1])
        else:
            if response.status != 408:
                self.fail(msg % response.read())

conn.close()

# Retry the request on a new connection, which should work
        self.persistent = True
        conn = self.HTTP_CONN
        conn.putrequest('GET', '/', skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.endheaders()
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.assertEqual(response.status, 200)
        self.body = response.read()
        self.assertBody(pov)
        conn.close()

def test_HTTP11_pipelining(self):
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

# Test pipelining. httplib doesn't support this directly.
        self.persistent = True
        conn = self.HTTP_CONN

# Put request 1
        conn.putrequest('GET', '/hello', skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.endheaders()

for trial in range(5):
            # Put next request
            conn._output(b'GET /hello HTTP/1.1')
            conn._output(ntob('Host: %s' % self.HOST, 'ascii'))
            conn._send_output()

# Retrieve previous response
            response = conn.response_class(conn.sock, method='GET')
            # there is a bug in python3 regarding the buffering of
            # ``conn.sock``. Until that bug get's fixed we will
            # monkey patch the ``response`` instance.
            # https://bugs.python.org/issue23377
            response.fp = conn.sock.makefile('rb', 0)
            response.begin()
            body = response.read(13)
            self.assertEqual(response.status, 200)
            self.assertEqual(body, b'Hello, world!')

# Retrieve final response
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        body = response.read()
        self.assertEqual(response.status, 200)
        self.assertEqual(body, b'Hello, world!')

conn.close()

def test_100_Continue(self):
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

self.persistent = True
        conn = self.HTTP_CONN

# Try a page without an Expect request header first.
        # Note that httplib's response.begin automatically ignores
        # 100 Continue responses, so we must manually check for it.
        try:
            conn.putrequest('POST', '/upload', skip_host=True)
            conn.putheader('Host', self.HOST)
            conn.putheader('Content-Type', 'text/plain')
            conn.putheader('Content-Length', '4')
            conn.endheaders()
            conn.send(ntob("d'oh"))
            response = conn.response_class(conn.sock, method='POST')
            version, status, reason = response._read_status()
            self.assertNotEqual(status, 100)
        finally:
            conn.close()

# Now try a page with an Expect header...
        try:
            conn.connect()
            conn.putrequest('POST', '/upload', skip_host=True)
            conn.putheader('Host', self.HOST)
            conn.putheader('Content-Type', 'text/plain')
            conn.putheader('Content-Length', '17')
            conn.putheader('Expect', '100-continue')
            conn.endheaders()
            response = conn.response_class(conn.sock, method='POST')

# ...assert and then skip the 100 response
            version, status, reason = response._read_status()
            self.assertEqual(status, 100)
            while True:
                line = response.fp.readline().strip()
                if line:
                    self.fail(
                        '100 Continue should not output any headers. Got %r' %
                        line)
                else:
                    break

# ...send the body
            body = b'I am a small file'
            conn.send(body)

# ...get the final response
            response.begin()
            self.status, self.headers, self.body = webtest.shb(response)
            self.assertStatus(200)
            self.assertBody("thanks for '%s'" % body)
        finally:
            conn.close()

class ConnectionTests(helper.CPWebCase):
    setup_server = staticmethod(setup_server)

def test_readall_or_close(self):
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

if self.scheme == 'https':
            self.HTTP_CONN = HTTPSConnection
        else:
            self.HTTP_CONN = HTTPConnection

# Test a max of 0 (the default) and then reset to what it was above.
        old_max = cherrypy.server.max_request_body_size
        for new_max in (0, old_max):
            cherrypy.server.max_request_body_size = new_max

self.persistent = True
            conn = self.HTTP_CONN

# Get a POST page with an error
            conn.putrequest('POST', '/err_before_read', skip_host=True)
            conn.putheader('Host', self.HOST)
            conn.putheader('Content-Type', 'text/plain')
            conn.putheader('Content-Length', '1000')
            conn.putheader('Expect', '100-continue')
            conn.endheaders()
            response = conn.response_class(conn.sock, method='POST')

# ...assert and then skip the 100 response
            version, status, reason = response._read_status()
            self.assertEqual(status, 100)
            while True:
                skip = response.fp.readline().strip()
                if not skip:
                    break

# ...send the body
            conn.send(ntob('x' * 1000))

# ...get the final response
            response.begin()
            self.status, self.headers, self.body = webtest.shb(response)
            self.assertStatus(500)

# Now try a working page with an Expect header...
            conn._output(b'POST /upload HTTP/1.1')
            conn._output(ntob('Host: %s' % self.HOST, 'ascii'))
            conn._output(b'Content-Type: text/plain')
            conn._output(b'Content-Length: 17')
            conn._output(b'Expect: 100-continue')
            conn._send_output()
            response = conn.response_class(conn.sock, method='POST')

# ...send the body
            body = b'I am a small file'
            conn.send(body)

def test_No_Message_Body(self):
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

# Set our HTTP_CONN to an instance so it persists between requests.
        self.persistent = True

# Make a 204 request on the same connection.
        self.getPage('/custom/204')
        self.assertStatus(204)
        self.assertNoHeader('Content-Length')
        self.assertBody('')
        self.assertNoHeader('Connection')

# Make a 304 request on the same connection.
        self.getPage('/custom/304')
        self.assertStatus(304)
        self.assertNoHeader('Content-Length')
        self.assertBody('')
        self.assertNoHeader('Connection')

def test_Chunked_Encoding(self):
        if cherrypy.server.protocol_version != 'HTTP/1.1':
            return self.skip()

if (hasattr(self, 'harness') and
                'modpython' in self.harness.__class__.__name__.lower()):
            # mod_python forbids chunked encoding
            return self.skip()

self.PROTOCOL = 'HTTP/1.1'

# Set our HTTP_CONN to an instance so it persists between requests.
        self.persistent = True
        conn = self.HTTP_CONN

# Try a normal chunked request (with extensions)
        body = ntob('8;key=value\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n'
                    'Content-Type: application/json\r\n'
                    '\r\n')
        conn.putrequest('POST', '/upload', skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.putheader('Transfer-Encoding', 'chunked')
        conn.putheader('Trailer', 'Content-Type')
        # Note that this is somewhat malformed:
        # we shouldn't be sending Content-Length.
        # RFC 2616 says the server should ignore it.
        conn.putheader('Content-Length', '3')
        conn.endheaders()
        conn.send(body)
        response = conn.getresponse()
        self.status, self.headers, self.body = webtest.shb(response)
        self.assertStatus('200 OK')
        self.assertBody("thanks for '%s'" % b'xx\r\nxxxxyyyyy')

# Try a chunked request that exceeds server.max_request_body_size.
        # Note that the delimiters and trailer are included.
        body = ntob('3e3\r\n' + ('x' * 995) + '\r\n0\r\n\r\n')
        conn.putrequest('POST', '/upload', skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.putheader('Transfer-Encoding', 'chunked')
        conn.putheader('Content-Type', 'text/plain')
        # Chunked requests don't need a content-length
        # #        conn.putheader("Content-Length", len(body))
        conn.endheaders()
        conn.send(body)
        response = conn.getresponse()
        self.status, self.headers, self.body = webtest.shb(response)
        self.assertStatus(413)
        conn.close()

def test_Content_Length_in(self):
        # Try a non-chunked request where Content-Length exceeds
        # server.max_request_body_size. Assert error before body send.
        self.persistent = True
        conn = self.HTTP_CONN
        conn.putrequest('POST', '/upload', skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.putheader('Content-Type', 'text/plain')
        conn.putheader('Content-Length', '9999')
        conn.endheaders()
        response = conn.getresponse()
        self.status, self.headers, self.body = webtest.shb(response)
        self.assertStatus(413)
        self.assertBody('The entity sent with the request exceeds '
                        'the maximum allowed bytes.')
        conn.close()

def test_Content_Length_out_preheaders(self):
        # Try a non-chunked response where Content-Length is less than
        # the actual bytes in the response body.
        self.persistent = True
        conn = self.HTTP_CONN
        conn.putrequest('GET', '/custom_cl?body=I+have+too+many+bytes&cl=5',
                        skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.endheaders()
        response = conn.getresponse()
        self.status, self.headers, self.body = webtest.shb(response)
        self.assertStatus(500)
        self.assertBody(
            'The requested resource returned more bytes than the '
            'declared Content-Length.')
        conn.close()

def test_Content_Length_out_postheaders(self):
        # Try a non-chunked response where Content-Length is less than
        # the actual bytes in the response body.
        self.persistent = True
        conn = self.HTTP_CONN
        conn.putrequest(
            'GET', '/custom_cl?body=I+too&body=+have+too+many&cl=5',
            skip_host=True)
        conn.putheader('Host', self.HOST)
        conn.endheaders()
        response = conn.getresponse()
        self.status, self.headers, self.body = webtest.shb(response)
        self.assertStatus(200)
        self.assertBody('I too')
        conn.close()

def test_598(self):
        tmpl = '{scheme}://{host}:{port}/one_megabyte_of_a/'
        url = tmpl.format(
            scheme=self.scheme,
            host=self.HOST,
            port=self.PORT,
        )
        remote_data_conn = urllib.request.urlopen(url)
        buf = remote_data_conn.read(512)
        time.sleep(timeout * 0.6)
        remaining = (1024 * 1024) - 512
        while remaining:
            data = remote_data_conn.read(remaining)
            if not data:
                break
            else:
                buf += data
            remaining -= len(data)

self.assertEqual(len(buf), 1024 * 1024)
        self.assertEqual(buf, ntob('a' * 1024 * 1024))
        self.assertEqual(remaining, 0)
        remote_data_conn.close()

def setup_upload_server():

class Root:
        @cherrypy.expose
        def upload(self):
            if not cherrypy.request.method == 'POST':
                raise AssertionError("'POST' != request.method %r" %
                                     cherrypy.request.method)
            return "thanks for '%s'" % tonative(cherrypy.request.body.read())

cherrypy.tree.mount(Root())
    cherrypy.config.update({
        'server.max_request_body_size': 1001,
        'server.socket_timeout': 10,
        'server.accepted_queue_size': 5,
        'server.accepted_queue_timeout': 0.1,
    })

reset_names = 'ECONNRESET', 'WSAECONNRESET'
socket_reset_errors = [
    getattr(errno, name)
    for name in reset_names
    if hasattr(errno, name)
]
'reset error numbers available on this platform'

socket_reset_errors += [
    # Python 3.5 raises an http.client.RemoteDisconnected
    # with this message
    'Remote end closed connection without response',
]

class LimitedRequestQueueTests(helper.CPWebCase):
    setup_server = staticmethod(setup_upload_server)

def test_queue_full(self):
        conns = []
        overflow_conn = None

try:
            # Make 15 initial requests and leave them open, which should use
            # all of wsgiserver's WorkerThreads and fill its Queue.
            for i in range(15):
                conn = self.HTTP_CONN(self.HOST, self.PORT)
                conn.putrequest('POST', '/upload', skip_host=True)
                conn.putheader('Host', self.HOST)
                conn.putheader('Content-Type', 'text/plain')
                conn.putheader('Content-Length', '4')
                conn.endheaders()
                conns.append(conn)

# Now try a 16th conn, which should be closed by the
            # server immediately.
            overflow_conn = self.HTTP_CONN(self.HOST, self.PORT)
            # Manually connect since httplib won't let us set a timeout
            for res in socket.getaddrinfo(self.HOST, self.PORT, 0,
                                          socket.SOCK_STREAM):
                af, socktype, proto, canonname, sa = res
                overflow_conn.sock = socket.socket(af, socktype, proto)
                overflow_conn.sock.settimeout(5)
                overflow_conn.sock.connect(sa)
                break

overflow_conn.putrequest('GET', '/', skip_host=True)
            overflow_conn.putheader('Host', self.HOST)
            overflow_conn.endheaders()
            response = overflow_conn.response_class(
                overflow_conn.sock,
                method='GET',
            )
            try:
                response.begin()
            except socket.error as exc:
                if exc.args[0] in socket_reset_errors:
                    pass  # Expected.
                else:
                    tmpl = (
                        'Overflow conn did not get RST. '
                        'Got {exc.args!r} instead'
                    )
                    raise AssertionError(tmpl.format(**locals()))
            except BadStatusLine:
                # This is a special case in OS X. Linux and Windows will
                # RST correctly.
                assert sys.platform == 'darwin'
            else:
                raise AssertionError('Overflow conn did not get RST ')
        finally:
            for conn in conns:
                conn.send(b'done')
                response = conn.response_class(conn.sock, method='POST')
                response.begin()
                self.body = response.read()
                self.assertBody("thanks for 'done'")
                self.assertEqual(response.status, 200)
                conn.close()
            if overflow_conn:
                overflow_conn.close()

class BadRequestTests(helper.CPWebCase):
    setup_server = staticmethod(setup_server)

def test_No_CRLF(self):
        self.persistent = True

conn = self.HTTP_CONN
        conn.send(b'GET /hello HTTP/1.1\n\n')
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.body = response.read()
        self.assertBody('HTTP requires CRLF terminators')
        conn.close()

conn.connect()
        conn.send(b'GET /hello HTTP/1.1\r\n\n')
        response = conn.response_class(conn.sock, method='GET')
        response.begin()
        self.body = response.read()
        self.assertBody('HTTP requires CRLF terminators')
        conn.close()

cmd: