'Dispose asyncio.ProactorEventLoop() when finish
I'm running an automation tests with pytest. One of the first tests is updating Python packages. From time to time the subprocess stuck on subprocess.stdout.readline() So, I tried another solution I found using asyncio.ProactorEventLoop() It seems to work as expected but it is endless logging "Using proactor: IocpProactor". How do I disable this logging or alternatively, subprocess.stdout.readline() with working timeout (I have tried using Timer but it does not help since stdout.readline() is stuck)
def run_python_process(cmd: List[str], timeout: int, show_logs=True, logger=None):
if not show_logs:
logger = None
elif not logger:
logger = logging.getLogger()
if sys.platform == "win32":
loop = asyncio.ProactorEventLoop() # For subprocess' pipes on Windows
asyncio.set_event_loop(loop)
else:
loop = asyncio.get_event_loop()
return_code, stdout = loop.run_until_complete(run_command(' '.join(cmd), timeout=timeout, logger=logger))
loop.close()
return CompletedProcess([], return_code, stdout, None)
async def run_command(cmd, timeout=None, logger=None):
# Start child process
# NOTE: universal_newlines parameter is not supported
process = await asyncio.create_subprocess_shell(cmd, stdout=PIPE, stderr=STDOUT)
stdout = ''
# Read line (sequence of bytes ending with b'\n') asynchronously
while True:
try:
line = await asyncio.wait_for(process.stdout.readline(), timeout)
except asyncio.TimeoutError:
pass
else:
if not line: # EOF
break
else:
d_line = line.decode('ascii')
stdout += d_line
if logger:
logger.info(d_line.strip())
continue # While some criterium is satisfied
process.kill() # Timeout or some criterion is not satisfied
break
return await process.wait(), stdout # Wait for the child process to exit
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
| Solution | Source |
|---|
