12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667 |
- from s3transfer.delete import DeleteObjectTask
- from tests import BaseTaskTest
- class TestDeleteObjectTask(BaseTaskTest):
- def setUp(self):
- super().setUp()
- self.bucket = 'mybucket'
- self.key = 'mykey'
- self.extra_args = {}
- self.callbacks = []
- def get_delete_task(self, **kwargs):
- default_kwargs = {
- 'client': self.client,
- 'bucket': self.bucket,
- 'key': self.key,
- 'extra_args': self.extra_args,
- }
- default_kwargs.update(kwargs)
- return self.get_task(DeleteObjectTask, main_kwargs=default_kwargs)
- def test_main(self):
- self.stubber.add_response(
- 'delete_object',
- service_response={},
- expected_params={
- 'Bucket': self.bucket,
- 'Key': self.key,
- },
- )
- task = self.get_delete_task()
- task()
- self.stubber.assert_no_pending_responses()
- def test_extra_args(self):
- self.extra_args['MFA'] = 'mfa-code'
- self.extra_args['VersionId'] = '12345'
- self.stubber.add_response(
- 'delete_object',
- service_response={},
- expected_params={
- 'Bucket': self.bucket,
- 'Key': self.key,
-
-
- 'MFA': 'mfa-code',
- 'VersionId': '12345',
- },
- )
- task = self.get_delete_task()
- task()
- self.stubber.assert_no_pending_responses()
|