<commit_before>from cherrypy.test import test
test.prefer_parent_path()

import httplib
import socket
import sys
import time
timeout = 0.1


import cherrypy
from cherrypy.test import webtest


pov = 'pPeErRsSiIsStTeEnNcCeE oOfF vViIsSiIoOnN'

def setup_server():
    class Root:
        
        def index(self):
            return pov
        index.exposed = True
        page1 = index
        page2 = index
        page3 = index
        
        def hello(self):
            return "Hello, world!"
        hello.exposed = True
        
        def stream(self, set_cl=False):
            if set_cl:
                cherrypy.response.headers['Content-Length'] = 10
            
            def content():
                for x in xrange(10):
                    yield str(x)
            
            return content()
        stream.exposed = True
        stream._cp_config = {'response.stream': True}
        
        def upload(self):
            return ("thanks for '%s' (%s)" %
                    (cherrypy.request.body.read(),
                     cherrypy.request.headers['Content-Type']))
        upload.exposed = True
        
        def custom(self, response_code):
            cherrypy.response.status = response_code
            return "Code = %s" % response_code
        custom.exposed = True
    
    cherrypy.tree.mount(Root())
    cherrypy.config.update({
        'server.max_request_body_size': 100,
        'environment': 'test_suite',
        })


from cherrypy.test import helper

class ConnectionTests(helper.CPWebCase):
    
    def test_HTTP11(self):
        if cherrypy.server.protocol_version != "HTTP/1.1":
            print "skipped ",
            return
        
        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(httplib.NotConnected, self.getPage, "/")
    
    def test_Streaming_no_len(self):
        self._streaming(set_cl=False)
    
    def test_Streaming_with_len(self):
        self._streaming(set_cl=True)
    
    def _streaming(self, set_cl):
        if cherrypy.server.protocol_version != "HTTP/1.1":
            print "skipped ",
            return
        
        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")
        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")
            
            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(httplib.NotConnected, self.getPage, "/")
        
        self.assertStatus('200 OK')
        self.assertBody('0123456789')
    
    def test_HTTP11_Timeout(self):
        if cherrypy.server.protocol_version != "HTTP/1.1":
            print "skipped ",
            return
        
        old_timeout = None
        try:
            httpserver = cherrypy.server.httpservers.keys()[0]
            old_timeout = httpserver.timeout
        except (AttributeError, IndexError):
            print "skipped ",
            return
        
        try:
            httpserver.timeout = timeout
            self.PROTOCOL = "HTTP/1.1"
            
            # Make an initial request
            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 a second request on the same socket
            conn._output('GET /hello HTTP/1.1')
            conn._output("Host: %s" % self.HOST)
            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('GET /hello HTTP/1.1')
            conn._output("Host: %s" % self.HOST)
            conn._send_output()
            response = conn.response_class(conn.sock, method="GET")
            try:
                response.begin()
            except:
                if not isinstance(sys.exc_info()[1],
                                  (socket.error, httplib.BadStatusLine)):
                    self.fail("Writing to timed out socket didn't fail"
                              " as it should have: %s" % sys.exc_info()[1])
            else:<commit_msg>Partial fix for #625 (SSL: Writing to timed out socket didn't fail as it should have). This fixes test_conn by simply using a larger timeout. But Demo.stream is still broken.<commit_after>from cherrypy.test import test
test.prefer_parent_path()

import httplib
import socket
import sys
import time
timeout = 0.1


import cherrypy
from cherrypy.test import webtest


pov = 'pPeErRsSiIsStTeEnNcCeE oOfF vViIsSiIoOnN'

def setup_server():
    class Root:
        
        def index(self):
            return pov
        index.exposed = True
        page1 = index
        page2 = index
        page3 = index
        
        def hello(self):
            return "Hello, world!"
        hello.exposed = True
        
        def stream(self, set_cl=False):
            if set_cl:
                cherrypy.response.headers['Content-Length'] = 10
            
            def content():
                for x in xrange(10):
                    yield str(x)
            
            return content()
        stream.exposed = True
        stream._cp_config = {'response.stream': True}
        
        def upload(self):
            return ("thanks for '%s' (%s)" %
                    (cherrypy.request.body.read(),
                     cherrypy.request.headers['Content-Type']))
        upload.exposed = True
        
        def custom(self, response_code):
            cherrypy.response.status = response_code
            return "Code = %s" % response_code
        custom.exposed = True
    
    cherrypy.tree.mount(Root())
    cherrypy.config.update({
        'server.max_request_body_size': 100,
        'environment': 'test_suite',
        })


from cherrypy.test import helper

class ConnectionTests(helper.CPWebCase):
    
    def test_HTTP11(self):
        if cherrypy.server.protocol_version != "HTTP/1.1":
            print "skipped ",
            return
        
        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(httplib.NotConnected, self.getPage, "/")
    
    def test_Streaming_no_len(self):
        self._streaming(set_cl=False)
    
    def test_Streaming_with_len(self):
        self._streaming(set_cl=True)
    
    def _streaming(self, set_cl):
        if cherrypy.server.protocol_version != "HTTP/1.1":
            print "skipped ",
            return
        
        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")
        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")
            
            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(httplib.NotConnected, self.getPage, "/")
        
        self.assertStatus('200 OK')
        self.assertBody('0123456789')
    
    def test_HTTP11_Timeout(self):
        if cherrypy.server.protocol_version != "HTTP/1.1":
            print "skipped ",
            return
        
        old_timeout = None
        try:
            httpserver = cherrypy.server.httpservers.keys()[0]
            old_timeout = httpserver.timeout
        except (AttributeError, IndexError):
            print "skipped ",
            return
        
        try:
            httpserver.timeout = timeout
            self.PROTOCOL = "HTTP/1.1"
            
            # Make an initial request
            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 a second request on the same socket
            conn._output('GET /hello HTTP/1.1')
            conn._output("Host: %s" % self.HOST)
            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!")
            
            if self.scheme == "https":
                # Adjust the timeout up to cover the higher timeouts of SSL
                # (the wsgiserver ssl_timeout default is 3 seconds)
                tout = 4
            else:
                tout = timeout * 2
            
            # Wait for our socket timeout
            time.sleep(tout)
            
            # Make another request on the same socket, which should error
            conn._output('GET /hello HTTP/1.1')
            conn._output("Host: %s" % self.HOST)
            conn._send_output()
            response = conn.response_class(conn.sock, method="GET")
            try:
                response.begin()
            except:
                if not isinstance(sys.exc_info()[1],
                                  (socket.error, httplib.BadStatusLine)):
                    self.fail("Writing to timed out socket didn't fail"
                              " as it should have: %s" % sys.exc_info()[1])
            else:
{'fraction_non_alphanumeric': 0.09675090252707581, 'fraction_numerical': 0.010324909747292419, 'mean_word_length': 1.6148763451010006, 'pattern_counts': {'":': 11, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5493842', 'n_tokens_mistral': 3763, 'n_tokens_neox': 3612, 'n_words': 876}