def METHOD_NAME(session_mocker):
    mock = session_mocker.mock_module
    class FakeConnection(mock.NonCallableMagicMock):
        callbacks = []
        server_info = ServerInfo("127.0.0.1", (4, 3))
        local_port = 1234
        def __init__(self, *args, **kwargs):
            kwargs["spec"] = Bolt
            super().__init__(*args, **kwargs)
            self.attach_mock(mock.Mock(return_value=True), "is_reset_mock")
            self.attach_mock(mock.Mock(return_value=False), "defunct")
            self.attach_mock(mock.Mock(return_value=False), "stale")
            self.attach_mock(mock.Mock(return_value=False), "closed")
            self.attach_mock(mock.Mock(return_value=False), "socket")
            self.attach_mock(mock.Mock(return_value=False), "re_auth")
            self.attach_mock(mock.MagicMock(spec=AuthManager),
                             "auth_manager")
            self.unresolved_address = next(iter(args), "localhost")
            def close_side_effect():
                self.closed.return_value = True
            self.attach_mock(mock.MagicMock(side_effect=close_side_effect),
                             "close")
            self.socket.attach_mock(
                mock.Mock(return_value=None), "get_deadline"
            )
            def set_deadline_side_effect(deadline):
                deadline = Deadline.from_timeout_or_deadline(deadline)
                self.socket.get_deadline.return_value = deadline
            self.socket.attach_mock(
                mock.Mock(side_effect=set_deadline_side_effect), "set_deadline"
            )
        @property
        def is_reset(self):
            if self.closed.return_value or self.defunct.return_value:
                raise AssertionError(
                    "is_reset should not be called on a closed or defunct "
                    "connection."
                )
            return self.is_reset_mock()
        def fetch_message(self, *args, **kwargs):
            if self.callbacks:
                cb = self.callbacks.pop(0)
                cb()
            return super().__getattr__("fetch_message")(*args, **kwargs)
        def fetch_all(self, *args, **kwargs):
            while self.callbacks:
                cb = self.callbacks.pop(0)
                cb()
            return super().__getattr__("fetch_all")(*args, **kwargs)
        def __getattr__(self, name):
            parent = super()
            def build_message_handler(name):
                def func(*args, **kwargs):
                    def callback():
                        for cb_name, param_count in (
                            ("on_success", 1),
                            ("on_summary", 0)
                        ):
                            cb = kwargs.get(cb_name, None)
                            if callable(cb):
                                try:
                                    param_count = \
                                        len(inspect.signature(cb).parameters)
                                except ValueError:
                                    # e.g. built-in method as cb
                                    pass
                                if param_count == 1:
                                    res = cb({})
                                else:
                                    res = cb()
                                try:
                                    res  # maybe the callback is async
                                except TypeError:
                                    pass  # or maybe it wasn't ;)
                    self.callbacks.append(callback)
                return func
            method_mock = parent.__getattr__(name)
            if name in ("run", "commit", "pull", "rollback", "discard"):
                method_mock.side_effect = build_message_handler(name)
            return method_mock
    return FakeConnection