diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/dummy_server.py | 3 | ||||
| -rw-r--r-- | test/test_filepost.py | 31 | ||||
| -rw-r--r-- | test/test_withdummy.py | 20 |
3 files changed, 50 insertions, 4 deletions
diff --git a/test/dummy_server.py b/test/dummy_server.py index 676ba5f8..8e79b0bd 100644 --- a/test/dummy_server.py +++ b/test/dummy_server.py @@ -39,6 +39,9 @@ class TestingApp(object): size = int(request.params.get('upload_size', '0')) file = request.params.get(param) + if file == None: + return Response("Missing parameter: %s" % param, status='400') + if not isinstance(file, FieldStorage): return Response("Not a file: %s" % param, status='400') diff --git a/test/test_filepost.py b/test/test_filepost.py new file mode 100644 index 00000000..599cd037 --- /dev/null +++ b/test/test_filepost.py @@ -0,0 +1,31 @@ +import unittest + +import sys +sys.path.append('../') + +from urllib3 import filepost +from StringIO import StringIO + +class TestFilePost(unittest.TestCase): + def test_generator(self): + fields = { + 'foo': 'bar', + 'somefile': ('name.txt', StringIO('trolololol')), + } + + stream = filepost.MultipartEncoderGenerator(fields, boundary="boundary") + body = ''.join(chunk for chunk in filepost.IterStreamer(stream)) + self.assertEqual(body, u'--boundary\r\nContent-Disposition: form-data; name="somefile"; filename="name.txt"\r\nContent-Type: text/plain\r\n\r\ntrolololol\r\n--boundary\r\nContent-Disposition: form-data; name="foo"\r\nContent-Type: text/plain\r\n\r\nbar\r\n--boundary--\r\n') + + def test_len(self): + fields = { + 'foo': 'bar', + 'somefile': ('name.txt', StringIO('trolololol')), + } + + iterdata = filepost.MultipartEncoderGenerator(fields, boundary="boundary") + predicted_size = len(iterdata) + + body = ''.join(chunk for chunk in filepost.IterStreamer(iterdata)) + + self.assertEqual(len(body), predicted_size) diff --git a/test/test_withdummy.py b/test/test_withdummy.py index a5f79f33..69e4efb7 100644 --- a/test/test_withdummy.py +++ b/test/test_withdummy.py @@ -21,7 +21,7 @@ class TestConnectionPool(unittest.TestCase): r = self.http_pool.get_url('/', retries=1) if r.data != "Dummy server!": raise Exception("Got unexpected response: %s" % r.data) - except Exception, e: + except MaxRetryError, e: raise Exception("Dummy server not running, make sure HOST and PORT correspond to the dummy server: %s" % e.message) return super(TestConnectionPool, self).__init__(*args, **kw) @@ -57,11 +57,23 @@ class TestConnectionPool(unittest.TestCase): r = self.http_pool.post_url('/upload', fields=fields) self.assertEquals(r.status, 200, r.data) - def test_unicode_upload(self): - fields = { - u'\xe2\x99\xa5': (u'\xe2\x99\xa5.txt', u'\xe2\x99\xa5'), + def _make_fields(self, s): + return { + u'upload_param': s, + u'upload_filename': '%s.txt' % s, + u'upload_size': u'3', + s: ('%s.txt' % s, s), } + def test_unicode_decoded(self): + fields = self._make_fields('\xe2\x99\xa5') + + r = self.http_pool.post_url('/upload', fields=fields) + self.assertEquals(r.status, 200, r.data) + + def test_unicode_encoded(self): + fields = self._make_fields(u'\u2665') + r = self.http_pool.post_url('/upload', fields=fields) self.assertEquals(r.status, 200, r.data) |