(self)
| 633 | |
| 634 | class TestGetObjectTask(BaseTaskTest): |
| 635 | def setUp(self): |
| 636 | super().setUp() |
| 637 | self.bucket = 'mybucket' |
| 638 | self.key = 'mykey' |
| 639 | self.extra_args = {} |
| 640 | self.callbacks = [] |
| 641 | self.max_attempts = 5 |
| 642 | self.io_executor = BoundedExecutor(1000, 1) |
| 643 | self.content = b'my content' |
| 644 | self.stream = BytesIO(self.content) |
| 645 | self.fileobj = WriteCollector() |
| 646 | self.osutil = OSUtils() |
| 647 | self.io_chunksize = 64 * (1024**2) |
| 648 | self.task_cls = GetObjectTask |
| 649 | self.download_output_manager = DownloadSeekableOutputManager( |
| 650 | self.osutil, self.transfer_coordinator, self.io_executor |
| 651 | ) |
| 652 | |
| 653 | def get_download_task(self, **kwargs): |
| 654 | default_kwargs = { |
nothing calls this directly
no test coverage detected