"""

Copyright (c) 2007  Gustavo Niemeyer <gustavo@niemeyer.net>

Graceful platform for test doubles in Python (mocks, stubs, fakes, and dummies).

"""

import __builtin__

import tempfile

import unittest

import inspect

import shutil

import types

import sys

import os

import gc

if sys.version_info < (2, 4):

    from sets import Set as set # pragma: nocover

__all__ = ["Mocker", "expect", "IS", "CONTAINS", "IN", "MATCH",

           "ANY", "ARGS", "KWARGS"]

__author__ = "Gustavo Niemeyer <gustavo@niemeyer.net>"

__license__ = "PSF License"

__version__ = "0.10.1"

ERROR_PREFIX = "[Mocker] "

# --------------------------------------------------------------------

# Exceptions

class MatchError(AssertionError):

    """Raised when an unknown expression is seen in playback mode."""

# --------------------------------------------------------------------

# Helper for chained-style calling.

class expect(object):

    """This is a simple helper that allows a different call-style.

    With this class one can comfortably do chaining of calls to the

    mocker object responsible by the object being handled. For instance::

        expect(obj.attr).result(3).count(1, 2)

    Is the same as::

        obj.attr

        mocker.result(3)

        mocker.count(1, 2)

    """

    def __init__(self, mock, attr=None):

        self._mock = mock

        self._attr = attr

    def __getattr__(self, attr):

        return self.__class__(self._mock, attr)

    def __call__(self, *args, **kwargs):

        getattr(self._mock.__mocker__, self._attr)(*args, **kwargs)

        return self

# --------------------------------------------------------------------

# Extensions to Python's unittest.

