(self)
| 165 | self.stubber.assert_no_pending_responses() |
| 166 | |
| 167 | def test_extra_args(self): |
| 168 | self.extra_args['RequestPayer'] = 'requester' |
| 169 | self.stubber.add_response( |
| 170 | 'upload_part_copy', |
| 171 | service_response={'CopyPartResult': {'ETag': self.result_etag}}, |
| 172 | expected_params={ |
| 173 | 'Bucket': self.bucket, |
| 174 | 'Key': self.key, |
| 175 | 'CopySource': self.copy_source, |
| 176 | 'UploadId': self.upload_id, |
| 177 | 'PartNumber': self.part_number, |
| 178 | 'CopySourceRange': self.copy_source_range, |
| 179 | 'RequestPayer': 'requester', |
| 180 | }, |
| 181 | ) |
| 182 | task = self.get_copy_task() |
| 183 | self.assertEqual( |
| 184 | task(), {'PartNumber': self.part_number, 'ETag': self.result_etag} |
| 185 | ) |
| 186 | self.stubber.assert_no_pending_responses() |
| 187 | |
| 188 | def test_callbacks_invoked(self): |
| 189 | subscriber = RecordingSubscriber() |
nothing calls this directly
no test coverage detected