def _patch_asyncio() -> None:
    # This patches asyncio to add a sync_wait method to the event
    # loop. This method can then be called from within a task
    # including a synchronous function called from a task. Sadly it
    # requires the python Task and Future implementations, which
    # invokes some performance cost.
    asyncio.Task = asyncio.tasks._CTask = asyncio.tasks.Task = asyncio.tasks._PyTask  # type: ignore
    asyncio.Future = (  # type: ignore
        asyncio.futures._CFuture  # type: ignore
    ) = asyncio.futures.Future = asyncio.futures._PyFuture  # type: ignore # noqa

    current_policy = asyncio.get_event_loop_policy()
    if hasattr(asyncio, "unix_events"):
        target_policy = asyncio.unix_events._UnixDefaultEventLoopPolicy
    else:
        target_policy = object  # type: ignore

    if not isinstance(current_policy, target_policy):
        raise RuntimeError("Flask Patching only works with the default event loop policy")

    _patch_loop()
    _patch_task() 

def _parse_sdcard_list(self, done_cb):
        self.log.debug('Comms: _parse_sdcard_list')

        # setup callback to receive and parse listing data
        files = []
        f = asyncio.Future()
        self.redirect_incoming(lambda x: self._rcv_sdcard_line(x, files, f))

        # issue command
        self._write('M20\n')

        # wait for it to complete and get all the lines
        # add a long timeout in case it fails and we don't want to wait for ever
        try:
            await asyncio.wait_for(f, 10)

        except asyncio.TimeoutError:
            self.log.warning("Comms: Timeout waiting for sd card list")
            files = []

        self.redirect_incoming(None)

        # call upstream callback with results
        done_cb(files) 

def wait_for_iterator(self, connection_observer, connection_observer_future):
        """
        Version of wait_for() intended to be used by Python3 to implement awaitable object.

        Note: we don't have timeout parameter here. If you want to await with timeout please do use asyncio machinery.
        For ex.:  await asyncio.wait_for(connection_observer, timeout=10)

        :param connection_observer: The one we are awaiting for.
        :param connection_observer_future: Future of connection-observer returned from submit().
        :return: iterator
        """
        self.logger.debug("go foreground: {!r}".format(connection_observer))

        # assuming that connection_observer.start() / runner.submit(connection_observer)
        # has already scheduled future via asyncio.ensure_future
        assert asyncio.futures.isfuture(connection_observer_future)

        return connection_observer_future.__iter__()
        # Note: even if code is so simple we can't move it inside ConnectionObserver.__await__() since different runners
        # may provide different iterator implementing awaitable
        # Here we know, connection_observer_future is asyncio.Future (precisely asyncio.tasks.Task)
        # and we know it has __await__() method. 

def run_async_coroutine(self, coroutine_to_run, timeout):
        """Start coroutine in dedicated thread and await its result with timeout"""
        start_time = time.time()
        coro_future = self.start_async_coroutine(coroutine_to_run)
        # run_coroutine_threadsafe returns future as concurrent.futures.Future() and not asyncio.Future
        # so, we can await it with timeout inside current thread
        try:
            coro_result = coro_future.result(timeout=timeout)
            self.logger.debug("scheduled {} returned {}".format(coroutine_to_run, coro_result))
            return coro_result
        except concurrent.futures.TimeoutError:
            passed = time.time() - start_time
            raise MolerTimeout(timeout=timeout,
                               kind="run_async_coroutine({})".format(coroutine_to_run),
                               passed_time=passed)
        except concurrent.futures.CancelledError:
            raise 

def data_received(self, data, recv_time):
        """
        Await initial prompt of started shell command.

        After that - do real forward
        """
        if not self._shell_operable.done():
            decoded_data = self.moler_connection.decode(data)
            self.logger.debug("<|{}".format(data))
            assert isinstance(decoded_data, str)
            self.read_buffer += decoded_data
            if re.search(self.prompt, self.read_buffer, re.MULTILINE):
                self._notify_on_connect()
                self._shell_operable.set_result(True)  # makes Future done
                # TODO: should we remove initial prompt as it is inside raw.terminal?
                # TODO: that way (maybe) we won't see it in logs
                data_str = re.sub(self.prompt, '', self.read_buffer, re.MULTILINE)
                data = self.moler_connection.encode(data_str)
            else:
                return
        self.logger.debug("<|{}".format(data))
        super(AsyncioTerminal, self).data_received(data) 