class MockerTestCase(unittest.TestCase):

    """unittest.TestCase subclass with Mocker support.

    @ivar mocker: The mocker instance.

    This is a convenience only.  Mocker may easily be used with the

    standard C{unittest.TestCase} class if wanted.

    Test methods have a Mocker instance available on C{self.mocker}.

    At the end of each test method, expectations of the mocker will

    be verified, and any requested changes made to the environment

    will be restored.

    In addition to the integration with Mocker, this class provides

    a few additional helper methods.

    """

    expect = expect

    def __init__(self, methodName="runTest"):

        # So here is the trick: we take the real test method, wrap it on

        # a function that do the job we have to do, and insert it in the

        # *instance* dictionary, so that getattr() will return our

        # replacement rather than the class method.

        test_method = getattr(self, methodName, None)

        if test_method is not None:

            def test_method_wrapper():

                try:

                    result = test_method()

                except:

                    raise

                else:

                    if (self.mocker.is_recording() and

                        self.mocker.get_events()):

                        raise RuntimeError("Mocker must be put in replay "

                                           "mode with self.mocker.replay()")

                    if (hasattr(result, "addCallback") and

                        hasattr(result, "addErrback")):

                        def verify(result):

                            self.mocker.verify()

                            return result

                        result.addCallback(verify)

                    else:

                        self.mocker.verify()

                    return result

            # Copy all attributes from the original method..

            for attr in dir(test_method):

                # .. unless they're present in our wrapper already.

                if not hasattr(test_method_wrapper, attr) or attr == "__doc__":

                    setattr(test_method_wrapper, attr,

                            getattr(test_method, attr))

            setattr(self, methodName, test_method_wrapper)

        # We could overload run() normally, but other well-known testing

        # frameworks do it as well, and some of them won't call the super,

        # which might mean that cleanup wouldn't happen.  With that in mind,

        # we make integration easier by using the following trick.

        run_method = self.run

        def run_wrapper(*args, **kwargs):

            try:

                return run_method(*args, **kwargs)

            finally:

                self.__cleanup()

        self.run = run_wrapper

        self.mocker = Mocker()

        self.__cleanup_funcs = []

        self.__cleanup_paths = []

        super(MockerTestCase, self).__init__(methodName)

    def __cleanup(self):

        for path in self.__cleanup_paths:

            if os.path.isfile(path):

                os.unlink(path)

            elif os.path.isdir(path):

                shutil.rmtree(path)

        self.mocker.restore()

        for func, args, kwargs in self.__cleanup_funcs:

            func(*args, **kwargs)

    def addCleanup(self, func, *args, **kwargs):

        self.__cleanup_funcs.append((func, args, kwargs))

    def makeFile(self, content=None, suffix="", prefix="tmp", basename=None,

                 dirname=None, path=None):

        """Create a temporary file and return the path to it.

        @param content: Initial content for the file.

        @param suffix: Suffix to be given to the file's basename.

        @param prefix: Prefix to be given to the file's basename.

        @param basename: Full basename for the file.

        @param dirname: Put file inside this directory.

        The file is removed after the test runs.

        """

        if path is not None:

            self.__cleanup_paths.append(path)

        elif basename is not None:

            if dirname is None:

                dirname = tempfile.mkdtemp()

                self.__cleanup_paths.append(dirname)

            path = os.path.join(dirname, basename)

        else:

            fd, path = tempfile.mkstemp(suffix, prefix, dirname)

            self.__cleanup_paths.append(path)

            os.close(fd)

            if content is None:

                os.unlink(path)

        if content is not None:

            file = open(path, "w")

            file.write(content)

            file.close()

        return path

    def makeDir(self, suffix="", prefix="tmp", dirname=None, path=None):

        """Create a temporary directory and return the path to it.

        @param suffix: Suffix to be given to the file's basename.

        @param prefix: Prefix to be given to the file's basename.

        @param dirname: Put directory inside this parent directory.

        The directory is removed after the test runs.

        """

        if path is not None:

            os.makedirs(path)

        else:

            path = tempfile.mkdtemp(suffix, prefix, dirname)

        self.__cleanup_paths.append(path)

        return path

    def failUnlessIs(self, first, second, msg=None):

        """Assert that C{first} is the same object as C{second}."""

        if first is not second:

            raise self.failureException(msg or "%r is not %r" % (first, second))

    def failIfIs(self, first, second, msg=None):

        """Assert that C{first} is not the same object as C{second}."""

        if first is second:

            raise self.failureException(msg or "%r is %r" % (first, second))

    def failUnlessIn(self, first, second, msg=None):

        """Assert that C{first} is contained in C{second}."""

        if first not in second:

            raise self.failureException(msg or "%r not in %r" % (first, second))

    def failUnlessStartsWith(self, first, second, msg=None):

        """Assert that C{first} starts with C{second}."""

        if first[:len(second)] != second:

            raise self.failureException(msg or "%r doesn't start with %r" %

                                               (first, second))

    def failIfStartsWith(self, first, second, msg=None):

        """Assert that C{first} doesn't start with C{second}."""

        if first[:len(second)] == second:

            raise self.failureException(msg or "%r starts with %r" %

                                               (first, second))

    def failUnlessEndsWith(self, first, second, msg=None):

        """Assert that C{first} starts with C{second}."""

        if first[len(first)-len(second):] != second:

            raise self.failureException(msg or "%r doesn't end with %r" %

                                               (first, second))

    def failIfEndsWith(self, first, second, msg=None):

        """Assert that C{first} doesn't start with C{second}."""

        if first[len(first)-len(second):] == second:

            raise self.failureException(msg or "%r ends with %r" %

                                               (first, second))

    def failIfIn(self, first, second, msg=None):

        """Assert that C{first} is not contained in C{second}."""

        if first in second:

            raise self.failureException(msg or "%r in %r" % (first, second))

    def failUnlessApproximates(self, first, second, tolerance, msg=None):

        """Assert that C{first} is near C{second} by at most C{tolerance}."""

        if abs(first - second) > tolerance:

            raise self.failureException(msg or "abs(%r - %r) > %r" %

                                        (first, second, tolerance))

    def failIfApproximates(self, first, second, tolerance, msg=None):

        """Assert that C{first} is far from C{second} by at least C{tolerance}.

        """

        if abs(first - second) <= tolerance:

            raise self.failureException(msg or "abs(%r - %r) <= %r" %

                                        (first, second, tolerance))

    def failUnlessMethodsMatch(self, first, second):

        """Assert that public methods in C{first} are present in C{second}.

        This method asserts that all public methods found in C{first} are also

        present in C{second} and accept the same arguments.  C{first} may

        have its own private methods, though, and may not have all methods

        found in C{second}.  Note that if a private method in C{first} matches

        the name of one in C{second}, their specification is still compared.

        This is useful to verify if a fake or stub class have the same API as

        the real class being simulated.

        """

        first_methods = dict(inspect.getmembers(first, inspect.ismethod))

        second_methods = dict(inspect.getmembers(second, inspect.ismethod))

        for name, first_method in first_methods.items():

            first_argspec = inspect.getargspec(first_method)

            first_formatted = inspect.formatargspec(*first_argspec)

            second_method = second_methods.get(name)

            if second_method is None:

                if name[:1] == "_":

                    continue # First may have its own private methods.

                raise self.failureException("%s.%s%s not present in %s" %

                    (first.__name__, name, first_formatted, second.__name__))

            second_argspec = inspect.getargspec(second_method)

            if first_argspec != second_argspec:

                second_formatted = inspect.formatargspec(*second_argspec)

                raise self.failureException("%s.%s%s != %s.%s%s" %

                    (first.__name__, name, first_formatted,

                     second.__name__, name, second_formatted))

    assertIs = failUnlessIs

    assertIsNot = failIfIs

    assertIn = failUnlessIn

    assertNotIn = failIfIn

    assertStartsWith = failUnlessStartsWith

    assertNotStartsWith = failIfStartsWith

    assertEndsWith = failUnlessEndsWith

    assertNotEndsWith = failIfEndsWith

    assertApproximates = failUnlessApproximates

    assertNotApproximates = failIfApproximates

    assertMethodsMatch = failUnlessMethodsMatch

    # The following are missing in Python < 2.4.

    assertTrue = unittest.TestCase.failUnless

    assertFalse = unittest.TestCase.failIf

    # The following is provided for compatibility with Twisted's trial.

    assertIdentical = assertIs

    assertNotIdentical = assertIsNot

    failUnlessIdentical = failUnlessIs

    failIfIdentical = failIfIs

