Source Code for Module cherrypy.test.test_conn

  1  from cherrypy.test import test 
  2  test.prefer_parent_path() 
  3   
  4  import httplib 
  5  import socket 
  6  import sys 
  7  import time 
  8  timeout = 0.1 
  9   
 10   
 11  import cherrypy 
 12  from cherrypy.test import webtest 
 13   
 14   
 15  pov = 'pPeErRsSiIsStTeEnNcCeE oOfF vViIsSiIoOnN' 
 16   
 17 -def setup_server(): 
 18      class Root: 
 19           
 20          def index(self): 
 21              return pov 
 22          index.exposed = True 
 23          page1 = index 
 24          page2 = index 
 25          page3 = index 
 26           
 27          def hello(self): 
 28              return "Hello, world!" 
 29          hello.exposed = True 
 30           
 31          def stream(self, set_cl=False): 
 32              if set_cl: 
 33                  cherrypy.response.headers['Content-Length'] = 10 
 34               
 35              def content(): 
 36                  for x in xrange(10): 
 37                      yield str(x) 
 38               
 39              return content() 
 40          stream.exposed = True 
 41          stream._cp_config = {'response.stream': True} 
 42           
 43          def upload(self): 
 44              return ("thanks for '%s' (%s)" % 
 45                      (cherrypy.request.body.read(), 
 46                       cherrypy.request.headers['Content-Type'])) 
 47          upload.exposed = True 
 48           
 49          def custom(self, response_code): 
 50              cherrypy.response.status = response_code 
 51              return "Code = %s" % response_code 
 52          custom.exposed = True 
 53       
 54      cherrypy.tree.mount(Root()) 
 55      cherrypy.config.update({ 
 56          'server.max_request_body_size': 100, 
 57          'environment': 'test_suite', 
 58          }) 
 59   
 60   
 61  from cherrypy.test import helper 
 62   
 63 -class ConnectionTests(helper.CPWebCase): 
 64       
 65 -    def test_HTTP11(self): 
 66          if cherrypy.server.protocol_version != "HTTP/1.1": 
 67              print "skipped ", 
 68              return 
 69           
 70          self.PROTOCOL = "HTTP/1.1" 
 71           
 72          self.persistent = True 
 73           
 74          # Make the first request and assert there's no "Connection: close". 
 75          self.getPage("/") 
 76          self.assertStatus('200 OK') 
 77          self.assertBody(pov) 
 78          self.assertNoHeader("Connection") 
 79           
 80          # Make another request on the same connection. 
 81          self.getPage("/page1") 
 82          self.assertStatus('200 OK') 
 83          self.assertBody(pov) 
 84          self.assertNoHeader("Connection") 
 85           
 86          # Test client-side close. 
 87          self.getPage("/page2", headers=[("Connection", "close")]) 
 88          self.assertStatus('200 OK') 
 89          self.assertBody(pov) 
 90          self.assertHeader("Connection", "close") 
 91           
 92          # Make another request on the same connection, which should error. 
 93          self.assertRaises(httplib.NotConnected, self.getPage, "/") 
 94       
 95 -    def test_Streaming_no_len(self): 
 96          self._streaming(set_cl=False) 
 97       
 98 -    def test_Streaming_with_len(self): 
 99          self._streaming(set_cl=True) 
