Method test_scheduled

t/integration/test_inspect.py:127–155

Tests listing scheduled tasks

(self, inspect)

    @flaky
    def test_scheduled(self, inspect):
        """Tests listing scheduled tasks"""
        exec_time = datetime.now(timezone.utc) + timedelta(seconds=5)
        res = add.apply_async([1, 2], {'z': 3}, eta=exec_time)
        ret = inspect.scheduled()
        assert len(ret) == 1
        assert ret[NODENAME] == [
            {
                'eta': exec_time.strftime('%Y-%m-%dT%H:%M:%S.%f') + '+00:00',
                'priority': 6,
                'request': {
                    'id': res.task_id,
                    'name': 't.integration.tasks.add',
                    'args': [1, 2],
                    'kwargs': {'z': 3},
                    'type': 't.integration.tasks.add',
                    'hostname': ANY,
                    'time_start': None,
                    'acknowledged': False,
                    'delivery_info': {
                        'exchange': '',
                        'routing_key': 'celery',
                        'priority': 0,
                        'redelivered': False
                    },
                    'worker_pid': None
                }
            }
        ]

    @flaky
    def test_query_task(self, inspect):