from tests.functional.s3 import BaseS3TransferCommandTest
class TestRmCommand(BaseS3TransferCommandTest):
prefix = 's3 rm'
def test_operations_used(self):
cmdline = '%s s3://bucket/key.txt' % self.prefix
self.run_cmd(cmdline, expected_rc=0)
self.assertEqual(
len(self.operations_called), 1, self.operations_called)
self.assertEqual(self.operations_called[0][0].name, 'DeleteObject')
def test_delete_with_request_payer(self):
cmdline = '%s s3://mybucket/mykey --request-payer' % self.prefix
self.run_cmd(cmdline, expected_rc=0)
self.assert_operations_called(
[
('DeleteObject', {
'Bucket': 'mybucket',
'Key': 'mykey',
'RequestPayer': 'requester'
})
]
)
def test_recursive_delete_with_requests(self):
cmdline = '%s s3://mybucket/ --recursive --request-payer' % self.prefix
self.parsed_responses = [
self.list_objects_response(['mykey']),
self.empty_response(),
]
self.run_cmd(cmdline, expected_rc=0)
self.assert_operations_called(
[
self.list_objects_request(
'mybucket', RequestPayer='requester'),
self.delete_object_request(
'mybucket', 'mykey', RequestPayer='requester'),
]
)