在 Python 中拦截 subprocess.Popen 调用

Question

我正在为旧版 Python 脚本编写功能测试，以便我可以对其进行单行更改而不会因恐惧而瘫痪。 ;)

有问题的脚本使用 subprocess.Popen 调用 wget(1) 以下载 XML 文件，然后对其进行解析：

def download_files():
    os.mkdir(FEED_DIR)
    os.chdir(FEED_DIR)

    wget_process = Popen(
        ["wget", "--quiet", "--output-document", "-", "ftp://foo.com/bar.tar"],
        stdout=PIPE
    )
    tar_process = Popen(["tar", "xf", "-"], stdin=wget_process.stdout)
    stdout, stderr = tar_process.communicate()

显然，最好将脚本修改为使用 HTTP 库而不是 exec-ing wget，但正如我所说，它是一个遗留脚本，所以我需要保持我的更改最小并且完全专注于业务要求，与如何获取XML文件无关。

对我来说显而易见的解决方案是拦截对 subprocess.Popen 的调用并返回我自己的测试 XML。 Intercept method calls in Python 演示了如何使用 setattr 来执行此操作，但我一定遗漏了一些东西：

Python 2.6.6 (r266:84292, Sep 15 2010, 16:22:56) 
[GCC 4.4.5] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import subprocess
>>> object.__getattribute__(subprocess, 'Popen')
<class 'subprocess.Popen'>
>>> attr = object.__getattribute__(subprocess, 'Popen')
>>> hasattr(attr, '__call__')
True
>>> def foo(): print('foo')
... 
>>> foo
<function foo at 0x7f8e3ced3c08>
>>> foo()
foo
>>> setattr(subprocess, '__call__', foo)
>>> getattr(subprocess, '__call__')
<function foo at 0x7f8e3ced3c08>
>>> subprocess.Popen([ r"tail", "-n 1", "x.txt" ], stdout = subprocess.PIPE)
<subprocess.Popen object at 0x7f8e3ced9cd0>
>>> tail: cannot open `x.txt' for reading: No such file or directory

如你所见，真正的 subprocess.Popen 正在被调用，尽管属性设置正确（至少在我大部分未受过训练的情况下）。这只是在交互式 Python 中运行的结果，还是我应该期望将这种代码放入我的测试脚本中得到相同的结果：

class MockProcess:
  def __init__(self, output):
    self.output = output

  def stderr(): pass
  def stdout(): return self.output

  def communicate():
    return stdout, stderr


# Runs script, returning output
#
def run_agent():
  real_popen = getattr(subprocess.Popen, '__call__')
  try:
    setattr(subprocess.Popen, '__call__', lambda *ignored: MockProcess('<foo bar="baz" />')
    )
    return real_popen(['myscript.py'], stdout = subprocess.PIPE).communicate()[0]
  finally:
    setattr(subprocess.Popen, '__call__', real_popen)

Answer 1

我的方法存在几个问题：

我没有意识到 args 在 Python 中很神奇，我也不需要 kwargs。

我正在替换 subprocess.Popen.__call__，而我应该替换 subprocess.Popen 本身。

最重要的是，替换Popen 显然只会影响当前进程，而不是我的代码想要为脚本执行的新进程。新的 run_agent 方法应如下所示：

def run_agent():
  real_popen = getattr(subprocess, 'Popen')
  try:
    setattr(subprocess, 'Popen', lambda *args, **kwargs: MockProcess('<foo bar="baz" />')
    imp.load_module(
      MY_SCRIPT.replace('.py', '').replace('.', '_'),
      file(SCRIPT_DIR),
      MY_SCRIPT,
      ('.py', 'r', imp.PY_SOURCE)
    )
  finally:
    setattr(subprocess.Popen, '__call__', real_popen)

我的互动会话中有错字。它应该是：

Python 2.6.6 (r266:84292, Sep 15 2010, 16:22:56) 
[GCC 4.4.5] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import subprocess
>>> setattr(subprocess, 'Popen', lambda *args, **kwargs: [1,2])
>>> subprocess.Popen([1], stdout=1)
[1, 2]

Answer 2

当然，FlexMock 的 Python 版本是更好的选择！

import subprocess
from cStringIO import StringIO
from flexmock import flexmock

def run_agent():
  flexmock(subprocess).should_receive('Popen').and_return(
      StringIO(''), StringIO('<foo bar="baz" />')
  )

Answer 3

您不是在失败的测试脚本中设置subprocess.__call__ 而不是subprocess.Popen.__call__ 吗？