88import functools
99import json
1010import os
11+ import socket
1112import ssl
1213import subprocess
1314import sys
@@ -108,11 +109,19 @@ def respond(self):
108109 return super (HelloWorldGateway , self ).respond ()
109110
110111
111- def make_tls_http_server (bind_addr , ssl_adapter , request ):
112+ def make_tls_http_server (
113+ bind_addr ,
114+ ssl_adapter ,
115+ request ,
116+ minthreads = 1 ,
117+ maxthreads = - 1 ,
118+ ):
112119 """Create and start an HTTP server bound to ``bind_addr``."""
113120 httpserver = HTTPServer (
114121 bind_addr = bind_addr ,
115122 gateway = HelloWorldGateway ,
123+ minthreads = minthreads ,
124+ maxthreads = maxthreads ,
116125 )
117126 # httpserver.gateway = HelloWorldGateway
118127 httpserver .ssl_adapter = ssl_adapter
@@ -133,6 +142,17 @@ def tls_http_server(request):
133142 return functools .partial (make_tls_http_server , request = request )
134143
135144
145+ @pytest .fixture
146+ def tls_single_thread_http_server (request ):
147+ """Provision a server creator as a fixture."""
148+ return functools .partial (
149+ make_tls_http_server ,
150+ request = request ,
151+ minthreads = 1 ,
152+ maxthreads = 1 ,
153+ )
154+
155+
136156@pytest .fixture
137157def ca ():
138158 """Provide a certificate authority via fixture."""
@@ -729,3 +749,65 @@ def test_http_over_https_error(
729749 format (** locals ())
730750 )
731751 assert expected_error_text in err_text
752+
753+
754+ @pytest .mark .parametrize ( # noqa: C901 # FIXME
755+ 'adapter_type' ,
756+ (
757+ 'builtin' ,
758+ 'pyopenssl' ,
759+ ),
760+ )
761+ @pytest .mark .parametrize (
762+ 'ip_addr' ,
763+ (
764+ ANY_INTERFACE_IPV4 ,
765+ pytest .param (ANY_INTERFACE_IPV6 , marks = missing_ipv6 ),
766+ ),
767+ )
768+ def test_server_timeout_before_content_error (
769+ tls_single_thread_http_server , adapter_type ,
770+ ca , ip_addr ,
771+ tls_certificate ,
772+ tls_certificate_chain_pem_path ,
773+ tls_certificate_private_key_pem_path ,
774+ ):
775+ """Ensure connection beyond timeout without sending content is handled."""
776+ # disable some flaky tests
777+ # https://github.com/cherrypy/cheroot/issues/225
778+ issue_225 = (
779+ IS_MACOS
780+ and adapter_type == 'builtin'
781+ )
782+ if issue_225 :
783+ pytest .xfail ('Test fails in Travis-CI' )
784+
785+ tls_adapter_cls = get_ssl_adapter_class (name = adapter_type )
786+ tls_adapter = tls_adapter_cls (
787+ tls_certificate_chain_pem_path , tls_certificate_private_key_pem_path ,
788+ )
789+ if adapter_type == 'pyopenssl' :
790+ tls_adapter .context = tls_adapter .get_context ()
791+
792+ tls_certificate .configure_cert (tls_adapter .context )
793+
794+ interface , _host , port = _get_conn_data (ip_addr )
795+ tlshttpserver = \
796+ tls_single_thread_http_server ((interface , port ), tls_adapter )
797+ tlshttpserver .timeout = 1
798+ interface , host , port = _get_conn_data (
799+ tlshttpserver .bind_addr ,
800+ )
801+
802+ fqdn = interface
803+ if ip_addr is ANY_INTERFACE_IPV6 :
804+ fqdn = '[{fqdn}]' .format (** locals ())
805+
806+ with socket .socket (socket .AF_INET6 , socket .SOCK_STREAM ) as s :
807+ s .connect ((ip_addr , port ))
808+ time .sleep (3 )
809+
810+ # second attempt will fail if server is dropping threads
811+ with socket .socket (socket .AF_INET6 , socket .SOCK_STREAM ) as s :
812+ s .connect ((ip_addr , port ))
813+ time .sleep (3 )
0 commit comments