pytest_mock.plugin

  1import asyncio
  2import builtins
  3import functools
  4import inspect
  5import unittest.mock
  6import warnings
  7from dataclasses import dataclass
  8from dataclasses import field
  9from typing import Any
 10from typing import Callable
 11from typing import Dict
 12from typing import Generator
 13from typing import Iterable
 14from typing import Iterator
 15from typing import List
 16from typing import Mapping
 17from typing import Optional
 18from typing import Tuple
 19from typing import Type
 20from typing import TypeVar
 21from typing import Union
 22from typing import cast
 23from typing import overload
 24
 25import pytest
 26
 27from ._util import get_mock_module
 28from ._util import parse_ini_boolean
 29
 30_T = TypeVar("_T")
 31
 32AsyncMockType = unittest.mock.AsyncMock
 33MockType = Union[
 34    unittest.mock.MagicMock,
 35    unittest.mock.AsyncMock,
 36    unittest.mock.NonCallableMagicMock,
 37]
 38
 39
 40class PytestMockWarning(UserWarning):
 41    """Base class for all warnings emitted by pytest-mock."""
 42
 43
 44@dataclass
 45class MockCacheItem:
 46    mock: MockType
 47    patch: Optional[Any] = None
 48
 49
 50@dataclass
 51class MockCache:
 52    """
 53    Cache MagicMock and Patcher instances so we can undo them later.
 54    """
 55
 56    cache: List[MockCacheItem] = field(default_factory=list)
 57
 58    def _find(self, mock: MockType) -> MockCacheItem:
 59        for mock_item in self.cache:
 60            if mock_item.mock is mock:
 61                return mock_item
 62        raise ValueError("This mock object is not registered")
 63
 64    def add(self, mock: MockType, **kwargs: Any) -> MockCacheItem:
 65        self.cache.append(MockCacheItem(mock=mock, **kwargs))
 66        return self.cache[-1]
 67
 68    def remove(self, mock: MockType) -> None:
 69        mock_item = self._find(mock)
 70        if mock_item.patch:
 71            mock_item.patch.stop()
 72        self.cache.remove(mock_item)
 73
 74    def clear(self) -> None:
 75        for mock_item in reversed(self.cache):
 76            if mock_item.patch is not None:
 77                mock_item.patch.stop()
 78        self.cache.clear()
 79
 80    def __iter__(self) -> Iterator[MockCacheItem]:
 81        return iter(self.cache)
 82
 83
 84class MockerFixture:
 85    """
 86    Fixture that provides the same interface to functions in the mock module,
 87    ensuring that they are uninstalled at the end of each test.
 88    """
 89
 90    def __init__(self, config: Any) -> None:
 91        self._mock_cache: MockCache = MockCache()
 92        self.mock_module = mock_module = get_mock_module(config)
 93        self.patch = self._Patcher(self._mock_cache, mock_module)  # type: MockerFixture._Patcher
 94        # aliases for convenience
 95        self.Mock = mock_module.Mock
 96        self.MagicMock = mock_module.MagicMock
 97        self.NonCallableMock = mock_module.NonCallableMock
 98        self.NonCallableMagicMock = mock_module.NonCallableMagicMock
 99        self.PropertyMock = mock_module.PropertyMock
