#!/usr/bin/python2.4

#

# Copyright 2008 Google Inc.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

"""Mox, an object-mocking framework for Python.

Mox works in the record-replay-verify paradigm.  When you first create

a mock object, it is in record mode.  You then programmatically set

the expected behavior of the mock object (what methods are to be

called on it, with what parameters, what they should return, and in

what order).

Once you have set up the expected mock behavior, you put it in replay

mode.  Now the mock responds to method calls just as you told it to.

If an unexpected method (or an expected method with unexpected

parameters) is called, then an exception will be raised.

Once you are done interacting with the mock, you need to verify that

all the expected interactions occured.  (Maybe your code exited

prematurely without calling some cleanup method!)  The verify phase

ensures that every expected method was called; otherwise, an exception

will be raised.

Suggested usage / workflow:

  # Create Mox factory

  my_mox = Mox()

  # Create a mock data access object

  mock_dao = my_mox.CreateMock(DAOClass)

  # Set up expected behavior

  mock_dao.RetrievePersonWithIdentifier('1').AndReturn(person)

  mock_dao.DeletePerson(person)

  # Put mocks in replay mode

  my_mox.ReplayAll()

  # Inject mock object and run test

  controller.SetDao(mock_dao)

  controller.DeletePersonById('1')

  # Verify all methods were called as expected

  my_mox.VerifyAll()

"""

from collections import deque

import inspect

import re

import types

import unittest

import stubout

class Error(AssertionError):

  """Base exception for this module."""

  pass

class ExpectedMethodCallsError(Error):

  """Raised when Verify() is called before all expected methods have been called

  """

  def __init__(self, expected_methods):

    """Init exception.

    Args:

      # expected_methods: A sequence of MockMethod objects that should have been

      #   called.

      expected_methods: [MockMethod]

    Raises:

      ValueError: if expected_methods contains no methods.

    """

    if not expected_methods:

      raise ValueError("There must be at least one expected method")

    Error.__init__(self)

    self._expected_methods = expected_methods

  def __str__(self):

    calls = "\n".join(["%3d.  %s" % (i, m)

                       for i, m in enumerate(self._expected_methods)])

    return "Verify: Expected methods never called:\n%s" % (calls,)

class UnexpectedMethodCallError(Error):

  """Raised when an unexpected method is called.

  This can occur if a method is called with incorrect parameters, or out of the

  specified order.

  """

  def __init__(self, unexpected_method, expected):

    """Init exception.

    Args:

      # unexpected_method: MockMethod that was called but was not at the head of

      #   the expected_method queue.

      # expected: MockMethod or UnorderedGroup the method should have

      #   been in.

      unexpected_method: MockMethod

      expected: MockMethod or UnorderedGroup

    """

    Error.__init__(self)

    self._unexpected_method = unexpected_method

    self._expected = expected

  def __str__(self):

    return "Unexpected method call: %s.  Expecting: %s" % \

      (self._unexpected_method, self._expected)

class UnknownMethodCallError(Error):

  """Raised if an unknown method is requested of the mock object."""

  def __init__(self, unknown_method_name):

    """Init exception.

    Args:

      # unknown_method_name: Method call that is not part of the mocked class's

      #   public interface.

      unknown_method_name: str

    """

    Error.__init__(self)

    self._unknown_method_name = unknown_method_name

  def __str__(self):

    return "Method called is not a member of the object: %s" % \

      self._unknown_method_name

