def _run_sync ( self , method : Callable , * args , * * kwargs ) -> Any : if self . loop . is_running ( ) : raise RuntimeError ( "Event loop is already running." ) if not self . is_connected : self . loop . run_until_complete ( self . connect ( ) ) task = asyncio . Task ( method ( * args , * * kwargs ) , loop = self . loop ) result = self . loop . run_until_complete ( task ) self . loop . run_until_complete ( self . quit ( ) ) return result
python asyncio behave different in linux window
Utility method to run commands synchronously for testing .
def _run_sync(self, method: Callable, *args, **kwargs) -> Any:
        """
        Utility method to run commands synchronously for testing.
        """
        if self.loop.is_running():
            raise RuntimeError("Event loop is already running.")

        if not self.is_connected:
            self.loop.run_until_complete(self.connect())

        task = asyncio.Task(method(*args, **kwargs), loop=self.loop)
        result = self.loop.run_until_complete(task)

        self.loop.run_until_complete(self.quit())

        return result