transfer_test_11.py 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566
  1. def testMultipartEncoding(self):
  2. # This is really a table test for various issues we've seen in
  3. # the past; see notes below for particular histories.
  4. test_cases = [
  5. # Python's mime module by default encodes lines that start
  6. # with "From " as ">From ", which we need to make sure we
  7. # don't run afoul of when sending content that isn't
  8. # intended to be so encoded. This test calls out that we
  9. # get this right. We test for both the multipart and
  10. # non-multipart case.
  11. 'line one\nFrom \nline two',
  12. # We had originally used a `six.StringIO` to hold the http
  13. # request body in the case of a multipart upload; for
  14. # bytes being uploaded in Python3, however, this causes
  15. # issues like this:
  16. # https://github.com/GoogleCloudPlatform/gcloud-python/issues/1760
  17. # We test below to ensure that we don't end up mangling
  18. # the body before sending.
  19. u'name,main_ingredient\nRäksmörgås,Räkor\nBaguette,Bröd',
  20. ]
  21. for upload_contents in test_cases:
  22. multipart_body = '{"body_field_one": 7}'
  23. upload_bytes = upload_contents.encode('ascii', 'backslashreplace')
  24. upload_config = base_api.ApiUploadInfo(
  25. accept=['*/*'],
  26. max_size=None,
  27. resumable_multipart=True,
  28. resumable_path=u'/resumable/upload',
  29. simple_multipart=True,
  30. simple_path=u'/upload',
  31. )
  32. url_builder = base_api._UrlBuilder('http://www.uploads.com')
  33. # Test multipart: having a body argument in http_request forces
  34. # multipart here.
  35. upload = transfer.Upload.FromStream(
  36. six.BytesIO(upload_bytes),
  37. 'text/plain',
  38. total_size=len(upload_bytes))
  39. http_request = http_wrapper.Request(
  40. 'http://www.uploads.com',
  41. headers={'content-type': 'text/plain'},
  42. body=multipart_body)
  43. upload.ConfigureRequest(upload_config, http_request, url_builder)
  44. self.assertEqual(
  45. 'multipart', url_builder.query_params['uploadType'])
  46. rewritten_upload_contents = b'\n'.join(
  47. http_request.body.split(b'--')[2].splitlines()[1:])
  48. self.assertTrue(rewritten_upload_contents.endswith(upload_bytes))
  49. # Test non-multipart (aka media): no body argument means this is
  50. # sent as media.
  51. upload = transfer.Upload.FromStream(
  52. six.BytesIO(upload_bytes),
  53. 'text/plain',
  54. total_size=len(upload_bytes))
  55. http_request = http_wrapper.Request(
  56. 'http://www.uploads.com',
  57. headers={'content-type': 'text/plain'})
  58. upload.ConfigureRequest(upload_config, http_request, url_builder)
  59. self.assertEqual(url_builder.query_params['uploadType'], 'media')
  60. rewritten_upload_contents = http_request.body
  61. self.assertTrue(rewritten_upload_contents.endswith(upload_bytes))