Source code for caspia.toolbox.ping

import asyncio


async def ping(ip_addr):
    """ Successully returns on success. Raises asyncio.TimeoutError if there is no response.
    Use asyncio.wait_for for explicit timeout.
    """
    process = await asyncio.create_subprocess_exec('ping',
                                                   ip_addr,
                                                   '-c',
                                                   '3',
                                                   stdout=asyncio.subprocess.PIPE,
                                                   stderr=asyncio.subprocess.PIPE)
    try:
        stdout, stderr = await process.communicate()
        if process.returncode:
            print(stdout)
            print(stderr)
            raise asyncio.TimeoutError(f'ping returned nonzero code {process.returncode}')
    except asyncio.CancelledError:
        process.kill()
        raise