Method update_state

test/backend/data_process/test_tasks.py:511–512

(self, **kw)

        self.states = []

    def update_state(self, **kw):
        self.states.append(kw)

    def retry(self, **kw):
        from celery.exceptions import Retry