(self)
| 36 | |
| 37 | class BaseUploadTest(BaseGeneralInterfaceTest): |
| 38 | def setUp(self): |
| 39 | super().setUp() |
| 40 | # TODO: We do not want to use the real MIN_UPLOAD_CHUNKSIZE |
| 41 | # when we're adjusting parts. |
| 42 | # This is really wasteful and fails CI builds because self.contents |
| 43 | # would normally use 10MB+ of memory. |
| 44 | # Until there's an API to configure this, we're patching this with |
| 45 | # a min size of 1. We can't patch MIN_UPLOAD_CHUNKSIZE directly |
| 46 | # because it's already bound to a default value in the |
| 47 | # chunksize adjuster. Instead we need to patch out the |
| 48 | # chunksize adjuster class. |
| 49 | self.adjuster_patch = mock.patch( |
| 50 | 's3transfer.upload.ChunksizeAdjuster', |
| 51 | lambda: ChunksizeAdjuster(min_size=1), |
| 52 | ) |
| 53 | self.adjuster_patch.start() |
| 54 | self.config = TransferConfig(max_request_concurrency=1) |
| 55 | self._manager = TransferManager(self.client, self.config) |
| 56 | |
| 57 | # Create a temporary directory with files to read from |
| 58 | self.tempdir = tempfile.mkdtemp() |
| 59 | self.filename = os.path.join(self.tempdir, 'myfile') |
| 60 | self.content = b'my content' |
| 61 | |
| 62 | with open(self.filename, 'wb') as f: |
| 63 | f.write(self.content) |
| 64 | |
| 65 | # Initialize some default arguments |
| 66 | self.bucket = 'mybucket' |
| 67 | self.key = 'mykey' |
| 68 | self.extra_args = {} |
| 69 | self.subscribers = [] |
| 70 | |
| 71 | # A list to keep track of all of the bodies sent over the wire |
| 72 | # and their order. |
| 73 | self.sent_bodies = [] |
| 74 | self.client.meta.events.register( |
| 75 | 'before-parameter-build.s3.*', self.collect_body |
| 76 | ) |
| 77 | |
| 78 | def tearDown(self): |
| 79 | super().tearDown() |
no test coverage detected