openstack/deb-python-eventlet
Code Issues Proposed changes

[svn r88] Refactored the body-parsing logic that used to be in read_body() into parsed_body() and added some docs. This means that if you were calling read_body expecting to get anything but a string out, you should change to parsed_body() instead. Refactored tests.py a little bit.

Browse Source
This commit is contained in:
which.linden
2008-02-13 22:13:16 -05:00
parent 96007ae8b3
commit fea78b65b7
3 changed files with 36 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
eventlet/db_pool.py
View File

@@ -63,7 +63,8 @@ connection pools keyed on host,databasename"""
new_kwargs['db'] = dbname
new_kwargs['host'] = host
new_kwargs.update(self.credentials_for(host))
dbpool = self._conn_pool_class(self._module, self._min_size, self._max_size, *self._args, **new_kwargs)
dbpool = self._conn_pool_class(self._module, min_size=self._min_size, max_size=self._max_size,
*self._args, **new_kwargs)
self._databases[key] = dbpool
return self._databases[key]

37
eventlet/httpd.py
View File

@@ -337,22 +337,39 @@ class Request(object):
return self._cached_body
def read_body(self):
""" Returns the string body that was read off the request, or
the empty string if there was no request body.
Requires a content-length header. Caches the body so multiple
calls to read_body() are free.
"""
if not hasattr(self, '_cached_body'):
length = self.get_header('content-length')
if length:
length = int(length)
if length:
self._cached_body = self.protocol.rfile.read(length)
else:
self._cached_body = ''
return self._cached_body
def parsed_body(self):
""" Returns the parsed version of the body, using the
content-type header to select from the parsers on the site
object.
If no parser is found, returns the string body from
read_body(). Caches the parsed body so multiple calls to
parsed_body() are free.
"""
if not hasattr(self, '_cached_parsed_body'):
if not hasattr(self, '_cached_body'):
length = self.get_header('content-length')
if length:
length = int(length)
if length:
self._cached_body = self.protocol.rfile.read(length)
else:
self._cached_body = ''
body = self._cached_body
body = self.read_body()
if hasattr(self.site, 'parsers'):
parser = self.site.parsers.get(
self.get_header('content-type'))
if parser is not None:
body = parser(body)
self._cached_parsed_body = body
self._cached_parsed_body = body
return self._cached_parsed_body
def override_body(self, body):

13
eventlet/tests.py
View File

@@ -23,7 +23,6 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
import atexit
import errno
import os
import sys
@@ -36,8 +35,8 @@ TestCase = unittest.TestCase
name = getattr(sys.modules['__main__'], '__name__', None)
main = unittest.main
# listing of all non-*_test test methods
eventlet_test_files = ['coros']
# listing of files containing doctests
doc_test_files = ['coros']
def find_command(command):
for dir in os.getenv('PATH', '/usr/bin:/usr/sbin').split(os.pathsep):
@@ -46,8 +45,8 @@ def find_command(command):
return p
raise IOError(errno.ENOENT, 'Command not found: %r' % command)
def run_all_tests(test_files = eventlet_test_files):
""" Runs all the unit tests in eventlet, returning immediately after the
def run_all_tests(test_files = doc_test_files):
""" Runs all the unit tests, returning immediately after the
first failed test.
Returns true if the tests all succeeded. This method is really much longer
@@ -81,8 +80,10 @@ def run_all_tests(test_files = eventlet_test_files):
failures, tests = doctest.testmod(test_module)
if failures:
return False
else:
print "OK"
return True
if __name__ == '__main__':
run_all_tests()
run_all_tests()