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)