def testMultipartEncoding(self): # This is really a table test for various issues we've seen in # the past; see notes below for particular histories. test_cases = [ # Python's mime module by default encodes lines that start # with "From " as ">From ", which we need to make sure we # don't run afoul of when sending content that isn't # intended to be so encoded. This test calls out that we # get this right. We test for both the multipart and # non-multipart case. 'line one\nFrom \nline two', # We had originally used a `six.StringIO` to hold the http # request body in the case of a multipart upload; for # bytes being uploaded in Python3, however, this causes # issues like this: # https://github.com/GoogleCloudPlatform/gcloud-python/issues/1760 # We test below to ensure that we don't end up mangling # the body before sending. u'name,main_ingredient\nRäksmörgås,Räkor\nBaguette,Bröd', ] for upload_contents in test_cases: multipart_body = '{"body_field_one": 7}' upload_bytes = upload_contents.encode('ascii', 'backslashreplace') upload_config = base_api.ApiUploadInfo( accept=['*/*'], max_size=None, resumable_multipart=True, resumable_path=u'/resumable/upload', simple_multipart=True, simple_path=u'/upload', ) url_builder = base_api._UrlBuilder('http://www.uploads.com') # Test multipart: having a body argument in http_request forces # multipart here. upload = transfer.Upload.FromStream( six.BytesIO(upload_bytes), 'text/plain', total_size=len(upload_bytes)) http_request = http_wrapper.Request( 'http://www.uploads.com', headers={'content-type': 'text/plain'}, body=multipart_body) upload.ConfigureRequest(upload_config, http_request, url_builder) self.assertEqual( 'multipart', url_builder.query_params['uploadType']) rewritten_upload_contents = b'\n'.join( http_request.body.split(b'--')[2].splitlines()[1:]) self.assertTrue(rewritten_upload_contents.endswith(upload_bytes)) # Test non-multipart (aka media): no body argument means this is # sent as media. upload = transfer.Upload.FromStream( six.BytesIO(upload_bytes), 'text/plain', total_size=len(upload_bytes)) http_request = http_wrapper.Request( 'http://www.uploads.com', headers={'content-type': 'text/plain'}) upload.ConfigureRequest(upload_config, http_request, url_builder) self.assertEqual(url_builder.query_params['uploadType'], 'media') rewritten_upload_contents = http_request.body self.assertTrue(rewritten_upload_contents.endswith(upload_bytes))