100        if hasattr(mock_module, "AsyncMock"):
101            self.AsyncMock = mock_module.AsyncMock
102        self.call = mock_module.call
103        self.ANY = mock_module.ANY
104        self.DEFAULT = mock_module.DEFAULT
105        self.sentinel = mock_module.sentinel
106        self.mock_open = mock_module.mock_open
107        if hasattr(mock_module, "seal"):
108            self.seal = mock_module.seal
109
110    def create_autospec(
111        self, spec: Any, spec_set: bool = False, instance: bool = False, **kwargs: Any
112    ) -> MockType:
113        m: MockType = self.mock_module.create_autospec(
114            spec, spec_set, instance, **kwargs
115        )
116        self._mock_cache.add(m)
117        return m
118
119    def resetall(
120        self, *, return_value: bool = False, side_effect: bool = False
121    ) -> None:
122        """
123        Call reset_mock() on all patchers started by this fixture.
124
125        :param bool return_value: Reset the return_value of mocks.
126        :param bool side_effect: Reset the side_effect of mocks.
127        """
128        supports_reset_mock_with_args: Tuple[Type[Any], ...]
129        if hasattr(self, "AsyncMock"):
130            supports_reset_mock_with_args = (self.Mock, self.AsyncMock)
131        else:
132            supports_reset_mock_with_args = (self.Mock,)
133
134        for mock_item in self._mock_cache:
135            # See issue #237.
136            if not hasattr(mock_item.mock, "reset_mock"):
137                continue
138            # NOTE: The mock may be a dictionary
139            if hasattr(mock_item.mock, "spy_return_list"):
140                mock_item.mock.spy_return_list = []
141            if isinstance(mock_item.mock, supports_reset_mock_with_args):
142                mock_item.mock.reset_mock(
143                    return_value=return_value, side_effect=side_effect
144                )
145            else:
146                mock_item.mock.reset_mock()
147
148    def stopall(self) -> None:
149        """
150        Stop all patchers started by this fixture. Can be safely called multiple
151        times.
152        """
153        self._mock_cache.clear()
154
155    def stop(self, mock: unittest.mock.MagicMock) -> None:
156        """
157        Stops a previous patch or spy call by passing the ``MagicMock`` object
158        returned by it.
159        """
160        self._mock_cache.remove(mock)
161
162    def spy(self, obj: object, name: str) -> MockType:
163        """
164        Create a spy of method. It will run method normally, but it is now
165        possible to use `mock` call features with it, like call count.
166
167        :param obj: An object.
168        :param name: A method in object.
169        :return: Spy object.
170        """
171        method = getattr(obj, name)
172
173        def wrapper(*args, **kwargs):
174            spy_obj.spy_return = None
175            spy_obj.spy_exception = None
176            try:
177                r = method(*args, **kwargs)
178            except BaseException as e:
179                spy_obj.spy_exception = e
180                raise
181            else:
182                spy_obj.spy_return = r
183                spy_obj.spy_return_list.append(r)
184            return r
185
186        async def async_wrapper(*args, **kwargs):
187            spy_obj.spy_return = None
188            spy_obj.spy_exception = None
189            try:
190                r = await method(*args, **kwargs)
191            except BaseException as e:
192                spy_obj.spy_exception = e
193                raise
194            else:
195                spy_obj.spy_return = r
196                spy_obj.spy_return_list.append(r)
197            return r
198
199        if asyncio.iscoroutinefunction(method):
200            wrapped = functools.update_wrapper(async_wrapper, method)
201        else:
202            wrapped = functools.update_wrapper(wrapper, method)
203
204        autospec = inspect.ismethod(method) or inspect.isfunction(method)
205
206        spy_obj = self.patch.object(obj, name, side_effect=wrapped, autospec=autospec)
207        spy_obj.spy_return = None
208        spy_obj.spy_return_list = []
209        spy_obj.spy_exception = None
210        return spy_obj
211
212    def stub(self, name: Optional[str] = None) -> unittest.mock.MagicMock:
213        """
214        Create a stub method. It accepts any arguments. Ideal to register to
215        callbacks in tests.
216
217        :param name: the constructed stub's name as used in repr
218        :return: Stub object.
219        """
220        return cast(
221            unittest.mock.MagicMock,
222            self.mock_module.MagicMock(spec=lambda *args, **kwargs: None, name=name),
223        )
224
225    def async_stub(self, name: Optional[str] = None) -> AsyncMockType:
226        """
227        Create a async stub method. It accepts any arguments. Ideal to register to
228        callbacks in tests.
229
230        :param name: the constructed stub's name as used in repr
231        :return: Stub object.
232        """
233        return cast(
234            AsyncMockType,
235            self.mock_module.AsyncMock(spec=lambda *args, **kwargs: None, name=name),
236        )
237
238    class _Patcher:
239        """
240        Object to provide the same interface as mock.patch, mock.patch.object,
241        etc. We need this indirection to keep the same API of the mock package.
242        """
243
244        DEFAULT = object()
245
246        def __init__(self, mock_cache, mock_module):
247            self.__mock_cache = mock_cache
248            self.mock_module = mock_module
249
250        def _start_patch(
251            self, mock_func: Any, warn_on_mock_enter: bool, *args: Any, **kwargs: Any
252        ) -> MockType:
253            """Patches something by calling the given function from the mock
254            module, registering the patch to stop it later and returns the
255            mock object resulting from the mock call.
256            """
257            p = mock_func(*args, **kwargs)
258            mocked: MockType = p.start()
259            self.__mock_cache.add(mock=mocked, patch=p)
260            if hasattr(mocked, "reset_mock"):
261                # check if `mocked` is actually a mock object, as depending on autospec or target
262                # parameters `mocked` can be anything
263                if hasattr(mocked, "__enter__") and warn_on_mock_enter:
264                    mocked.__enter__.side_effect = lambda: warnings.warn(
265                        "Mocks returned by pytest-mock do not need to be used as context managers. "
266                        "The mocker fixture automatically undoes mocking at the end of a test. "
267                        "This warning can be ignored if it was triggered by mocking a context manager. "
268                        "https://pytest-mock.readthedocs.io/en/latest/remarks.html#usage-as-context-manager",
269                        PytestMockWarning,
270                        stacklevel=5,
271                    )
272            return mocked
273
274        def object(
275            self,
276            target: object,
277            attribute: str,
278            new: object = DEFAULT,
279            spec: Optional[object] = None,
280            create: bool = False,
281            spec_set: Optional[object] = None,
282            autospec: Optional[object] = None,
283            new_callable: object = None,
284            **kwargs: Any,
285        ) -> MockType:
286            """API to mock.patch.object"""
287            if new is self.DEFAULT:
288                new = self.mock_module.DEFAULT
289            return self._start_patch(
290                self.mock_module.patch.object,
291                True,
292                target,
293                attribute,
294                new=new,
295                spec=spec,
296                create=create,
297                spec_set=spec_set,
298                autospec=autospec,
299                new_callable=new_callable,
300                **kwargs,
301            )
302
303        def context_manager(
304            self,
305            target: builtins.object,
306            attribute: str,
307            new: builtins.object = DEFAULT,
308            spec: Optional[builtins.object] = None,
309            create: bool = False,
310            spec_set: Optional[builtins.object] = None,
311            autospec: Optional[builtins.object] = None,
312            new_callable: builtins.object = None,
313            **kwargs: Any,
314        ) -> MockType:
315            """This is equivalent to mock.patch.object except that the returned mock
316            does not issue a warning when used as a context manager."""
317            if new is self.DEFAULT:
318                new = self.mock_module.DEFAULT
319            return self._start_patch(
320                self.mock_module.patch.object,
321                False,
322                target,
323                attribute,
324                new=new,
325                spec=spec,
326                create=create,
327                spec_set=spec_set,
328                autospec=autospec,
329                new_callable=new_callable,
330                **kwargs,
331            )
332
333        def multiple(
334            self,
335            target: builtins.object,
336            spec: Optional[builtins.object] = None,
337            create: bool = False,
338            spec_set: Optional[builtins.object] = None,
339            autospec: Optional[builtins.object] = None,
340            new_callable: Optional[builtins.object] = None,
341            **kwargs: Any,
342        ) -> Dict[str, MockType]:
343            """API to mock.patch.multiple"""
344            return self._start_patch(
345                self.mock_module.patch.multiple,
346                True,
347                target,
348                spec=spec,
349                create=create,
350                spec_set=spec_set,
351                autospec=autospec,
352                new_callable=new_callable,
353                **kwargs,
354            )
355
356        def dict(
357            self,
358            in_dict: Union[Mapping[Any, Any], str],
359            values: Union[Mapping[Any, Any], Iterable[Tuple[Any, Any]]] = (),
360            clear: bool = False,
361            **kwargs: Any,
362        ) -> Any:
363            """API to mock.patch.dict"""
364            return self._start_patch(
365                self.mock_module.patch.dict,
366                True,
367                in_dict,
368                values=values,
369                clear=clear,
370                **kwargs,
371            )
372
373        @overload
374        def __call__(
375            self,
376            target: str,
377            new: None = ...,
378            spec: Optional[builtins.object] = ...,
379            create: bool = ...,
380            spec_set: Optional[builtins.object] = ...,
381            autospec: Optional[builtins.object] = ...,
382            new_callable: None = ...,
383            **kwargs: Any,
384        ) -> MockType: ...
385
386        @overload
387        def __call__(
388            self,
389            target: str,
390            new: _T,
391            spec: Optional[builtins.object] = ...,
392            create: bool = ...,
393            spec_set: Optional[builtins.object] = ...,
394            autospec: Optional[builtins.object] = ...,
395            new_callable: None = ...,
396            **kwargs: Any,
397        ) -> _T: ...
398
399        @overload
400        def __call__(
401            self,
402            target: str,
403            new: None,
404            spec: Optional[builtins.object],
405            create: bool,
406            spec_set: Optional[builtins.object],
407            autospec: Optional[builtins.object],
408            new_callable: Callable[[], _T],
409            **kwargs: Any,
410        ) -> _T: ...
411
412        @overload
413        def __call__(
414            self,
415            target: str,
416            new: None = ...,
417            spec: Optional[builtins.object] = ...,
418            create: bool = ...,
419            spec_set: Optional[builtins.object] = ...,
420            autospec: Optional[builtins.object] = ...,
421            *,
422            new_callable: Callable[[], _T],
423            **kwargs: Any,
424        ) -> _T: ...
425
426        def __call__(
427            self,
428            target: str,
429            new: builtins.object = DEFAULT,
430            spec: Optional[builtins.object] = None,
431            create: bool = False,
432            spec_set: Optional[builtins.object] = None,
433            autospec: Optional[builtins.object] = None,
434            new_callable: Optional[Callable[[], Any]] = None,
435            **kwargs: Any,
436        ) -> Any:
437            """API to mock.patch"""
438            if new is self.DEFAULT:
439                new = self.mock_module.DEFAULT
440            return self._start_patch(
441                self.mock_module.patch,
442                True,
443                target,
444                new=new,
445                spec=spec,
446                create=create,
447                spec_set=spec_set,
448                autospec=autospec,
449                new_callable=new_callable,
450                **kwargs,
451            )
452
453
454def _mocker(pytestconfig: Any) -> Generator[MockerFixture, None, None]:
455    """
456    Return an object that has the same interface to the `mock` module, but
457    takes care of automatically undoing all patches after each test method.
458    """
459    result = MockerFixture(pytestconfig)
460    yield result
461    result.stopall()
462
463
464mocker = pytest.fixture()(_mocker)  # default scope is function
465class_mocker = pytest.fixture(scope="class")(_mocker)
466module_mocker = pytest.fixture(scope="module")(_mocker)
467package_mocker = pytest.fixture(scope="package")(_mocker)
468session_mocker = pytest.fixture(scope="session")(_mocker)
469
470
471_mock_module_patches = []  # type: List[Any]
472_mock_module_originals = {}  # type: Dict[str, Any]
473
474
475def assert_wrapper(
476    __wrapped_mock_method__: Callable[..., Any], *args: Any, **kwargs: Any
477) -> None:
478    __tracebackhide__ = True
479    try:
480        __wrapped_mock_method__(*args, **kwargs)
481        return
482    except AssertionError as e:
483        if getattr(e, "_mock_introspection_applied", 0):
484            msg = str(e)
485        else:
486            __mock_self = args[0]
487            msg = str(e)
488            if __mock_self.call_args is not None:
489                actual_args, actual_kwargs = __mock_self.call_args
490                introspection = ""
491                try:
492                    assert actual_args == args[1:]
493                except AssertionError as e_args:
494                    introspection += "\nArgs:\n" + str(e_args)
495                try:
496                    assert actual_kwargs == kwargs
497                except AssertionError as e_kwargs:
498                    introspection += "\nKwargs:\n" + str(e_kwargs)
499                if introspection:
500                    msg += "\n\npytest introspection follows:\n" + introspection
501        e = AssertionError(msg)
502        e._mock_introspection_applied = True  # type:ignore[attr-defined]
503        raise e
504
505
506def assert_has_calls_wrapper(
507    __wrapped_mock_method__: Callable[..., Any], *args: Any, **kwargs: Any
508) -> None:
509    __tracebackhide__ = True
510    try:
511        __wrapped_mock_method__(*args, **kwargs)
512        return
513    except AssertionError as e:
514        any_order = kwargs.get("any_order", False)
515        if getattr(e, "_mock_introspection_applied", 0) or any_order:
516            msg = str(e)
517        else:
518            __mock_self = args[0]
519            msg = str(e)
520            if __mock_self.call_args_list is not None:
521                actual_calls = list(__mock_self.call_args_list)
522                expect_calls = args[1]
523                introspection = ""
524                from itertools import zip_longest
525
526                for actual_call, expect_call in zip_longest(actual_calls, expect_calls):
527                    if actual_call is not None:
528                        actual_args, actual_kwargs = actual_call
529                    else:
530                        actual_args = tuple()
531                        actual_kwargs = {}
532
533                    if expect_call is not None:
534                        _, expect_args, expect_kwargs = expect_call
535                    else:
536                        expect_args = tuple()
537                        expect_kwargs = {}
538
539                    try:
540                        assert actual_args == expect_args
541                    except AssertionError as e_args:
542                        introspection += "\nArgs:\n" + str(e_args)
543                    try:
544                        assert actual_kwargs == expect_kwargs
545                    except AssertionError as e_kwargs:
546                        introspection += "\nKwargs:\n" + str(e_kwargs)
547                if introspection:
548                    msg += "\n\npytest introspection follows:\n" + introspection
549        e = AssertionError(msg)
550        e._mock_introspection_applied = True  # type:ignore[attr-defined]
551        raise e
552
553
554def wrap_assert_not_called(*args: Any, **kwargs: Any) -> None:
555    __tracebackhide__ = True
556    assert_wrapper(_mock_module_originals["assert_not_called"], *args, **kwargs)
557
558
559def wrap_assert_called_with(*args: Any, **kwargs: Any) -> None:
560    __tracebackhide__ = True
561    assert_wrapper(_mock_module_originals["assert_called_with"], *args, **kwargs)
562
563
564def wrap_assert_called_once(*args: Any, **kwargs: Any) -> None:
565    __tracebackhide__ = True
566    assert_wrapper(_mock_module_originals["assert_called_once"], *args, **kwargs)
567
568
569def wrap_assert_called_once_with(*args: Any, **kwargs: Any) -> None:
570    __tracebackhide__ = True
571    assert_wrapper(_mock_module_originals["assert_called_once_with"], *args, **kwargs)
572
573
574def wrap_assert_has_calls(*args: Any, **kwargs: Any) -> None:
575    __tracebackhide__ = True
576    assert_has_calls_wrapper(
577        _mock_module_originals["assert_has_calls"], *args, **kwargs
578    )
579
580
581def wrap_assert_any_call(*args: Any, **kwargs: Any) -> None:
582    __tracebackhide__ = True
583    assert_wrapper(_mock_module_originals["assert_any_call"], *args, **kwargs)
584
585
586def wrap_assert_called(*args: Any, **kwargs: Any) -> None:
587    __tracebackhide__ = True
588    assert_wrapper(_mock_module_originals["assert_called"], *args, **kwargs)
589
590
591def wrap_assert_not_awaited(*args: Any, **kwargs: Any) -> None:
592    __tracebackhide__ = True
593    assert_wrapper(_mock_module_originals["assert_not_awaited"], *args, **kwargs)
594
595
596def wrap_assert_awaited_with(*args: Any, **kwargs: Any) -> None:
597    __tracebackhide__ = True
598    assert_wrapper(_mock_module_originals["assert_awaited_with"], *args, **kwargs)
599
600
601def wrap_assert_awaited_once(*args: Any, **kwargs: Any) -> None:
602    __tracebackhide__ = True
603    assert_wrapper(_mock_module_originals["assert_awaited_once"], *args, **kwargs)
604
605
606def wrap_assert_awaited_once_with(*args: Any, **kwargs: Any) -> None:
607    __tracebackhide__ = True
608    assert_wrapper(_mock_module_originals["assert_awaited_once_with"], *args, **kwargs)
609
610
611def wrap_assert_has_awaits(*args: Any, **kwargs: Any) -> None:
612    __tracebackhide__ = True
613    assert_wrapper(_mock_module_originals["assert_has_awaits"], *args, **kwargs)
614
615
616def wrap_assert_any_await(*args: Any, **kwargs: Any) -> None:
617    __tracebackhide__ = True
618    assert_wrapper(_mock_module_originals["assert_any_await"], *args, **kwargs)
619
620
621def wrap_assert_awaited(*args: Any, **kwargs: Any) -> None:
622    __tracebackhide__ = True
623    assert_wrapper(_mock_module_originals["assert_awaited"], *args, **kwargs)
624
625
626def wrap_assert_methods(config: Any) -> None:
627    """
628    Wrap assert methods of mock module so we can hide their traceback and
629    add introspection information to specified argument asserts.
630    """
631    # Make sure we only do this once
632    if _mock_module_originals:
633        return
634
635    mock_module = get_mock_module(config)
636
637    wrappers = {
638        "assert_called": wrap_assert_called,
639        "assert_called_once": wrap_assert_called_once,
640        "assert_called_with": wrap_assert_called_with,
641        "assert_called_once_with": wrap_assert_called_once_with,
642        "assert_any_call": wrap_assert_any_call,
643        "assert_has_calls": wrap_assert_has_calls,
644        "assert_not_called": wrap_assert_not_called,
645    }
646    for method, wrapper in wrappers.items():
647        try:
648            original = getattr(mock_module.NonCallableMock, method)
649        except AttributeError:  # pragma: no cover
650            continue
651        _mock_module_originals[method] = original
652        patcher = mock_module.patch.object(mock_module.NonCallableMock, method, wrapper)
653        patcher.start()
654        _mock_module_patches.append(patcher)
655
656    if hasattr(mock_module, "AsyncMock"):
657        async_wrappers = {
658            "assert_awaited": wrap_assert_awaited,
659            "assert_awaited_once": wrap_assert_awaited_once,
660            "assert_awaited_with": wrap_assert_awaited_with,
661            "assert_awaited_once_with": wrap_assert_awaited_once_with,
662            "assert_any_await": wrap_assert_any_await,
663            "assert_has_awaits": wrap_assert_has_awaits,
664            "assert_not_awaited": wrap_assert_not_awaited,
665        }
666        for method, wrapper in async_wrappers.items():
667            try:
668                original = getattr(mock_module.AsyncMock, method)
669            except AttributeError:  # pragma: no cover
670                continue
671            _mock_module_originals[method] = original
672            patcher = mock_module.patch.object(mock_module.AsyncMock, method, wrapper)
673            patcher.start()
674            _mock_module_patches.append(patcher)
675
676    config.add_cleanup(unwrap_assert_methods)
677
678
679def unwrap_assert_methods() -> None:
680    for patcher in _mock_module_patches:
681        try:
682            patcher.stop()
683        except RuntimeError as e:
684            # a patcher might have been stopped by user code (#137)
685            # so we need to catch this error here and ignore it;
686            # unfortunately there's no public API to check if a patch
687            # has been started, so catching the error it is
688            if str(e) == "stop called on unstarted patcher":
689                pass
690            else:
691                raise
692    _mock_module_patches[:] = []
693    _mock_module_originals.clear()
694
695
696def pytest_addoption(parser: Any) -> None:
697    parser.addini(
698        "mock_traceback_monkeypatch",
699        "Monkeypatch the mock library to improve reporting of the "
700        "assert_called_... methods",
701        default=True,
702    )
703    parser.addini(
704        "mock_use_standalone_module",
705        'Use standalone "mock" (from PyPI) instead of builtin "unittest.mock" '
706        "on Python 3",
707        default=False,
708    )
709
710
711def pytest_configure(config: Any) -> None:
712    tb = config.getoption("--tb", default="auto")
713    if (
714        parse_ini_boolean(config.getini("mock_traceback_monkeypatch"))
715        and tb != "native"
716    ):
717        wrap_assert_methods(config)