pytest_mock.plugin

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