pytest_mock.plugin

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