<div dir="ltr">Hi all,<div><br></div><div>I&#39;m not sure if I&#39;m addressing this to the right mailing list, but I think a lot of you are into testing and Python so I&#39;ll try my luck anyways.</div><div><br></div><div>So I&#39;m trying out a pet project to learn asyncio and I&#39; having trouble writing test cases for this project and I&#39;m looking for some help getting started :-)</div><div><br></div><div>The pet project is a clone of Foreman/Honcho written with asyncio.  One of the building blocks I&#39;m having problems with is this &quot;run_once()&quot; coroutine which basically starts a subprocess, streams output (with timestamp and process label prefix) to stdout and kills the process on request to cancel.</div><div><br></div><div>At its core, I have the following loop:</div><div><br></div><blockquote style="margin:0px 0px 0px 40px;border:none;padding:0px">pending = {...}<br>while True:<br>  done, pending = yield from asyncio.wait(<br>    pending,<br>    return_when=asyncio.FIRST_COMPLETED,<br>  )<br>  for future in done:<br>    if future is x:<br>      # ...<br>    if future is y:<br>      # ...<br></blockquote><br><div>(This loop&#39;s structure is heavily inspired by my experience with ZMQ.  If you know of a better way to write this, please let me know.)</div><div><br></div><div>I want to test all possible code paths in the &quot;for future in done&quot; loop and I&#39;m trying to achieve that by sending a specific input, then waiting for the input to be processed.  Then, I can have each test will run a different path.</div><div><br class="">The path that shows that I&#39;m correctly streaming output to stdout with annotations is proving pretty hard to test since the &quot;run_once()&quot; coroutine does not complete until the child process exits.<br></div><div><br></div><div>The problem I&#39;m having is that I cannot figure out how to have the test and this coroutine alternate executions.  Existing solutions for testing asyncio code seems to work mostly with networked services where the test can simply connect a socket and act like a client (the test can trivially &quot;resp = loop.run_until_complete(client.sendall(req))&quot;).</div><div><br></div><div>So far, I&#39;ve been trying to emulate this by replacing sys.stdout with the write end of a socketpair() and the read end in my test to follow this pattern, but I&#39;m having trouble.</div><div><br></div><div>Any suggestions?</div><div><br></div><div>The answer is posted up on StackOverflow[1] if you want to answer there instead of replying to this email.</div><div><br></div><div>[1]: <a href="http://stackoverflow.com/q/34126076/313063">http://stackoverflow.com/q/34126076/313063</a></div><div><br></div><div>Thanks in advance,</div><div><br></div><div>André</div><div><br></div></div>