(self, bucket_name, key_name, contents='', extra_args=None)
| 901 | return bucket_name |
| 902 | |
| 903 | def put_object(self, bucket_name, key_name, contents='', extra_args=None): |
| 904 | client = self.create_client_for_bucket(bucket_name) |
| 905 | call_args = {'Bucket': bucket_name, 'Key': key_name, 'Body': contents} |
| 906 | if extra_args is not None: |
| 907 | call_args.update(extra_args) |
| 908 | response = client.put_object(**call_args) |
| 909 | self.addCleanup(self.delete_key, bucket_name, key_name) |
| 910 | extra_head_params = {} |
| 911 | if extra_args: |
| 912 | extra_head_params = dict( |
| 913 | (k, v) |
| 914 | for (k, v) in extra_args.items() |
| 915 | if k in self._PUT_HEAD_SHARED_EXTRAS |
| 916 | ) |
| 917 | self.wait_until_key_exists( |
| 918 | bucket_name, |
| 919 | key_name, |
| 920 | extra_params=extra_head_params, |
| 921 | ) |
| 922 | return response |
| 923 | |
| 924 | def delete_bucket(self, bucket_name, attempts=5, delay=5): |
| 925 | self.remove_all_objects(bucket_name) |
no test coverage detected