python - 如何在非缓冲模式下运行 Perl 的 `` prove`` TAP harness？

Question

作为在 Linux 上用 Python 3[.4-.6] 编写的测试套件的一部分，我必须运行一些第 3 方测试。第 3 方测试是 bash 脚本。它们旨在与 Perl's prove TAP harness 一起运行.一个 bash 脚本可以包含多达数千个单独的测试——其中一些可以无限期地挂起。超时后，我想终止测试脚本并收集有关卡住位置的一些信息。

因为 bash 脚本会创建自己的进程，所以我尝试将整个 prove 进程树隔离到一个新的进程组中，这样我最终可以将整个进程组作为一个整体杀死错误的。因为测试必须以 root 权限运行，所以我使用 sudo -b 创建一个具有 root 权限的新进程组。这种策略(与以一种或另一种方式使用 setsid 相反)是我在 this question at SE Unix&Linux 上收到的评论的结果。

问题是，如果我在通过 Python 的 子进程使用 sudo -b 启动时“过早地”杀死它，我会丢失 prove TAP 工具的所有输出。打开。

我将它隔离成一个简单的测试用例。下面是一个名为job.t的bash测试脚本:

#!/bin/bash

MAXCOUNT=20
echo "1..$MAXCOUNT"
for (( i=1; i<=$MAXCOUNT; i++ ))
do
   echo "ok $i"
   sleep 1
done

为了比较，我还编写了一个名为 job.py 的 Python 脚本，它产生或多或少相同的输出并表现出相同的行为:

import sys
import time
if __name__ == '__main__':
    maxcount = 20
    print('1..%d' % maxcount)
    for i in range(1, maxcount + 1):
        sys.stdout.write('ok %d\n' % i)
        time.sleep(1)

最后但同样重要的是，以下是我名为 demo.py 的精简版“Python 测试基础设施”:

import psutil # get it with "pip install psutil"
import os
import signal
import subprocess

def run_demo(cmd, timeout_after_seconds, signal_code):
    print('DEMO: %s' % ' '.join(cmd))
    proc = subprocess.Popen(cmd, stdout = subprocess.PIPE, stderr = subprocess.PIPE)
    try:
        outs, errs = proc.communicate(timeout = timeout_after_seconds)
    except subprocess.TimeoutExpired:
        print('KILLED!')
        kill_pid = _get_pid(cmd)
        subprocess.Popen(['sudo', 'kill', '-%d' % signal_code, '--', '-%d' % os.getpgid(kill_pid)]).wait()
        outs, errs = proc.communicate()
    print('Got our/err:', outs.decode('utf-8'), errs.decode('utf-8'))

def _get_pid(cmd_line_list):
    for pid in psutil.pids():
        proc = psutil.Process(pid)
        if cmd_line_list == proc.cmdline():
            return proc.pid
    raise # TODO some error ...

if __name__ == '__main__':
    timeout_sec = 5
    # Works, output is captured and eventually printed
    run_demo(['sudo', '-b', 'python', 'job.py'], timeout_sec, signal.SIGINT)
    # Failes, output is NOT captured (i.e. printed) and therefore lost
    run_demo(['sudo', '-b', 'prove', '-v', os.path.join(os.getcwd(), 'job.t')], timeout_sec, signal.SIGINT)

启动 demo.py 时，它会运行例程 run_demo 两次 - 使用不同的配置。两次都启动了一个具有 root 权限的新进程组。两次，“测试作业”每秒打印一次新行 (ok [line number]) - 理论上为 20 秒/20 行。但是，这两个脚本都有 5 秒的超时，并且整个进程组在该超时后被终止。

当 run_demo 第一次运行我的小 Python 脚本 job.py 时，该脚本的所有输出一直到它被杀死时都是捕获并打印成功。当 run_demo 在 prove 之上使用演示 bash 测试脚本 job.t 第二次运行时，没有捕获输出，只有空字符串被打印出来。

user@computer:~> python demo.py 
DEMO: sudo -b python job.py
KILLED!
Got our/err: 1..20
ok 1
ok 2
ok 3
ok 4
ok 5
ok 6
 Traceback (most recent call last):
  File "job.py", line 11, in <module>
    time.sleep(1)
KeyboardInterrupt

DEMO: sudo -b prove -v /full/path/to/job.t
KILLED!
Got our/err:  
user@computer:~>

这里发生了什么，我该如何解决？

即如何中断/终止使用 prove(及其整个进程组)运行的 bash 测试脚本，以便我可以捕获其输出？

编辑:它是 suggested in an answer观察到的行为的发生是因为 Perl 缓冲了它的输出。在单个 Perl 脚本中，可以将其关闭。但是，没有明显的选项允许关闭 prove [-v] 的缓冲。我怎样才能做到这一点？

---

我可以通过直接使用 bash 运行我的测试作业来解决这个问题。以下命令必须从

run_demo(['sudo', '-b', 'prove', '-v', os.path.join(os.getcwd(), 'job.t')], timeout_sec, signal.SIGINT)

到

run_demo(['sudo', '-b', 'bash', os.path.join(os.getcwd(), 'job.t')], timeout_sec, signal.SIGINT)

这样，我就不会得到 prove 打印的测试统计信息，但我可以自己生成它们。

Answer 1

默认情况下，当 STDOUT 连接到终端时，许多程序(包括 perl)的 STDOUT 是行缓冲的(在换行符上刷新)，并且是 block 缓冲的(在文件缓冲区时刷新)已满)否则(例如，当它连接到管道时)。

您可以通过使用伪 tty (ptty) 而不是管道来欺骗此类程序使用行缓冲。为此，unbuffer 是您的 friend 。在 Ubuntu 上，这是 expect 包的一部分(sudo apt install expect)。

来自docs :

unbuffer disables the output buffering that occurs when program output is redirected from non-interactive programs. For example, suppose you are watching the output from a fifo by running it through od and then more.

od -c /tmp/fifo | more

You will not see anything until a full page of output has been produced.

You can disable this automatic buffering as follows:

unbuffer od -c /tmp/fifo | more

我尝试了您的示例代码并得到了与您描述的相同的结果(感谢您的 Minimal, Complete, and Verifiable example !)。

然后我改变了

run_demo(['sudo', '-b', 'prove', '-v', os.path.join(os.getcwd(), 'job.t')], timeout_sec, signal.SIGINT)

到

run_demo(['sudo', '-b', 'unbuffer', 'prove', '-v', os.path.join(os.getcwd(), 'job.t')], timeout_sec, signal.SIGINT)

也就是说:我只是将 unbuffer 添加到 prove 命令之前。然后输出是:

DEMO: sudo -b python job.py
KILLED!
Got our/err: 1..20
ok 1
ok 2
ok 3
ok 4
ok 5
ok 6
 Traceback (most recent call last):
  File "job.py", line 8, in <module>
    time.sleep(1)
KeyboardInterrupt

DEMO: sudo -b unbuffer prove -v /home/dirk/w/sam/p/job.t
KILLED!
Got our/err: /home/dirk/w/sam/p/job.t .. 
1..20
ok 1
ok 2
ok 3
ok 4
ok 5