Test that if soft timeout happens in task but is managed by task, chord still get results normally
(self, manager)
| 326 | i -= 1 |
| 327 | |
| 328 | def test_chord_soft_timeout_recuperation(self, manager): |
| 329 | """Test that if soft timeout happens in task but is managed by task, |
| 330 | chord still get results normally |
| 331 | """ |
| 332 | if not manager.app.conf.result_backend.startswith('redis'): |
| 333 | raise pytest.skip('Requires redis result backend.') |
| 334 | |
| 335 | c = chord([ |
| 336 | # return 3 |
| 337 | add.s(1, 2), |
| 338 | # return 0 after managing soft timeout |
| 339 | delayed_sum_with_soft_guard.s( |
| 340 | [100], pause_time=2 |
| 341 | ).set( |
| 342 | soft_time_limit=1 |
| 343 | ), |
| 344 | ]) |
| 345 | result = c(delayed_sum.s(pause_time=0)).get() |
| 346 | assert result == 3 |
| 347 | |
| 348 | def test_chain_error_handler_with_eta(self, manager): |
| 349 | try: |