(self)
| 2259 | self.assertFalse(r3.is_revoked()) # Cleared when it was handled. |
| 2260 | |
| 2261 | def test_chain_errors(self): |
| 2262 | state1, state2 = [], [] |
| 2263 | |
| 2264 | @self.huey.task() |
| 2265 | def task_a(n): |
| 2266 | if n < 1: |
| 2267 | raise TestError(n) |
| 2268 | return n - 1 |
| 2269 | |
| 2270 | @self.huey.task() |
| 2271 | def task_e1(n, err): |
| 2272 | state1.append(repr(err)) |
| 2273 | if n < 1: |
| 2274 | raise TestError(n) |
| 2275 | |
| 2276 | @self.huey.task() |
| 2277 | def task_e2(n, err): |
| 2278 | state2.append((n, repr(err))) |
| 2279 | return n |
| 2280 | |
| 2281 | pipe = task_a.s(1).then(task_a) |
| 2282 | pipe.on_complete.error(task_e1, 0).error(task_e2, 3) |
| 2283 | |
| 2284 | # We get result wrappers for the invocations of task_a. |
| 2285 | r1, r2 = self.huey.enqueue(pipe) |
| 2286 | self.execute_next() # First invocation, returns 0. |
| 2287 | self.execute_next() # Second, raises error. |
| 2288 | |
| 2289 | self.assertEqual(len(self.huey), 1) |
| 2290 | self.execute_next() # Error handler (also fails). |
| 2291 | self.assertEqual(len(self.huey), 1) |
| 2292 | |
| 2293 | self.assertEqual(self.execute_next(), 3) # Second error handler. |
| 2294 | |
| 2295 | self.assertEqual(state1, ['TestError(0)']) |
| 2296 | self.assertEqual(state2, [(3, 'TestError(0)')]) |
| 2297 | |
| 2298 | self.assertEqual(r1(), 0) |
| 2299 | self.assertRaises(TaskException, r2.get) |
| 2300 | self.assertEqual(len(self.huey), 0) |
| 2301 | |
| 2302 | |
| 2303 | class TestTaskLocking(BaseTestCase): |
nothing calls this directly
no test coverage detected