QNX平台下Python+Pytest调用含tail的Shell命令失败问题

我来帮你梳理下这个问题的根源，结合QNX和Python2.7的特殊环境，给你几个可行的解决思路：

先明确下你的环境和问题 #

环境信息 #

• 平台：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

核心问题 #

你需要实现“监控服务日志，出现特定关键词就终止服务”的逻辑，直接在Shell里运行相关命令没问题，但用Python调用时遇到了各种异常：

• 用subprocess执行时测试无限冻结
• 用os.popen/os.system时报tail: read failed in '/path/to/logfile' (Invalid argument)
• 替换tail为cat时出现--stdout--: Broken pipe错误

你的代码逻辑如下：

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)

可行的解决思路 #

1. 用Python原生逻辑替代Shell管道（最推荐） #

Shell管道在Python子进程中经常因为IO缓冲、进程退出顺序的问题出bug，尤其是QNX这类非Linux系统对管道的处理有差异。完全用Python代码实现监控逻辑，可控性强，还能避免兼容性问题：

import time
import re
import os
import signal
import threading

def check_kill_process(expected_output, expected_rc):
    test_log = File(r"/path/to/log")
    erase_log_entry = "Action"
    service=MyService()
    service.start()
    time.sleep(2)

    # 定义后台监控线程的逻辑
    def monitor_log():
        with open(test_log.file_path, 'r') as log_file:
            # 先跳到文件末尾，跳过已有的旧日志
            log_file.seek(0, 2)
            while True:
                line = log_file.readline()
                if not line:
                    # 没有新内容时短暂休眠，避免占用过多CPU
                    time.sleep(0.1)
                    continue
                # 匹配到目标关键词就终止服务
                if re.search(erase_log_entry, line):
                    os.kill(int(service.pid), signal.SIGTERM)
                    break

    # 启动后台监控线程，设置为守护线程随主进程退出
    monitor_thread = threading.Thread(target=monitor_log)
    monitor_thread.daemon = True
    monitor_thread.start()

    # 触发服务动作
    service.action()
    # 等待监控线程完成，设置超时避免无限等待
    monitor_thread.join(timeout=10)

    # 读取日志并验证结果
    with open(test_log.file_path, "r") as f:
        output = f.read()
    assert re.search(expected_output, output)

这种方式不需要依赖Shell工具，在Python2.7下完全兼容，还能更灵活地处理日志匹配逻辑。

2. 修正Shell命令的调用方式（如果一定要用Shell） #

如果坚持用Shell管道，需要解决QNX环境下的几个特殊问题：

• QNX的tail -f和Linux实现有差异，试试用tail -F（大写F），它会在日志文件被轮转后自动重新打开，可能解决"Invalid argument"的错误
• 用subprocess.Popen时要正确处理IO流，避免因为管道未关闭导致进程挂起，同时创建进程组方便后续清理残留进程

示例代码：

import time
import os
import signal
import subprocess

def check_kill_process(expected_output, expected_rc):
    test_log = File(r"/path/to/log")
    erase_log_entry = "Action"
    service=MyService()
    service.start()
    time.sleep(2)

    # 格式化Shell命令，用tail -F替代tail -f
    kill_command = """tail -F {log_path} | grep --line-buffered {entry} | while read ; do kill {pid} ; done""".format(
        log_path=test_log.file_path,
        entry=erase_log_entry,
        pid=service.pid
    )

    # 启动子进程，重定向输出到/dev/null，创建进程组方便后续清理
    proc = subprocess.Popen(
        kill_command,
        shell=True,
        stdout=open(os.devnull, 'w'),
        stderr=open(os.devnull, 'w'),
        preexec_fn=os.setsid
    )

    try:
        # 触发服务动作
        service.action()
        # 等待动作完成，可根据实际情况调整超时
        time.sleep(5)
        # 杀死整个进程组，避免残留后台进程
        os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
    finally:
        # 读取剩余输出，避免产生僵尸进程
        proc.communicate()

    # 验证日志结果
    with open(test_log.file_path, "r") as f:
        output = f.read()
    assert re.search(expected_output, output)

3. 排查QNX系统的特殊限制 #

QNX的文件系统和进程模型和Linux不同，出现错误可能和系统特性有关：

• 确认日志文件的权限，确保Python进程有读取权限
• 如果QNX支持procfs，可以尝试通过cat /proc/<service_pid>/fd/<log_fd>的方式读取服务的实时日志，避免直接访问日志文件的问题
• 检查grep --line-buffered在QNX上是否支持，若不支持，试试用stdbuf -oL grep强制开启行缓冲（如果QNX提供stdbuf工具）

内容的提问来源于stack exchange，提问作者Razvan Badea