Python mock

queue.py

from boto.sqs import connect_to_region
from boto.sqs.message import Message


class Queue(object):

    def __init__(self, name):
        self._queue = self._get_queue(name)

    def _get_queue(self, name):
        sqs_connection = connect_to_region('eu-west-1')
        return sqs_connection.get_queue(name)

    @property
    def is_empty(self):
        return self._queue.count() == 0

    def push(self, *messages):
        for message in messages:
            envelope = Message()
            envelope.set_body(message)
            self._queue.write(envelope)

    def pop(self):
        if self.is_empty:
            return None
        message = self._queue.read()
        return message.get_body()

READEME.md

link: http://www.alexandrejoseph.com/blog/2015-08-21-python-mock-example.html

Directory structure:

pyqueue
├── README.md
├── pyqueue
│   ├── __init__.py
│   ├── queue.py
├── req.txt
└── tests
    ├── __init__.py
    └── test_queue.py

Test:

$ cd pyqueue
$ pip install -r req.txt
$ py.test tests

req.txt

boto==2.36.0
mock==1.0.1
py==1.4.26
pytest==2.6.4

tests.__init__.py

import os
import sys

TEST_DIR = os.path.dirname(os.path.realpath(__file__))
sys.path.insert(0, TEST_DIR + '/../')

tests.test_queue.py

import unittest
from boto.sqs.message import Message
from mock import patch, Mock, PropertyMock

from pyqueue.queue import Queue


class QueueTestCase(unittest.TestCase):

    @patch.object(Queue, '_get_queue')
    # This is the same thing
    # @patch('pyqueue.queue.Queue._get_queue')
    def test_queue_initialization(self, get_queue_mock):
        """
        Can use either `patch` or `patch.object` as Queue object is imported.
        * To check that a method called only once:
            `assert_called_once_with`
        * To check the last call: `assert_called_with`
        * To check that a particular call is done among other:
            `assert_any_call`
        """
        queue = Queue('foo')
        get_queue_mock.assert_called_once_with('foo')
        assert queue._queue == get_queue_mock.return_value

    # Mock the imported module
    @patch('pyqueue.queue.connect_to_region')
    def test_get_queue(self, connect_to_region_mock):
        """
        Two way to know if a method (i.e. a mock) have been called:
         * my_mock.called: returns a boolean regardless the number
           of call
         * my_mock.call_count: returns the actual number of call
        """
        sqs_connection_mock = Mock()
        sqs_connection_mock.get_queue.return_value = 'bar'

        connect_to_region_mock.return_value = sqs_connection_mock

        queue = Queue('foo')
        assert connect_to_region_mock.called
        assert queue._queue == 'bar'
        sqs_connection_mock.get_queue.assert_called_once_with('foo')

    @patch.object(Queue, '_get_queue')
    def test_is_empty_should_return_false(self, get_queue_mock):
        queue_mock = Mock()
        queue_mock.count.return_value = 10

        get_queue_mock.return_value = queue_mock

        queue = Queue('foo')
        assert queue.is_empty is False

    @patch('pyqueue.queue.Message', spec=Message)
    @patch.object(Queue, '_get_queue')
    def test_push_multiple_messages(self, get_queue_mock, message_mock):
        """
        Notice the decoration and parameter order: the first
        parameter is the closest to the function name (thing how
        decorator are called).

        Same as previously we start by mocking the queue
        initialization.

        Then we create a Message container (envelope) that match the
        specification of a real Message object with spec=Message.
        It means that if the object signature change or if we have a
        typo in the code and the test it will raise (Mock object has
        no attribute ...).

        Finally we check that every message is well written in the
        queue.
        """
        queue_mock = Mock()
        get_queue_mock.return_value = queue_mock

        envelope_mock = Mock(spec=Message)
        message_mock.return_value = envelope_mock

        queue = Queue('foo')
        queue.push('foo', 'bar')

        assert queue_mock.write.call_count == 2
        envelope_mock.set_body.assert_any_call('foo')

    @patch.object(Queue, '_get_queue')
    @patch.object(Queue, 'is_empty', new_callable=PropertyMock)
    def test_pop_empty_queue_should_return_none(self,
                                                is_empty_mock,
                                                get_queue_mock):
        """
        `new_callable=PropertyMock`
        """
        queue_mock = Mock()
        get_queue_mock.return_value = queue_mock
        is_empty_mock.return_value = True

        queue = Queue('foo')
        assert queue.pop() is None
        assert queue_mock.read.called is False
        is_empty_mock.assert_called_once_with()