(self, bucket_name, key_name, contents='', extra_args=None)
| 580 | return bucket_name |
| 581 | |
| 582 | def put_object(self, bucket_name, key_name, contents='', extra_args=None): |
| 583 | client = self._create_client_for_bucket(bucket_name) |
| 584 | call_args = {'Bucket': bucket_name, 'Key': key_name, 'Body': contents} |
| 585 | if extra_args is not None: |
| 586 | call_args.update(extra_args) |
| 587 | response = client.put_object(**call_args) |
| 588 | extra_head_params = {} |
| 589 | if extra_args: |
| 590 | extra_head_params = dict( |
| 591 | (k, v) |
| 592 | for (k, v) in extra_args.items() |
| 593 | if k in self._PUT_HEAD_SHARED_EXTRAS |
| 594 | ) |
| 595 | self.wait_until_key_exists( |
| 596 | bucket_name, |
| 597 | key_name, |
| 598 | extra_params=extra_head_params, |
| 599 | ) |
| 600 | return response |
| 601 | |
| 602 | def delete_bucket(self, bucket_name, attempts=5, delay=5): |
| 603 | self.remove_all_objects(bucket_name) |
nothing calls this directly
no test coverage detected