(self)
| 326 | assert tb.get_state(tid) == states.REVOKED |
| 327 | |
| 328 | def test_mark_as_retry(self): |
| 329 | tb = DatabaseBackend(self.uri, app=self.app) |
| 330 | tid = uuid() |
| 331 | try: |
| 332 | raise KeyError('foo') |
| 333 | except KeyError as exception: |
| 334 | import traceback |
| 335 | trace = '\n'.join(traceback.format_stack()) |
| 336 | tb.mark_as_retry(tid, exception, traceback=trace) |
| 337 | assert tb.get_state(tid) == states.RETRY |
| 338 | assert isinstance(tb.get_result(tid), KeyError) |
| 339 | assert tb.get_traceback(tid) == trace |
| 340 | |
| 341 | def test_mark_as_failure(self): |
| 342 | tb = DatabaseBackend(self.uri, app=self.app) |
nothing calls this directly
no test coverage detected