Python 和 Pytest 调用包含 tail 的 shell 命令失败

Question

环境： 平台 qnx——Python 2.7.12、pytest-4.6.9、py-1.8.1、pluggy-0.13.1 插件：json-report-1.2.1、shell-0.2.3

注意： 我知道 Python2.7 很旧且不受支持，但目前 QNX 没有其他版本可用。

问题：

我正在运行一个测试，当某个关键字出现在其日志中时，它应该终止服务。为此，我需要它在后台运行。 为此，我使用以下 shell 命令：

def test_kill_process():
    expected_output="XXXXXXXXX"
    expected_rc=0
    check_kill_process(expected_output, expected_rc)

import os
def check_kill_process(expected_output, expected_rc):
    test_log = File(r"/path/to/log")
    erase_log_entry = "Action"
    service=MyService()
    service.start()
    sleep(2)
    kill_command = "tail -f " + test_log.file_path + " | grep --line-buffered " + erase_log_entry + \
            " | while read ; do kill " + service.pid + " ; done &"
    os.popen(kill_command)
    service.action()
    f = open(test_log.file_path, "r")
    output = f.read()
    assert re.search(expected_output, output)

================================================ ==========================

没有 Pytest 甚至 Python，就像一个魅力。

如果我尝试使用 subprocess 模块运行命令，测试会无限期冻结。 如果我尝试使用 os.popen 或 os.system，命令以错误结束：

tail:  read failed in '/path/to/logfile' (Invalid argument)

此外，如果我尝试同样的事情，只有一只“猫”我会得到这个：

--stdout--: Broken pipe

如果有人有任何想法，请提前感谢！

Answer 1

根据经验，subprocess 和 popen 喜欢命令序列而不是字符串。我认为技术原因是因为不同的系统如何解释字符串路径与参数。查看popen documetation 以获得更深入的解释。

但是例如，如果我想执行命令git commit -m "fixes a bug."，我将不得不将该字符串命令拆分为一个以空格分隔的列表。

# command as a string
cmd = 'git commit -m "fixes a bug."'

# command as sequence of arguments. Consider using shlex.split() instead
cmd = cmd.split(' ')

# now call popen with the new formatted arguments
os.popen(cmd)

使您的脚本更加健壮。对于更复杂的情况，您可以使用内置函数 shlex.split() 将字符串拆分为类似 shell 的语法。

此外，您似乎也在命令中使用| 管道。在调用 popen 时，您必须通过 shell=True 才能使管道正常工作。

Answer 2

不要使用subprocess（这通常是单一python的症状），为什么不直接使用open(logfile, 'r')并在循环中使用file.readline()来读取文件？这将与使用 tail 产生相同的效果，而无需子进程的开销（生成的代码也更清晰，更易于理解）。

这应该非常适合您尝试做的事情：

def wait_for_line(filename):
    f = open(filename, 'r')
    while 1:
        if line := f.readline():
            if re.match(some_regex, line):
                return True
        else:
            sleep(0.5)

这是一个完整的工作示例：

import threading
from time import sleep
import re

FILENAME = 'some_file'
REGEX = r'^foo\s...!\s\d+$'
REGEX_MATCH = "foo bar! 123"

def perform_other_action(filename):
    f = open(filename, 'w')
    sleep(5) # just for effect
    f.write(REGEX_MATCH)


def wait_for_line(filename):
    f = open(filename, 'r')
    while 1:
        if line := f.readline():
            if re.match(REGEX, line):
                return True
        else:
            sleep(0.5) # throttle the read frequency if no new lines are found

with open(FILENAME, 'w') as f: # make sure file exists
    pass

t = threading.Thread(target=wait_for_line, args=(FILENAME,))
t.start() # returns immediately, wait_for_line runs in background
perform_other_action(FILENAME)
t.join()