Method test_active

t/integration/test_inspect.py:100–124

Tests listing active tasks

(self, inspect)

    @flaky
    def test_active(self, inspect):
        """Tests listing active tasks"""
        res = sleeping.delay(5)
        sleep(1)
        ret = inspect.active()
        assert len(ret) == 1
        assert ret[NODENAME] == [
            {
                'id': res.task_id,
                'name': 't.integration.tasks.sleeping',
                'args': [5],
                'kwargs': {},
                'type': 't.integration.tasks.sleeping',
                'hostname': ANY,
                'time_start': ANY,
                'acknowledged': True,
                'delivery_info': {
                    'exchange': '',
                    'routing_key': 'celery',
                    'priority': 0,
                    'redelivered': False
                },
                'worker_pid': ANY
            }
        ]

    @flaky
    def test_scheduled(self, inspect):