(self)
| 186 | self.stubber.assert_no_pending_responses() |
| 187 | |
| 188 | def test_callbacks_invoked(self): |
| 189 | subscriber = RecordingSubscriber() |
| 190 | self.callbacks.append(subscriber.on_progress) |
| 191 | self.stubber.add_response( |
| 192 | 'upload_part_copy', |
| 193 | service_response={'CopyPartResult': {'ETag': self.result_etag}}, |
| 194 | expected_params={ |
| 195 | 'Bucket': self.bucket, |
| 196 | 'Key': self.key, |
| 197 | 'CopySource': self.copy_source, |
| 198 | 'UploadId': self.upload_id, |
| 199 | 'PartNumber': self.part_number, |
| 200 | 'CopySourceRange': self.copy_source_range, |
| 201 | }, |
| 202 | ) |
| 203 | task = self.get_copy_task() |
| 204 | self.assertEqual( |
| 205 | task(), {'PartNumber': self.part_number, 'ETag': self.result_etag} |
| 206 | ) |
| 207 | self.stubber.assert_no_pending_responses() |
| 208 | self.assertEqual(subscriber.calculate_bytes_seen(), self.size) |
nothing calls this directly
no test coverage detected