(self)
| 444 | |
| 445 | class TestUploadSubmissionTask(BaseSubmissionTaskTest): |
| 446 | def setUp(self): |
| 447 | super().setUp() |
| 448 | self.tempdir = tempfile.mkdtemp() |
| 449 | self.filename = os.path.join(self.tempdir, 'myfile') |
| 450 | self.content = b'0' * (MIN_UPLOAD_CHUNKSIZE * 3) |
| 451 | self.config.multipart_chunksize = MIN_UPLOAD_CHUNKSIZE |
| 452 | self.config.multipart_threshold = MIN_UPLOAD_CHUNKSIZE * 5 |
| 453 | |
| 454 | with open(self.filename, 'wb') as f: |
| 455 | f.write(self.content) |
| 456 | |
| 457 | self.bucket = 'mybucket' |
| 458 | self.key = 'mykey' |
| 459 | self.extra_args = {} |
| 460 | self.subscribers = [] |
| 461 | |
| 462 | # A list to keep track of all of the bodies sent over the wire |
| 463 | # and their order. |
| 464 | self.sent_bodies = [] |
| 465 | self.client.meta.events.register( |
| 466 | 'before-parameter-build.s3.*', self.collect_body |
| 467 | ) |
| 468 | |
| 469 | self.call_args = self.get_call_args() |
| 470 | self.transfer_future = self.get_transfer_future(self.call_args) |
| 471 | self.submission_main_kwargs = { |
| 472 | 'client': self.client, |
| 473 | 'config': self.config, |
| 474 | 'osutil': self.osutil, |
| 475 | 'request_executor': self.executor, |
| 476 | 'transfer_future': self.transfer_future, |
| 477 | } |
| 478 | self.submission_task = self.get_task( |
| 479 | UploadSubmissionTask, main_kwargs=self.submission_main_kwargs |
| 480 | ) |
| 481 | |
| 482 | def tearDown(self): |
| 483 | super().tearDown() |
nothing calls this directly
no test coverage detected