Intercepting subprocess.Popen call in 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. ;)

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()

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.

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

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:

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

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

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]

Of course, the Python version of FlexMock is an even better choice!

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

Aren't you setting subprocess.__call__ instead of subprocess.Popen.__call__ in your test script that's failing?