def channel(key=None):
  global _channel_futures
  if key:
    if key in _channel_futures:
      return await _channel_futures[key]
    future = asyncio.Future()
    _channel_futures[key] = future
  try:
    channel = await (await _connect()).channel()
    if key:
      future.set_result(channel)
      asyncio.get_event_loop().create_task(_wait_channel(channel, key))
    return channel
  except Exception as e:
    future.set_exception(e)
    del _channel_futures[key]
    raise 

def wait_for(self, event, predicate, result=None):
        """Waits for a DISPATCH'd event that meets the predicate.

        Parameters
        -----------
        event: :class:`str`
            The event name in all upper case to wait for.
        predicate
            A function that takes a data parameter to check for event
            properties. The data parameter is the 'd' key in the JSON message.
        result
            A function that takes the same data parameter and executes to send
            the result to the future. If ``None``, returns the data.

        Returns
        --------
        asyncio.Future
            A future to wait for.
        """

        future = self.loop.create_future()
        entry = EventListener(event=event, predicate=predicate, result=result, future=future)
        self._dispatch_listeners.append(entry)
        return future 

def admin_apps_approve(
        self, *, app_id: str = None, request_id: str = None, **kwargs
    ) -> Union[Future, SlackResponse]:
        """Approve an app for installation on a workspace.

        Either app_id or request_id is required.
        These IDs can be obtained either directly via the app_requested event,
        or by the admin.apps.requests.list method.

        Args:
            app_id (str): The id of the app to approve. e.g. 'A12345'
            request_id (str): The id of the request to approve. e.g. 'Ar12345'
        Raises:
            SlackRequestError: If neither or both the `app_id` and `request_id` args are specified.
        """
        if app_id:
            kwargs.update({"app_id": app_id})
        elif request_id:
            kwargs.update({"request_id": request_id})
        else:
            raise e.SlackRequestError(
                "The app_id or request_id argument must be specified."
            )

        return self.api_call("admin.apps.approve", json=kwargs) 

def admin_teams_settings_setDefaultChannels(
        self, *, team_id: str, channel_ids: Union[str, List[str]], **kwargs
    ) -> Union[Future, SlackResponse]:
        """Set the default channels of a workspace.

        Args:
            team_id (str): ID of the team.
            channel_ids (str or list): A list of channel_ids.
                At least one channel is required. e.g. ['C1A2B3C4D', 'C26Z25Y24']
        """
        kwargs.update({"team_id": team_id})
        if isinstance(channel_ids, list):
            kwargs.update({"channel_ids": ",".join(channel_ids)})
        else:
            kwargs.update({"channel_ids": channel_ids})
        return self.api_call(
            "admin.teams.settings.setDefaultChannels", http_verb="GET", params=kwargs
        ) 

def admin_usergroups_addChannels(
        self,
        *,
        team_id: str,
        usergroup_id: str,
        channel_ids: Union[str, List[str]],
        **kwargs
    ) -> Union[Future, SlackResponse]:
        """Add one or more default channels to an IDP group.

        Args:
            team_id (str): The workspace to add default channels in. e.g. 'T1234'
            usergroup_id (str): ID of the IDP group to add default channels for. e.g. 'S1234'
            channel_ids (str or list): Comma separated string of channel IDs. e.g. 'C123,C234' or ['C123', 'C234']
        """
        kwargs.update({"team_id": team_id, "usergroup_id": usergroup_id})
        if isinstance(channel_ids, list):
            kwargs.update({"channel_ids": ",".join(channel_ids)})
        else:
            kwargs.update({"channel_ids": channel_ids})
        return self.api_call("admin.usergroups.addChannels", json=kwargs) 

def admin_users_invite(
        self, *, team_id: str, email: str, channel_ids: Union[str, List[str]], **kwargs
    ) -> Union[Future, SlackResponse]:
        """Invite a user to a workspace.

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
            email (str): The email address of the person to invite. e.g. 'joe@email.com'
            channel_ids (str or list): A list of channel_ids for this user to join.
                At least one channel is required. e.g. ['C1A2B3C4D', 'C26Z25Y24']
        """
        kwargs.update({"team_id": team_id, "email": email})
        if isinstance(channel_ids, list):
            kwargs.update({"channel_ids": ",".join(channel_ids)})
        else:
            kwargs.update({"channel_ids": channel_ids})
        return self.api_call("admin.users.invite", json=kwargs) 