class Mox(object):

  """Mox: a factory for creating mock objects."""

  # A list of types that should be stubbed out with MockObjects (as

  # opposed to MockAnythings).

  _USE_MOCK_OBJECT = [types.ClassType, types.InstanceType, types.ModuleType,

                      types.ObjectType, types.TypeType]

  def __init__(self):

    """Initialize a new Mox."""

    self._mock_objects = []

    self.stubs = stubout.StubOutForTesting()

  def CreateMock(self, class_to_mock):

    """Create a new mock object.

    Args:

      # class_to_mock: the class to be mocked

      class_to_mock: class

    Returns:

      MockObject that can be used as the class_to_mock would be.

    """

    new_mock = MockObject(class_to_mock)

    self._mock_objects.append(new_mock)

    return new_mock

  def CreateMockAnything(self):

    """Create a mock that will accept any method calls.

    This does not enforce an interface.

    """

    new_mock = MockAnything()

    self._mock_objects.append(new_mock)

    return new_mock

  def ReplayAll(self):

    """Set all mock objects to replay mode."""

    for mock_obj in self._mock_objects:

      mock_obj._Replay()

  def VerifyAll(self):

    """Call verify on all mock objects created."""

    for mock_obj in self._mock_objects:

      mock_obj._Verify()

  def ResetAll(self):

    """Call reset on all mock objects.  This does not unset stubs."""

    for mock_obj in self._mock_objects:

      mock_obj._Reset()

  def StubOutWithMock(self, obj, attr_name, use_mock_anything=False):

    """Replace a method, attribute, etc. with a Mock.

    This will replace a class or module with a MockObject, and everything else

    (method, function, etc) with a MockAnything.  This can be overridden to

    always use a MockAnything by setting use_mock_anything to True.

    Args:

      obj: A Python object (class, module, instance, callable).

      attr_name: str.  The name of the attribute to replace with a mock.

      use_mock_anything: bool. True if a MockAnything should be used regardless

        of the type of attribute.

    """

    attr_to_replace = getattr(obj, attr_name)

    if type(attr_to_replace) in self._USE_MOCK_OBJECT and not use_mock_anything:

      stub = self.CreateMock(attr_to_replace)

    else:

      stub = self.CreateMockAnything()

    self.stubs.Set(obj, attr_name, stub)

  def UnsetStubs(self):

    """Restore stubs to their original state."""

    self.stubs.UnsetAll()

def Replay(*args):

  """Put mocks into Replay mode.

  Args:

    # args is any number of mocks to put into replay mode.

  """

  for mock in args:

    mock._Replay()

def Verify(*args):

  """Verify mocks.

  Args:

    # args is any number of mocks to be verified.

  """

  for mock in args:

    mock._Verify()

def Reset(*args):

  """Reset mocks.

  Args:

    # args is any number of mocks to be reset.

  """

  for mock in args:

    mock._Reset()

class MockAnything:

  """A mock that can be used to mock anything.

  This is helpful for mocking classes that do not provide a public interface.

  """

  def __init__(self):

    """ """

    self._Reset()

  def __getattr__(self, method_name):

    """Intercept method calls on this object.

     A new MockMethod is returned that is aware of the MockAnything's

     state (record or replay).  The call will be recorded or replayed

     by the MockMethod's __call__.

    Args:

      # method name: the name of the method being called.

      method_name: str

    Returns:

      A new MockMethod aware of MockAnything's state (record or replay).

    """

    return self._CreateMockMethod(method_name)

  def _CreateMockMethod(self, method_name, method_to_mock=None):

    """Create a new mock method call and return it.

    Args:

      # method_name: the name of the method being called.

      # method_to_mock: The actual method being mocked, used for introspection.

      method_name: str

      method_to_mock: a method object

    Returns:

      A new MockMethod aware of MockAnything's state (record or replay).

    """

    return MockMethod(method_name, self._expected_calls_queue,

                      self._replay_mode, method_to_mock=method_to_mock)

  def __nonzero__(self):

    """Return 1 for nonzero so the mock can be used as a conditional."""

    return 1

  def __eq__(self, rhs):

    """Provide custom logic to compare objects."""

    return (isinstance(rhs, MockAnything) and

            self._replay_mode == rhs._replay_mode and

            self._expected_calls_queue == rhs._expected_calls_queue)

  def __ne__(self, rhs):

    """Provide custom logic to compare objects."""

    return not self == rhs

  def _Replay(self):

    """Start replaying expected method calls."""

    self._replay_mode = True

  def _Verify(self):

    """Verify that all of the expected calls have been made.

    Raises:

      ExpectedMethodCallsError: if there are still more method calls in the

        expected queue.

    """

    # If the list of expected calls is not empty, raise an exception

    if self._expected_calls_queue:

      # The last MultipleTimesGroup is not popped from the queue.

      if (len(self._expected_calls_queue) == 1 and

          isinstance(self._expected_calls_queue[0], MultipleTimesGroup) and

          self._expected_calls_queue[0].IsSatisfied()):

        pass

      else:

        raise ExpectedMethodCallsError(self._expected_calls_queue)

  def _Reset(self):

    """Reset the state of this mock to record mode with an empty queue."""

    # Maintain a list of method calls we are expecting

    self._expected_calls_queue = deque()

    # Make sure we are in setup mode, not replay mode

    self._replay_mode = False

