| 12345678910111213141516171819202122232425262728 |
- def testAutoTransferCompressed(self):
- """Test that automatic transfers are compressed.
- Ensure uploads with the compressed, resumable, and automatic transfer
- flags set call StreamInChunks. StreamInChunks is tested in an earlier
- test.
- """
- # 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)
- upload.strategy = transfer.RESUMABLE_UPLOAD
- # Mock the upload to return the sample response.
- with mock.patch.object(transfer.Upload,
- 'StreamInChunks') 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')
- # Ensure the mock was called.
- self.assertTrue(mock_result.called)
|