def myFuture(future):
  await asyncio.sleep(1)
  future.set_result("My Future Has Completed") 

def main():
  future = asyncio.Future()
  await asyncio.ensure_future(myFuture(future))
  print(future.result()) 

def __init__(self, *args, **kwargs):
        self._loop = kwargs['loop'] if 'loop' in kwargs \
            else asyncio.get_event_loop()
        self._is_running = False
        self._run_complete = asyncio.Future(loop = self._loop) 

def ask(self, target, message):
        assert isinstance(message, QueryMessage)
        if not message.result:
            message.result = asyncio.Future(loop = self._loop)
          
        yield from self.tell(target, message)
        return (yield from message.result) 

def __init__(self, remote_queue: asyncio.Queue) -> None:
        self.remote_queue = remote_queue
        self.local_queue: asyncio.Queue = asyncio.Queue()
        self.accepted = False
        self.task: Optional[asyncio.Future] = None 

def _cancel_tasks(tasks: Set[asyncio.Future]) -> None:
    # Cancel any pending, and wait for the cancellation to
    # complete i.e. finish any remaining work.
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    _raise_exceptions(tasks) 

def _raise_exceptions(tasks: Set[asyncio.Future]) -> None:
    # Raise any unexcepted exceptions
    for task in tasks:
        if not task.cancelled() and task.exception() is not None:
            raise task.exception() 

def future_builder(return_val: object) -> Future:
    result = Future()
    result.set_result(return_val)
    return result 

def test_bot_telemetry_middleware(self):
        req = Mock()
        req.headers = {"Content-Type": "application/json"}
        req.json = MagicMock(return_value=Future())
        req.json.return_value.set_result("mock body")

        async def handler(value):
            return value

        sut = await bot_telemetry_middleware(req, handler)

        assert "mock body" in aiohttp_telemetry_middleware._REQUEST_BODIES.values()
        aiohttp_telemetry_middleware._REQUEST_BODIES.clear()
        assert req == sut 

def create_skill_handler_for_testing(self, adapter) -> SkillHandlerInstanceForTests:
        mock_bot = Mock()
        mock_bot.on_turn = MagicMock(return_value=Future())
        mock_bot.on_turn.return_value.set_result(Mock())

        return SkillHandlerInstanceForTests(
            adapter,
            mock_bot,
            self._test_id_factory,
            Mock(),
            AuthenticationConfiguration(),
        ) 

def test_on_send_to_conversation(self):
        self._conversation_id = await self._test_id_factory.create_skill_conversation_id(
            self._conversation_reference
        )

        mock_adapter = Mock()
        mock_adapter.continue_conversation = MagicMock(return_value=Future())
        mock_adapter.continue_conversation.return_value.set_result(Mock())
        mock_adapter.send_activities = MagicMock(return_value=Future())
        mock_adapter.send_activities.return_value.set_result([])

        sut = self.create_skill_handler_for_testing(mock_adapter)

        activity = Activity(type=ActivityTypes.message, attachments=[], entities=[])
        TurnContext.apply_conversation_reference(activity, self._conversation_reference)

        assert not activity.caller_id

        await sut.test_on_send_to_conversation(
            self._claims_identity, self._conversation_id, activity
        )

        args, kwargs = mock_adapter.continue_conversation.call_args_list[0]

        assert isinstance(args[0], ConversationReference)
        assert callable(args[1])
        assert isinstance(kwargs["claims_identity"], ClaimsIdentity)

        await args[1](
            TurnContext(
                mock_adapter,
                conversation_reference_extension.get_continuation_activity(
                    self._conversation_reference
                ),
            )
        )
        assert activity.caller_id is None 

