| 1234567891011121314151617181920212223242526272829303132 |
- def testStreamInChunksCompressed(self):
- """Test that StreamInChunks will handle compression correctly."""
- # Create and configure 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)
- upload.strategy = transfer.RESUMABLE_UPLOAD
- # Set the chunk size so the entire stream is uploaded.
- upload.chunksize = len(self.sample_data)
- # Mock the upload to return the sample response.
- with mock.patch.object(transfer.Upload,
- '_Upload__SendMediaRequest') as mock_result, \
- mock.patch.object(http_wrapper,
- 'MakeRequest') as make_request:
- mock_result.return_value = self.response
- make_request.return_value = self.response
- # Initialization.
- upload.InitializeUpload(self.request, 'http')
- upload.StreamInChunks()
- # Get the uploaded request and end position of the stream.
- (request, _), _ = mock_result.call_args_list[0]
- # Ensure the mock was called.
- self.assertTrue(mock_result.called)
- # Ensure the correct content encoding was set.
- self.assertEqual(request.headers['Content-Encoding'], 'gzip')
- # Ensure the stream was compresed.
- self.assertLess(len(request.body), len(self.sample_data))
|
