2011-03-02 128 views
3

我正在写一个功能测试的遗留Python脚本,这样我可以做一个行修改时没有被恐惧瘫痪。 )拦截subprocess.Popen调用在Python

在考虑中的脚本调用的wget使用subprocess.Popen下载一个XML文件,然后将其解析(1):

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) 

回答

3

几个问题,我的方法:

我没有意识到,参数表是神奇的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] 
1

你是不是在你的测试脚本设置subprocess.__call__,而不是subprocess.Popen.__call__那是失败的?

+0

是的,那是我的一个问题。我还发现了另外两个,这要感谢一位同事。 : -/ – 2011-03-02 11:43:42

+0

嘿,如果它现在有效,你应该是:-),而不是: - /。 ;-)顺便说一句,你可以用你发现的工作回答你自己的问题! – DSM 2011-03-02 11:47:49

+0

是的,我现在:)看到下面我的自我答案。 – 2011-03-02 12:02:14

3

当然,Python版本的FlexMock是更好的选择!

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" />') 
)