拦截Python中的subprocess.Popen调用 [英] Intercepting subprocess.Popen call in Python

问题描述

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

I'm writing a functional test for a legacy Python script so that I can make a one-line change to it without being paralysed by fear. ;)

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

The script in question invokes wget(1) using subprocess.Popen to download an XML file which is then parsed:

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库而不是执行wget更为可取，但是正如我所说的，这是一个遗留脚本，因此我需要将更改保持在最低限度，并完全专注于业务要求，这与XML文件的获取方式无关.

Obviously, it would be preferable to modify the script to use an HTTP library instead of exec-ing wget, but as I said, it is a legacy script, so I need to keep my change minimal and absolutely focused on the business requirement, which has nothing to do with how the XML file is obtained.

对我来说，显而易见的解决方案是拦截对 subprocess.Popen 的调用并返回我自己的测试XML. Python中的拦截方法调用演示了如何使用 setattr 为此，但是我必须缺少一些东西:

The obvious solution to me is to intercept the call to subprocess.Popen and return my own test XML. Intercept method calls in Python demonstrates how to use setattr to do this, but I must be missing something:

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中运行此代码的结果，还是我期望将这种代码放入测试脚本中得到相同的结果?

As you can see, the real subprocess.Popen is being called, despite the attribute being set correctly (at least to my largely untrained eye). Is this just a result of running this in interactive Python, or should I expect the same result from dropping this sort of code into my test script:

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)

推荐答案

我的方法存在几个问题:

Several problems with my approach:

我没有意识到args在Python中是神奇的，也没有意识到我也需要kwargs.

I didn't realise that args is magical in Python, nor that I needed kwargs as well.

当我应该替换subprocess.Popen本身时，我正在替换subprocess.Popen.__call__.

I was replacing subprocess.Popen.__call__, when I should be replacing subprocess.Popen itself.

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

Most importantly, replacing Popen is obviously only going to affect the current process, not the new one that my code wanted to exec for the script. The new run_agent method should look like this:

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)

我在互动环节打错了字.它应显示为:

I had a typo in my interactive session. It should read:

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]

这篇关于拦截Python中的subprocess.Popen调用的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！