(self)
| 121 | signals.task_postrun.receivers[:] = [] |
| 122 | |
| 123 | def test_with_success_receivers(self): |
| 124 | on_success = Mock() |
| 125 | signals.task_success.connect(on_success) |
| 126 | try: |
| 127 | self.trace(self.add, (2, 2), {}) |
| 128 | on_success.assert_called() |
| 129 | finally: |
| 130 | signals.task_success.receivers[:] = [] |
| 131 | |
| 132 | def test_when_chord_part(self): |
| 133 | @self.app.task(shared=False) |