test_async_subprocess.py

import sys
import shlex
import asyncio
import os

async def run_command_async(cmd):
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
        )
    return proc


async def stdout(proc):
    '''
    Redirects stdout in binary form to named pipe stdout.
    The program runs until the pipe was read.
    '''
    try:
        fifo = os.mkfifo('stdout')
    except FileExistsError:
        pass
    with open('stdout', 'wb') as file:
        while True:
            data = await proc.stdout.read()
            if not data:
                break
            # print(data.decode(), file=sys.stdout)
            file.write(data)


async def stderr(proc):
    '''
    Redirects stderr as utf8 to the named pipe stderr.
    '''
    try:
        fifo = os.mkfifo('stderr')
    except FileExistsError:
        pass
    with open('stderr', 'w') as file:
        while True:
            data = await proc.stderr.read()
            if not data:
                break
            #print(data.decode(), file=sys.stderr)
            file.write(data.decode())


async def main(cmd):
    cmd = ' '.join(cmd)
    proc = await run_command_async(cmd)
    #await stdout(proc)
    err = stderr(proc)
    out = stdout(proc)
    err_task = asyncio.create_task(err)
    out_task = asyncio.create_task(out)
    done, pending = await asyncio.wait({err_task, out_task})
    #for task in done:
    #    await task

loop = asyncio.get_event_loop()
#loop.run_until_complete(main(sys.argv[1:]))
asyncio.run(main(sys.argv[1:]))