(self)
| 73 | self.stubber.assert_no_pending_responses() |
| 74 | |
| 75 | def test_callbacks_invoked(self): |
| 76 | subscriber = RecordingSubscriber() |
| 77 | self.callbacks.append(subscriber.on_progress) |
| 78 | self.stubber.add_response( |
| 79 | 'copy_object', |
| 80 | service_response={}, |
| 81 | expected_params={ |
| 82 | 'Bucket': self.bucket, |
| 83 | 'Key': self.key, |
| 84 | 'CopySource': self.copy_source, |
| 85 | }, |
| 86 | ) |
| 87 | task = self.get_copy_task() |
| 88 | task() |
| 89 | |
| 90 | self.stubber.assert_no_pending_responses() |
| 91 | self.assertEqual(subscriber.calculate_bytes_seen(), self.size) |
| 92 | |
| 93 | |
| 94 | class TestCopyPartTask(BaseCopyTaskTest): |
nothing calls this directly
no test coverage detected