Method test_main

tests/unit/s3transfer/test_upload.py:632–657

(self)

class TestPutObjectTask(BaseUploadTest):
    def test_main(self):
        extra_args = {'Metadata': {'foo': 'bar'}}
        with open(self.filename, 'rb') as fileobj:
            task = self.get_task(
                PutObjectTask,
                main_kwargs={
                    'client': self.client,
                    'fileobj': fileobj,
                    'bucket': self.bucket,
                    'key': self.key,
                    'extra_args': extra_args,
                },
            )
            self.stubber.add_response(
                method='put_object',
                service_response={},
                expected_params={
                    'Body': ANY,
                    'Bucket': self.bucket,
                    'Key': self.key,
                    'Metadata': {'foo': 'bar'},
                },
            )
            task()
            self.stubber.assert_no_pending_responses()
            self.assertEqual(self.sent_bodies, [self.content])


class TestUploadPartTask(BaseUploadTest):