Commits

Robert Brewer committed d025861

Similar test and fix for chunked read readline()

  • Participants
  • Parent commits e597882

Comments (0)

Files changed (2)

cheroot/server.py

             else:
                 data += self.buffer
                 self.buffer = EMPTY
-    
+
     def readline(self, size=None):
         data = EMPTY
         while True:
             if size and len(data) >= size:
                 return data
-            
+
             if not self.buffer:
                 self._fetch()
                 if not self.buffer:
                     # EOF
                     return data
-            
+
             newline_pos = self.buffer.find(LF)
             if size:
                 if newline_pos == -1:
                     data += self.buffer[:remaining]
                     self.buffer = self.buffer[remaining:]
                 else:
-                    remaining = min(size - len(data), newline_pos)
+                    remaining = min(size - len(data), newline_pos + 1)
                     data += self.buffer[:remaining]
                     self.buffer = self.buffer[remaining:]
             else:
                 if newline_pos == -1:
                     data += self.buffer
+                    self.buffer = EMPTY
                 else:
-                    data += self.buffer[:newline_pos]
-                    self.buffer = self.buffer[newline_pos:]
+                    remaining = newline_pos + 1
+                    data += self.buffer[:remaining]
+                    self.buffer = self.buffer[remaining:]
     
     def readlines(self, sizehint=0):
         # Shamelessly stolen from StringIO

cheroot/test/test_core.py

                 output = req.environ['wsgi.input'].read()
                 return output.decode("ISO-8859-1")
 
+            def echo_lines(self, req, resp):
+                output = []
+                while True:
+                    line = req.environ['wsgi.input'].readline().decode("ISO-8859-1")
+                    if not line:
+                        break
+                    output.append(line)
+                return output
+
             def normal(self, req, resp):
                 return "normal"
             
         self.assertStatus(200)
         self.assertBody("I am a request body")
 
+    def test_chunked_request_payload_readline(self):
+        if self.scheme == "https":
+            c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
+        else:
+            c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
+        c.putrequest("POST", "/echo_lines")
+        c.putheader("Transfer-Encoding", "chunked")
+        c.endheaders()
+        c.send(ntob("13\r\nI am a\nrequest body\r\n0\r\n\r\n"))
+        response = c.getresponse()
+        self.body = response.read()
+        c.close()
+        self.status = str(response.status)
+        self.assertStatus(200)
+        self.assertBody("I am a\nrequest body")
+