Da

✏️ Editing: test_popen_docs.py

# NB: This file is used in the documentation, if you make changes, ensure
#     you update the line numbers in popen.txt!

from subprocess import Popen, PIPE

def my_func():
    process = Popen(['svn', 'ls', '-R', 'foo'], stdout=PIPE, stderr=PIPE)
    out, err = process.communicate()
    if process.returncode:
        raise RuntimeError('something bad happened')
    return out

dotted_path = 'testfixtures.tests.test_popen_docs.Popen'

from unittest import TestCase

from testfixtures.mock import call
from testfixtures import Replacer, ShouldRaise, compare, SequenceComparison
from testfixtures.popen import MockPopen, PopenBehaviour

class TestMyFunc(TestCase):

def setUp(self):
        self.Popen = MockPopen()
        self.r = Replacer()
        self.r.replace(dotted_path, self.Popen)
        self.addCleanup(self.r.restore)

def test_example(self):
        # set up
        self.Popen.set_command('svn ls -R foo', stdout=b'o', stderr=b'e')

# testing of results
        compare(my_func(), b'o')

# testing calls were in the right order and with the correct parameters:
        process = call.Popen(['svn', 'ls', '-R', 'foo'], stderr=PIPE, stdout=PIPE)
        compare(Popen.all_calls, expected=[
            process,
            process.communicate()
        ])

def test_example_bad_returncode(self):
        # set up
        Popen.set_command('svn ls -R foo', stdout=b'o', stderr=b'e',
                          returncode=1)

# testing of error
        with ShouldRaise(RuntimeError('something bad happened')):
            my_func()

def test_communicate_with_input(self):
        # setup
        Popen = MockPopen()
        Popen.set_command('a command')
        # usage
        process = Popen('a command', stdout=PIPE, stderr=PIPE, shell=True)
        out, err = process.communicate('foo')
        # test call list
        compare(Popen.all_calls, expected=[
                process.root_call,
                process.root_call.communicate('foo'),
        ])

def test_read_from_stdout_and_stderr(self):
        # setup
        Popen = MockPopen()
        Popen.set_command('a command', stdout=b'foo', stderr=b'bar')
        # usage
        process = Popen('a command', stdout=PIPE, stderr=PIPE, shell=True)
        compare(process.stdout.read(), expected=b'foo')
        compare(process.stderr.read(), expected=b'bar')

def test_write_to_stdin(self):
        # setup
        Popen = MockPopen()
        Popen.set_command('a command')
        # usage
        process = Popen('a command', stdin=PIPE, shell=True)
        process.stdin.write('some text')
        process.stdin.close()
        # test call list
        compare(Popen.all_calls, expected=[
            process.root_call,
            process.root_call.stdin.write('some text'),
            process.root_call.stdin.close(),
        ])

def test_wait_and_return_code(self):
        # setup
        Popen = MockPopen()
        Popen.set_command('a command', returncode=3)
        # usage
        process = Popen('a command')
        compare(process.returncode, expected=None)
        # result checking
        compare(process.wait(), expected=3)
        compare(process.returncode, expected=3)
        # test call list
        compare(Popen.all_calls, expected=[
            call.Popen('a command'),
            call.Popen('a command').wait(),
        ])

def test_send_signal(self):
        # setup
        Popen = MockPopen()
        Popen.set_command('a command')
        # usage
        process = Popen('a command', stdout=PIPE, stderr=PIPE, shell=True)
        process.send_signal(0)
        # result checking
        compare(Popen.all_calls, expected=[
            process.root_call,
            process.root_call.send_signal(0),
        ])

def test_poll_until_result(self):
        # setup
        Popen = MockPopen()
        Popen.set_command('a command', returncode=3, poll_count=2)
        # example usage
        process = Popen('a command')
        while process.poll() is None:
            # you'd probably have a sleep here, or go off and
            # do some other work.
            pass
        # result checking
        compare(process.returncode, expected=3)
        compare(Popen.all_calls, expected=[
            process.root_call,
            process.root_call.poll(),
            process.root_call.poll(),
            process.root_call.poll(),
        ])

def test_default_behaviour(self):
        # set up
        self.Popen.set_default(stdout=b'o', stderr=b'e')

# testing of results
        compare(my_func(), b'o')

# testing calls were in the right order and with the correct parameters:
        root_call = call.Popen(['svn', 'ls', '-R', 'foo'],
                               stderr=PIPE, stdout=PIPE)
        compare(Popen.all_calls, expected=[
            root_call,
            root_call.communicate()
        ])

def test_multiple_responses(self):
        # set up
        behaviours = [
            PopenBehaviour(stderr=b'e', returncode=1),
            PopenBehaviour(stdout=b'o'),
        ]

def behaviour(command, stdin):
            return behaviours.pop(0)

self.Popen.set_command('svn ls -R foo', behaviour=behaviour)

# testing of error:
        with ShouldRaise(RuntimeError('something bad happened')):
            my_func()
        # testing of second call:
        compare(my_func(), b'o')

def test_count_down(self):
        # set up
        self.Popen.set_command('svn ls -R foo', behaviour=CustomBehaviour())
        # testing of error:
        with ShouldRaise(RuntimeError('something bad happened')):
            my_func()
        # testing of second call:
        compare(my_func(), b'o')

def test_multiple_processes(self):
        # set up
        self.Popen.set_command('process --batch=0', stdout=b'42')
        self.Popen.set_command('process --batch=1', stdout=b'13')

# testing of results
        compare(process_in_batches(2), expected=55)

# testing of process management:
        p1 = call.Popen('process --batch=0', shell=True, stderr=PIPE, stdout=PIPE)
        p2 = call.Popen('process --batch=1', shell=True, stderr=PIPE, stdout=PIPE)
        compare(Popen.all_calls, expected=[
            p1,
            p2,
            p1.communicate(),
            p2.communicate(),
        ])

def test_multiple_processes_unordered(self):
        # set up
        self.Popen.set_command('process --batch=0', stdout=b'42')
        self.Popen.set_command('process --batch=1', stdout=b'13')

# testing of results
        compare(process_in_batches(2), expected=55)

# testing of process management:
        p1 = call.Popen('process --batch=0', shell=True, stderr=PIPE, stdout=PIPE)
        p2 = call.Popen('process --batch=1', shell=True, stderr=PIPE, stdout=PIPE)
        compare(Popen.all_calls, expected=SequenceComparison(
            p2,
            p2.communicate(),
            p1,
            p1.communicate(),
            ordered=False
        ))

class CustomBehaviour(object):

def __init__(self, fail_count=1):
        self.fail_count = fail_count

def __call__(self, command, stdin):
        while self.fail_count > 0:
            self.fail_count -= 1
            return PopenBehaviour(stderr=b'e', returncode=1)
        return PopenBehaviour(stdout=b'o')

def process_in_batches(n):
    processes = []
    for i in range(n):
        processes.append(Popen('process --batch='+str(i),
                               stdout=PIPE, stderr=PIPE, shell=True))
    total = 0
    for process in processes:
        out, err = process.communicate()
        total += int(out)
    return total

📁 Sid Gifari File Manager