Method get_request

t/unit/worker/test_request.py:187–209

(self,
                    sig,
                    Request=Request,
                    exclude_headers=None,
                    headers=None,
                    **kwargs)

class test_Request(RequestCase):

    def get_request(self,
                    sig,
                    Request=Request,
                    exclude_headers=None,
                    headers=None,
                    **kwargs):
        msg = self.task_message_from_sig(self.app, sig)
        if headers is None:
            headers = msg.headers.copy()
            if exclude_headers:
                for header in exclude_headers:
                    headers.pop(header, None)
        return Request(
            msg,
            on_ack=Mock(name='on_ack'),
            on_reject=Mock(name='on_reject'),
            eventer=Mock(name='eventer'),
            app=self.app,
            connection_errors=(socket.error,),
            task=sig.type,
            headers=headers,
            **kwargs
        )

    def test_shadow(self):
        assert self.get_request(