(self)
| 1528 | self.app.conf.result_backend_max_retries = prev_max_retries |
| 1529 | |
| 1530 | def test_get_unsafe_exception(self): |
| 1531 | self.app.conf.result_backend_always_retry, prev = True, self.app.conf.result_backend_always_retry |
| 1532 | |
| 1533 | expected_exc = Exception("failed") |
| 1534 | try: |
| 1535 | b = BaseBackend(app=self.app) |
| 1536 | b._sleep = Mock() |
| 1537 | b._get_task_meta_for = Mock() |
| 1538 | b._get_task_meta_for.side_effect = [ |
| 1539 | expected_exc, |
| 1540 | {'status': states.SUCCESS, 'result': 42} |
| 1541 | ] |
| 1542 | try: |
| 1543 | b.get_task_meta(sentinel.task_id) |
| 1544 | assert False |
| 1545 | except Exception as exc: |
| 1546 | assert b._sleep.call_count == 0 |
| 1547 | assert exc == expected_exc |
| 1548 | finally: |
| 1549 | self.app.conf.result_backend_always_retry = prev |
| 1550 | |
| 1551 | def test_store_result_never_retries(self): |
| 1552 | self.app.conf.result_backend_always_retry, prev = False, self.app.conf.result_backend_always_retry |
nothing calls this directly
no test coverage detected