class MockObject(MockAnything, object):

  """A mock object that simulates the public/protected interface of a class."""

  def __init__(self, class_to_mock):

    """Initialize a mock object.

    This determines the methods and properties of the class and stores them.

    Args:

      # class_to_mock: class to be mocked

      class_to_mock: class

    """

    # This is used to hack around the mixin/inheritance of MockAnything, which

    # is not a proper object (it can be anything. :-)

    MockAnything.__dict__['__init__'](self)

    # Get a list of all the public and special methods we should mock.

    self._known_methods = set()

    self._known_vars = set()

    self._class_to_mock = class_to_mock

    for method in dir(class_to_mock):

      if callable(getattr(class_to_mock, method)):

        self._known_methods.add(method)

      else:

        self._known_vars.add(method)

  def __getattr__(self, name):

    """Intercept attribute request on this object.

    If the attribute is a public class variable, it will be returned and not

    recorded as a call.

    If the attribute is not a variable, it is handled like a method

    call. The method name is checked against the set of mockable

    methods, and a new MockMethod is returned that is aware of the

    MockObject's state (record or replay).  The call will be recorded

    or replayed by the MockMethod's __call__.

    Args:

      # name: the name of the attribute being requested.

      name: str

    Returns:

      Either a class variable or a new MockMethod that is aware of the state

      of the mock (record or replay).

    Raises:

      UnknownMethodCallError if the MockObject does not mock the requested

          method.

    """

    if name in self._known_vars:

      return getattr(self._class_to_mock, name)

    if name in self._known_methods:

      return self._CreateMockMethod(

          name,

          method_to_mock=getattr(self._class_to_mock, name))

    raise UnknownMethodCallError(name)

  def __eq__(self, rhs):

    """Provide custom logic to compare objects."""

    return (isinstance(rhs, MockObject) and

            self._class_to_mock == rhs._class_to_mock and

            self._replay_mode == rhs._replay_mode and

            self._expected_calls_queue == rhs._expected_calls_queue)

  def __setitem__(self, key, value):

    """Provide custom logic for mocking classes that support item assignment.

    Args:

      key: Key to set the value for.

      value: Value to set.

    Returns:

      Expected return value in replay mode.  A MockMethod object for the

      __setitem__ method that has already been called if not in replay mode.

    Raises:

      TypeError if the underlying class does not support item assignment.

      UnexpectedMethodCallError if the object does not expect the call to

        __setitem__.

    """

    setitem = self._class_to_mock.__dict__.get('__setitem__', None)

    # Verify the class supports item assignment.

    if setitem is None:

      raise TypeError('object does not support item assignment')

    # If we are in replay mode then simply call the mock __setitem__ method.

    if self._replay_mode:

      return MockMethod('__setitem__', self._expected_calls_queue,

                        self._replay_mode)(key, value)

    # Otherwise, create a mock method __setitem__.

    return self._CreateMockMethod('__setitem__')(key, value)

  def __getitem__(self, key):

    """Provide custom logic for mocking classes that are subscriptable.

    Args:

      key: Key to return the value for.

    Returns:

      Expected return value in replay mode.  A MockMethod object for the

      __getitem__ method that has already been called if not in replay mode.

    Raises:

      TypeError if the underlying class is not subscriptable.

      UnexpectedMethodCallError if the object does not expect the call to

        __setitem__.

    """

    getitem = self._class_to_mock.__dict__.get('__getitem__', None)

    # Verify the class supports item assignment.

    if getitem is None:

      raise TypeError('unsubscriptable object')

    # If we are in replay mode then simply call the mock __getitem__ method.

    if self._replay_mode:

      return MockMethod('__getitem__', self._expected_calls_queue,

                        self._replay_mode)(key)

    # Otherwise, create a mock method __getitem__.

    return self._CreateMockMethod('__getitem__')(key)

  def __contains__(self, key):

    """Provide custom logic for mocking classes that contain items.

    Args:

      key: Key to look in container for.

    Returns:

      Expected return value in replay mode.  A MockMethod object for the

      __contains__ method that has already been called if not in replay mode.

    Raises:

      TypeError if the underlying class does not implement __contains__

      UnexpectedMethodCaller if the object does not expect the call to

      __contains__.

    """

    contains = self._class_to_mock.__dict__.get('__contains__', None)

    if contains is None:

      raise TypeError('unsubscriptable object')

    if self._replay_mode:

      return MockMethod('__contains__', self._expected_calls_queue,

                        self._replay_mode)(key)

    return self._CreateMockMethod('__contains__')(key)

  def __call__(self, *params, **named_params):

    """Provide custom logic for mocking classes that are callable."""

    # Verify the class we are mocking is callable

    callable = self._class_to_mock.__dict__.get('__call__', None)

    if callable is None:

      raise TypeError('Not callable')

    # Because the call is happening directly on this object instead of a method,

    # the call on the mock method is made right here

    mock_method = self._CreateMockMethod('__call__')