Merge pull request #2 from arthurdarcet/multipart-transfer-encoding

support all the Content-Transfer-Encoding in the rfc1341
aio-libs · Mar 8, 2017 · 10a25ae · 10a25ae
2 parents 716e5c4 + cd18886
commit 10a25ae
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 8 deletions.
diff --git a/aiohttp/multipart.py b/aiohttp/multipart.py
@@ -455,7 +455,7 @@ def _decode_content_transfer(self, data):
             return base64.b64decode(data)
         elif encoding == 'quoted-printable':
             return binascii.a2b_qp(data)
-        elif encoding == 'binary':
+        elif encoding in ('binary', '8bit', '7bit'):
             return data
         else:
             raise RuntimeError('unknown content transfer encoding: {}'

diff --git a/tests/test_multipart.py b/tests/test_multipart.py
@@ -363,14 +363,17 @@ def test_read_with_content_transfer_encoding_quoted_printable(self):
         self.assertEqual(b'\xd0\x9f\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82,'
                          b' \xd0\xbc\xd0\xb8\xd1\x80!', result)
 
+    @pytest.mark.parametrize('encoding', [])
     def test_read_with_content_transfer_encoding_binary(self):
-        obj = aiohttp.multipart.BodyPartReader(
-            self.boundary, {CONTENT_TRANSFER_ENCODING: 'binary'},
-            Stream(b'\xd0\x9f\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82,'
-                   b' \xd0\xbc\xd0\xb8\xd1\x80!\r\n--:--'))
-        result = yield from obj.read(decode=True)
-        self.assertEqual(b'\xd0\x9f\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82,'
-                         b' \xd0\xbc\xd0\xb8\xd1\x80!', result)
+        data = b'\xd0\x9f\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82,' \
+               b' \xd0\xbc\xd0\xb8\xd1\x80!'
+        for encoding in ('binary', '8bit', '7bit'):
+            with self.subTest(encoding):
+                obj = aiohttp.multipart.BodyPartReader(
+                    self.boundary, {CONTENT_TRANSFER_ENCODING: encoding},
+                    Stream(data + b'\r\n--:--'))
+                result = yield from obj.read(decode=True)
+                self.assertEqual(data, result)
 
     def test_read_with_content_transfer_encoding_unknown(self):
         obj = aiohttp.multipart.BodyPartReader(