Method chord_context

t/unit/backends/test_redis.py:351–366

(self, size=1)

    @contextmanager
    def chord_context(self, size=1):
        with patch('celery.backends.redis.maybe_signature') as ms:
            request = Mock(name='request')
            request.id = 'id1'
            group_id = 'gid1'
            request.group = group_id
            request.group_index = None
            tasks = [
                self.create_task(i, group_id=request.group)
                for i in range(size)
            ]
            callback = ms.return_value = Signature('add')
            callback.id = 'id1'
            self.b.set_chord_size(group_id, size)
            callback.delay = Mock(name='callback.delay')
            yield tasks, request, callback

    def setup_method(self):
        self.Backend = self.get_backend()