# --------------------------------------------------------------------

# Mocker.

class classinstancemethod(object):

    def __init__(self, method):

        self.method = method

    def __get__(self, obj, cls=None):

        def bound_method(*args, **kwargs):

            return self.method(cls, obj, *args, **kwargs)

        return bound_method

class MockerBase(object):

    """Controller of mock objects.

    A mocker instance is used to command recording and replay of

    expectations on any number of mock objects.

    Expectations should be expressed for the mock object while in

    record mode (the initial one) by using the mock object itself,

    and using the mocker (and/or C{expect()} as a helper) to define

    additional behavior for each event.  For instance::

        mock = mocker.mock()

        mock.hello()

        mocker.result("Hi!")

        mocker.replay()

        assert mock.hello() == "Hi!"

        mock.restore()

        mock.verify()

    In this short excerpt a mock object is being created, then an

    expectation of a call to the C{hello()} method was recorded, and

    when called the method should return the value C{10}.  Then, the

    mocker is put in replay mode, and the expectation is satisfied by

    calling the C{hello()} method, which indeed returns 10.  Finally,

    a call to the L{restore()} method is performed to undo any needed

    changes made in the environment, and the L{verify()} method is

    called to ensure that all defined expectations were met.

    The same logic can be expressed more elegantly using the

    C{with mocker:} statement, as follows::

        mock = mocker.mock()

        mock.hello()

        mocker.result("Hi!")

        with mocker:

            assert mock.hello() == "Hi!"

    Also, the MockerTestCase class, which integrates the mocker on

    a unittest.TestCase subclass, may be used to reduce the overhead

    of controlling the mocker.  A test could be written as follows::

        class SampleTest(MockerTestCase):

            def test_hello(self):

                mock = self.mocker.mock()

                mock.hello()

                self.mocker.result("Hi!")

                self.mocker.replay()

                self.assertEquals(mock.hello(), "Hi!")

    """

    _recorders = []

    # For convenience only.

    on = expect

    class __metaclass__(type):

        def __init__(self, name, bases, dict):

            # Make independent lists on each subclass, inheriting from parent.

            self._recorders = list(getattr(self, "_recorders", ()))

    def __init__(self):

        self._recorders = self._recorders[:]

        self._events = []

        self._recording = True

        self._ordering = False

        self._last_orderer = None

    def is_recording(self):

        """Return True if in recording mode, False if in replay mode.

        Recording is the initial state.

        """

        return self._recording

    def replay(self):

        """Change to replay mode, where recorded events are reproduced.

        If already in replay mode, the mocker will be restored, with all

        expectations reset, and then put again in replay mode.

        An alternative and more comfortable way to replay changes is

        using the 'with' statement, as follows::

            mocker = Mocker()

            <record events>

            with mocker:

                <reproduce events>

        The 'with' statement will automatically put mocker in replay

        mode, and will also verify if all events were correctly reproduced

        at the end (using L{verify()}), and also restore any changes done

        in the environment (with L{restore()}).

        Also check the MockerTestCase class, which integrates the

        unittest.TestCase class with mocker.

        """

        if not self._recording:

            for event in self._events:

                event.restore()

        else:

            self._recording = False

        for event in self._events:

            event.replay()

    def restore(self):

        """Restore changes in the environment, and return to recording mode.

        This should always be called after the test is complete (succeeding

        or not).  There are ways to call this method automatically on

        completion (e.g. using a C{with mocker:} statement, or using the

        L{MockerTestCase} class.

        """

        if not self._recording:

            self._recording = True

            for event in self._events:

                event.restore()

    def reset(self):

        """Reset the mocker state.

        This will restore environment changes, if currently in replay

        mode, and then remove all events previously recorded.

        """

        if not self._recording:

            self.restore()

        self.unorder()

        del self._events[:]

    def get_events(self):

        """Return all recorded events."""

        return self._events[:]

    def add_event(self, event):

        """Add an event.

        This method is used internally by the implementation, and

        shouldn't be needed on normal mocker usage.

        """

        self._events.append(event)

        if self._ordering:

            orderer = event.add_task(Orderer(event.path))

            if self._last_orderer:

                orderer.add_dependency(self._last_orderer)

            self._last_orderer = orderer

        return event

    def verify(self):

        """Check if all expectations were met, and raise AssertionError if not.

        The exception message will include a nice description of which

        expectations were not met, and why.

        """

        errors = []

        for event in self._events:

            try:

                event.verify()

            except AssertionError, e:

                error = str(e)

                if not error:

                    raise RuntimeError("Empty error message from %r"

                                       % event)

                errors.append(error)

        if errors:

            message = [ERROR_PREFIX + "Unmet expectations:", ""]

            for error in errors:

                lines = error.splitlines()

                message.append("=> " + lines.pop(0))

                message.extend([" " + line for line in lines])

                message.append("")

            raise AssertionError(os.linesep.join(message))

    def mock(self, spec_and_type=None, spec=None, type=None,

             name=None, count=True):

        """Return a new mock object.

        @param spec_and_type: Handy positional argument which sets both

                     spec and type.

        @param spec: Method calls will be checked for correctness against

                     the given class.

        @param type: If set, the Mock's __class__ attribute will return

                     the given type.  This will make C{isinstance()} calls

                     on the object work.

        @param name: Name for the mock object, used in the representation of

                     expressions.  The name is rarely needed, as it's usually

                     guessed correctly from the variable name used.

        @param count: If set to false, expressions may be executed any number

                     of times, unless an expectation is explicitly set using

                     the L{count()} method.  By default, expressions are

                     expected once.

        """

        if spec_and_type is not None:

            spec = type = spec_and_type

        return Mock(self, spec=spec, type=type, name=name, count=count)

    def proxy(self, object, spec=True, type=True, name=None, count=True,

              passthrough=True):