(self)
| 2207 | self.assertEqual([r() for r in results], expected) |
| 2208 | |
| 2209 | def test_error_callback(self): |
| 2210 | state = [] |
| 2211 | @self.huey.task() |
| 2212 | def task_a(n): |
| 2213 | raise TestError(n) |
| 2214 | |
| 2215 | @self.huey.task() |
| 2216 | def task_e(err): |
| 2217 | state.append(repr(err)) |
| 2218 | return 1337 |
| 2219 | |
| 2220 | task = task_a.s(0).error(task_e) |
| 2221 | result = self.huey.enqueue(task) |
| 2222 | self.assertTrue(self.execute_next() is None) |
| 2223 | self.assertRaises(TaskException, result.get) |
| 2224 | self.assertEqual(len(self.huey), 1) |
| 2225 | self.assertEqual(self.execute_next(), 1337) |
| 2226 | self.assertEqual(state, ['TestError(0)']) |
| 2227 | |
| 2228 | def test_pipeline_error_midway(self): |
| 2229 | @self.huey.task() |
nothing calls this directly
no test coverage detected