(self)
| 429 | assert panel.handle('dump_schedule') |
| 430 | |
| 431 | def test_dump_reserved(self): |
| 432 | consumer = Consumer(self.app) |
| 433 | req = Request( |
| 434 | self.TaskMessage(self.mytask.name, args=(2, 2)), app=self.app, |
| 435 | ) # ^ need to keep reference for reserved_tasks WeakSet. |
| 436 | worker_state.task_reserved(req) |
| 437 | try: |
| 438 | panel = self.create_panel(consumer=consumer) |
| 439 | response = panel.handle('dump_reserved', {'safe': True}) |
| 440 | assert response[0]['name'] == self.mytask.name |
| 441 | assert response[0]['hostname'] == socket.gethostname() |
| 442 | worker_state.reserved_requests.clear() |
| 443 | assert not panel.handle('dump_reserved') |
| 444 | finally: |
| 445 | worker_state.reserved_requests.clear() |
| 446 | |
| 447 | def test_rate_limit_invalid_rate_limit_string(self): |
| 448 | e = self.panel.handle('rate_limit', arguments={ |
nothing calls this directly
no test coverage detected