| 12345678910111213141516171819202122232425262728293031 |
- def testMultipartCompressed(self):
- """Test that multipart uploads are compressed."""
- # Create the multipart configuration.
- upload_config = base_api.ApiUploadInfo(
- accept=['*/*'],
- max_size=None,
- simple_multipart=True,
- simple_path=u'/upload',)
- # Create the upload object.
- upload = transfer.Upload(
- stream=self.sample_stream,
- mime_type='text/plain',
- total_size=len(self.sample_data),
- close_stream=False,
- gzip_encoded=True)
- # Set a body to trigger multipart configuration.
- self.request.body = '{"body_field_one": 7}'
- # Configure the request.
- upload.ConfigureRequest(upload_config, self.request, self.url_builder)
- # Ensure the request is a multipart request now.
- self.assertEqual(
- self.url_builder.query_params['uploadType'], 'multipart')
- # Ensure the request is gzip encoded.
- self.assertEqual(self.request.headers['Content-Encoding'], 'gzip')
- # Ensure data is compressed
- self.assertLess(len(self.request.body), len(self.sample_data))
- # Ensure uncompressed data includes the sample data.
- with gzip.GzipFile(fileobj=self.request.body) as f:
- original = f.read()
- self.assertTrue(self.sample_data in original)
|