def test_on_reply_to_activity(self):
        self._conversation_id = await self._test_id_factory.create_skill_conversation_id(
            self._conversation_reference
        )

        mock_adapter = Mock()
        mock_adapter.continue_conversation = MagicMock(return_value=Future())
        mock_adapter.continue_conversation.return_value.set_result(Mock())
        mock_adapter.send_activities = MagicMock(return_value=Future())
        mock_adapter.send_activities.return_value.set_result([])

        sut = self.create_skill_handler_for_testing(mock_adapter)

        activity = Activity(type=ActivityTypes.message, attachments=[], entities=[])
        activity_id = str(uuid4())
        TurnContext.apply_conversation_reference(activity, self._conversation_reference)

        await sut.test_on_reply_to_activity(
            self._claims_identity, self._conversation_id, activity_id, activity
        )

        args, kwargs = mock_adapter.continue_conversation.call_args_list[0]

        assert isinstance(args[0], ConversationReference)
        assert callable(args[1])
        assert isinstance(kwargs["claims_identity"], ClaimsIdentity)

        await args[1](
            TurnContext(
                mock_adapter,
                conversation_reference_extension.get_continuation_activity(
                    self._conversation_reference
                ),
            )
        )
        assert activity.caller_id is None 

def wait_for_flow_control(self, stream_id):
        """
        Waits for a Future that fires when the flow control window is opened.
        """
        f = asyncio.Future()
        self.flow_control_futures[stream_id] = f
        await f 

def query_reader(ch: aioredis.Channel, cb: asyncio.Future):
        msg_list = []
        while await ch.wait_message():
            _, msg = await ch.get(encoding='utf-8')
            msg_dict = json.loads(msg)
            if 'empty' in msg_dict:
                if msg_dict['empty'] is False:
                    msg_list.append(msg_dict)
            else:
                msg_list.append(msg_dict)
            if ('bIsLast' not in msg_dict or msg_dict['bIsLast']) and not cb.done():
                cb.set_result(msg_list) 

def __init__(self,
                 port=8080,
                 prefix=None,
                 shutdown_grace_period=5,
                 shutdown_wait_period=1):
        """
         HTTP Transport for listening on http
         :param port: The port to lisen on (0.0.0.0 will always be used)
         :param prefix: the path prefix to remove from all url's
         :param shutdown_grace_period: Time to wait for server to shutdown
         before connections get forceably closed. The only way for connections
         to not be forcibly closed is to have some connection draining in front
         of the service for deploys. Most docker schedulers will do this for you.
         :param shutdown_wait_period: Time to wait after recieving the sigterm
         before starting shutdown 
         """
        self.port = port
        if prefix is None:
            prefix = ''
        self.prefix = prefix
        self._handler = None
        self._server = None
        self._loop = None
        self._done_future = asyncio.Future()
        self._connections = set()
        self.shutdown_grace_period = shutdown_grace_period
        self.shutdown_wait_period = shutdown_wait_period
        self.shutting_down = False
        self._config = {} 

def __init__(self, *, url, port=5672, queue='', virtualhost='/',
                 username='guest', password='guest',
                 ssl=False, verify_ssl=True, create_queue=True,
                 use_acks=False, heartbeat=20):
        super().__init__()
        self.host = url
        self.port = port
        self.virtualhost = virtualhost
        self.queue = queue
        self.username = username
        self.password = password
        self.ssl = ssl
        self.verify_ssl = verify_ssl
        self.create_queue = create_queue
        self._use_acks = use_acks
        self._transport = None
        self._protocol = None
        self.channel = None
        self._app = None
        self._loop = None
        self._consumer_tag = None
        self._counter = 0
        self._handler = None
        self._done_future = asyncio.Future()
        self._closing = False
        self._client = None
        self.heartbeat = heartbeat
        self._config = {}

        self.listeners = [] 

def _drain_helper(self):
        if self._connection_lost:
            raise ConnectionResetError('Connection lost')
        if not self._paused:
            return
        waiter = self._drain_waiter
        assert waiter is None or waiter.cancelled()
        waiter = asyncio.Future()
        self._drain_waiter = waiter
        await waiter 

def __init__(self, request, loop, after=None):
        self.container_id = None
        self.msg_id = None
        self.request = request
        self.data = bytes(request)
        self.future = asyncio.Future(loop=loop)
        self.after = after 

def disconnected(self: 'TelegramClient') -> asyncio.Future:
        """
        Property with a ``Future`` that resolves upon disconnection.

        Example
            .. code-block:: python

                # Wait for a disconnection to occur
                try:
                    await client.disconnected
                except OSError:
                    print('Error on disconnect')
        """
        return self._sender.disconnected