Захват вывода из subprocess.call, что у меня нет контроля над

Question

Я тестирую кусок кода Python, который использует subprocess.call(), поэтому я не могу контролировать этот вызов функции. Мне нужно сделать вывод из этого системного вызова для выполнения утверждений. Я попытался установить os.stdout для объекта StringIO, но это не отображает выходы системного вызова. Как решить эту проблему? Вот мой код до сих пор:

Код для тестирования (я не имею никакого контроля над этим):

def runFile(filename): 
    check_call("./" + filename) 

Моя попытка захвата вывода системного вызова:

import StringIO 
oldout = sys.stdout 
try: 
    sys.stdout = StringIO.StringIO() 
    runFile("somefile") 
    output = sys.stdout.getvalue() 
    assert output == "expected-output" 
finally: 
    sys.stdout = oldout 

Answer 1

Вы можете перенаправить свой собственный стандартный вывод на

rd, wr = os.pipe() 
oldstdout = os.dup(1) 
os.dup2(wr, 1) 
os.close(wr) 

, а затем читать из rd в то время как внешний процесс работает.

Это может не сработать, поскольку внешний процесс может писать больше, чем подходит в буфере. В этом случае вам придется порождать поток чтения.

После этого, вы восстановить старое состояние с

os.dup2(oldstdout, 1) 
os.close(oldstdout) 
os.close(rd) 

и продолжают нормально.

Answer 2

Модуль subprocess непосредственно записывает выходной поток с использованием os.write, поэтому изменение sys.stdout ничего не сделает.

Вот почему вы обычно должны указывать выходной поток, используемый с модулем подпроцесса, с использованием нотации ...([cmd, arg, ...], stdout=output_file, ...) и почему вы не можете использовать StringIO в качестве выходного файла, поскольку он не имеет fileno для записи.

Таким образом, единственный способ достичь того, что вы хотите сделать, - o monkeypatch subproces.check_call перед импортом модуля, который вам нужно проверить, чтобы он работал немного больше, как check_output.

Например:

import subprocess 

def patched_call(*popenargs, **kwargs): 
    if 'stdout' in kwargs: 
     raise ValueError('stdout argument not allowed, it will be overridden.') 
    process = subprocess.Popen(*popenargs, stdout=subprocess.PIPE, **kwargs) 
    output, unused_err = process.communicate() 
    # do something with output here 
    retcode = process.poll() 
    if retcode: 
     cmd = kwargs.get("args") 
     if cmd is None: 
      cmd = popenargs[0] 
     raise CalledProcessError(retcode, cmd, output=output) 
    return retcode 

subprocess.check_call = patched_call