如何异步捕获长时间运行的 Python 子进程的输出

Question

对于一个测试，我想在 Linux 上启动一个长时间运行的 Python 3 脚本，捕获它的一些输出，检查它是否符合预期并再次终止它。

我现在只是想让通用框架正常工作。

当我运行以下代码时，我希望输出包含line = "0"——python 脚本只是打印出递增的整数序列，每秒一个。而是完全跳过 for 循环。

import subprocess
from tempfile import NamedTemporaryFile
import time
import unittest

import asynchronousfilereader

class TestProcessSpawning(unittest.TestCase):
    def test_spawning_counter(self):
        counter_code = \
"""
import time
i = 0
while True:
    print(i)
    i = i + 1
    time.sleep(1)
"""

        with NamedTemporaryFile(mode="w", suffix='.py', delete=False)\
                as temp_file:
            temp_file.write(counter_code)
            file_name = temp_file.name

            # proc = subprocess.Popen(['ping', 'localhost'],
            #                         stdout=subprocess.PIPE, close_fds=True)
            proc = subprocess.Popen(['python3', file_name],
                                    stdout=subprocess.PIPE, close_fds=True)

            time.sleep(3)

            assert proc.returncode is None  # None => still running

            reader = asynchronousfilereader.AsynchronousFileReader(proc.stdout)

            time.sleep(3)  # give it a chance?

            for line in reader.readlines():
                print('line = "{}"'.format(line))
                break  # just grab first one

            proc.kill()

但是，如果我将 ['python3', file_name] 更改为 ['ping', 'localhost']，我 确实 会从 ping 中得到一行输出（我在 Linux 上运行，所以 ping 会一直产生输出，直到你停止它） .

知道为什么这似乎适用于其他类型的子进程但不适用于 python？

---

注意事项：

• 此代码使用 pip 提供的asynchronousfilereader
• 如果我只是在 shell 中运行 python3 <temp_file_name>，脚本会按预期运行，打印 0、1、2、...
• 几秒钟后None 的进程返回码表明进程正在运行（即不只是崩溃）
• 我已尝试按照here 的建议通过unbuffer 运行它：输出没有出现
• 我尝试将bufsize=1 传递给Popen：输出没有出现
• 如果我包含 -v 标志，我会得到很多输出，包括这个，这表明 python 正在尝试以交互方式运行，即使我提供了文件名：

Python 3.5.2（默认，2017 年 11 月 23 日，16:37:01）

[GCC 5.4.0 20160609] 在 Linux 上

输入“帮助”、“版权”、“信用”或“许可”以了解更多信息。

Answer 1

你需要刷新你写的文件：

with NamedTemporaryFile(mode="w", suffix='.py', delete=False) as temp_file:
    temp_file.write(counter_code)
    temp_file.flush()

您的程序太短而无法从缓冲区中取出，并且您正在运行一个空文件。

我还要确保程序在每次写入时都会刷新；对无缓冲标准输出运行 python3 -u 或在每次打印时使用 flush=True。这样您就不必在父程序中添加这么多的睡眠。

通过这些更改，单元测试打印出来

line = "b'0\n'"

还有几个关于子进程仍在运行且标准输出缓冲区在完成之前仍处于打开状态的警告。