100       
101 -    def _streaming(self, set_cl): 
102          if cherrypy.server.protocol_version == "HTTP/1.1": 
103              self.PROTOCOL = "HTTP/1.1" 
104               
105              self.persistent = True 
106               
107              # Make the first request and assert there's no "Connection: close". 
108              self.getPage("/") 
109              self.assertStatus('200 OK') 
110              self.assertBody(pov) 
111              self.assertNoHeader("Connection") 
112               
113              # Make another, streamed request on the same connection. 
114              if set_cl: 
115                  # When a Content-Length is provided, the content should stream 
116                  # without closing the connection. 
117                  self.getPage("/stream?set_cl=Yes") 
118                  self.assertHeader("Content-Length") 
119                  self.assertNoHeader("Connection", "close") 
120                  self.assertNoHeader("Transfer-Encoding") 
121                   
122                  self.assertStatus('200 OK') 
123                  self.assertBody('0123456789') 
124              else: 
125                  # When no Content-Length response header is provided, 
126                  # streamed output will either close the connection, or use 
127                  # chunked encoding, to determine transfer-length. 
128                  self.getPage("/stream") 
129                  self.assertNoHeader("Content-Length") 
130                  self.assertStatus('200 OK') 
131                  self.assertBody('0123456789') 
132                   
133                  chunked_response = False 
134                  for k, v in self.headers: 
135                      if k.lower() == "transfer-encoding": 
136                          if str(v) == "chunked": 
137                              chunked_response = True 
138                   
139                  if chunked_response: 
140                      self.assertNoHeader("Connection", "close") 
141                  else: 
142                      self.assertHeader("Connection", "close") 
143                       
144                      # Make another request on the same connection, which should error. 
145                      self.assertRaises(httplib.NotConnected, self.getPage, "/") 
146          else: 
147              self.PROTOCOL = "HTTP/1.0" 
148               
149              self.persistent = True 
150               
151              # Make the first request and assert Keep-Alive. 
152              self.getPage("/", headers=[("Connection", "Keep-Alive")]) 
153              self.assertStatus('200 OK') 
154              self.assertBody(pov) 
155              self.assertHeader("Connection", "Keep-Alive") 
156               
157              # Make another, streamed request on the same connection. 
158              if set_cl: 
159                  # When a Content-Length is provided, the content should 
160                  # stream without closing the connection. 
161                  self.getPage("/stream?set_cl=Yes", 
162                               headers=[("Connection", "Keep-Alive")]) 
163                  self.assertHeader("Content-Length") 
164                  self.assertHeader("Connection", "Keep-Alive") 
165                  self.assertNoHeader("Transfer-Encoding") 
166                  self.assertStatus('200 OK') 
167                  self.assertBody('0123456789') 
168              else: 
169                  # When a Content-Length is not provided, 
170                  # the server should close the connection. 
171                  self.getPage("/stream", headers=[("Connection", "Keep-Alive")]) 
172                  self.assertStatus('200 OK') 
173                  self.assertBody('0123456789') 
174                   
175                  self.assertNoHeader("Content-Length") 
176                  self.assertNoHeader("Connection", "Keep-Alive") 
177                  self.assertNoHeader("Transfer-Encoding") 
178                   
179                  # Make another request on the same connection, which should error. 
180                  self.assertRaises(httplib.NotConnected, self.getPage, "/") 
181       
182 -    def test_HTTP11_Timeout(self): 
183          if cherrypy.server.protocol_version != "HTTP/1.1": 
184              print "skipped ", 
185              return 
186           
187          old_timeout = None 
188          try: 
189              httpserver = cherrypy.server.httpservers.keys()[0] 
190              old_timeout = httpserver.timeout 
191          except (AttributeError, IndexError): 
192              print "skipped ", 
193              return 
194           
195          try: 
196              httpserver.timeout = timeout 
197              self.PROTOCOL = "HTTP/1.1" 
198               
199              # Make an initial request 
200              self.persistent = True 
201              conn = self.HTTP_CONN 
202              conn.putrequest("GET", "/", skip_host=True) 
203              conn.putheader("Host", self.HOST) 
204              conn.endheaders() 
205              response = conn.response_class(conn.sock, method="GET") 
206              response.begin() 
207              self.assertEqual(response.status, 200) 
208              self.body = response.read() 
209              self.assertBody(pov) 
210               
211              # Make a second request on the same socket 
212              conn._output('GET /hello HTTP/1.1') 
213              conn._output("Host: %s" % self.HOST) 
214              conn._send_output() 
215              response = conn.response_class(conn.sock, method="GET") 
216              response.begin() 
217              self.assertEqual(response.status, 200) 
218              self.body = response.read() 
219              self.assertBody("Hello, world!") 
220               
221              # Wait for our socket timeout 
222              time.sleep(timeout * 2) 
223               
224              # Make another request on the same socket, which should error 
225              conn._output('GET /hello HTTP/1.1') 
226              conn._output("Host: %s" % self.HOST) 
227              conn._send_output() 
228              response = conn.response_class(conn.sock, method="GET") 
229              try: 
230                  response.begin() 
231              except: 
232                  if not isinstance(sys.exc_info()[1], 
233                                    (socket.error, httplib.BadStatusLine)): 
234                      self.fail("Writing to timed out socket didn't fail" 
235                                " as it should have: %s" % sys.exc_info()[1]) 
236              else: 
237                  self.fail("Writing to timed out socket didn't fail" 
238                            " as it should have: %s" % 
239                            response.read()) 
240               
241              conn.close() 
242               
243              # Make another request on a new socket, which should work 
244              self.persistent = True 
245              conn = self.HTTP_CONN 
246              conn.putrequest("GET", "/", skip_host=True) 
247              conn.putheader("Host", self.HOST) 
248              conn.endheaders() 
249              response = conn.response_class(conn.sock, method="GET") 
250              response.begin() 
251              self.assertEqual(response.status, 200) 
252              self.body = response.read() 
253              self.assertBody(pov) 
254          finally: 
255              if old_timeout is not None: 
256                  httpserver.timeout = old_timeout 
257       
258 -    def test_HTTP11_pipelining(self): 
259          if cherrypy.server.protocol_version != "HTTP/1.1": 
260              print "skipped ", 
261              return 
262           
263          self.PROTOCOL = "HTTP/1.1" 
264           
265          # Test pipelining. httplib doesn't support this directly. 
266          self.persistent = True 
267          conn = self.HTTP_CONN 
268           
269          # Put request 1 
270          conn.putrequest("GET", "/hello", skip_host=True) 
271          conn.putheader("Host", self.HOST) 
272          conn.endheaders() 
273           
274          for trial in xrange(5): 
275              # Put next request 
276              conn._output('GET /hello HTTP/1.1') 
277              conn._output("Host: %s" % self.HOST) 
278              conn._send_output() 
279               
280              # Retrieve previous response 
281              response = conn.response_class(conn.sock, method="GET") 
282              response.begin() 
283              body = response.read() 
284              self.assertEqual(response.status, 200) 
285              self.assertEqual(body, "Hello, world!") 
286           
287          # Retrieve final response 
288          response = conn.response_class(conn.sock, method="GET") 
289          response.begin() 
290          body = response.read() 
291          self.assertEqual(response.status, 200) 
292          self.assertEqual(body, "Hello, world!") 
293           
294          conn.close() 
295       
296 -    def test_100_Continue(self): 
297          if cherrypy.server.protocol_version != "HTTP/1.1": 
298              print "skipped ", 
299              return 
300           
301          self.PROTOCOL = "HTTP/1.1" 
302           
303          self.persistent = True 
304          conn = self.HTTP_CONN 
305           
306          # Try a page without an Expect request header first. 
307          # Note that httplib's response.begin automatically ignores 
308          # 100 Continue responses, so we must manually check for it. 
309          conn.putrequest("POST", "/upload", skip_host=True) 
310          conn.putheader("Host", self.HOST) 
311          conn.putheader("Content-Type", "text/plain") 
312          conn.putheader("Content-Length", "4") 
313          conn.endheaders() 
314          conn.send("d'oh") 
315          response = conn.response_class(conn.sock, method="POST") 
316          version, status, reason = response._read_status() 
317          self.assertNotEqual(status, 100) 
318          conn.close() 
319           
320          # Now try a page with an Expect header... 
321          conn.connect() 
322          conn.putrequest("POST", "/upload", skip_host=True) 
323          conn.putheader("Host", self<