pytest

 56@final
 57@dataclasses.dataclass
 58class Cache:
 59    """Instance of the `cache` fixture."""
 60
 61    _cachedir: Path = dataclasses.field(repr=False)
 62    _config: Config = dataclasses.field(repr=False)
 63
 64    # Sub-directory under cache-dir for directories created by `mkdir()`.
 65    _CACHE_PREFIX_DIRS = "d"
 66
 67    # Sub-directory under cache-dir for values created by `set()`.
 68    _CACHE_PREFIX_VALUES = "v"
 69
 70    def __init__(
 71        self, cachedir: Path, config: Config, *, _ispytest: bool = False
 72    ) -> None:
 73        check_ispytest(_ispytest)
 74        self._cachedir = cachedir
 75        self._config = config
 76
 77    @classmethod
 78    def for_config(cls, config: Config, *, _ispytest: bool = False) -> Cache:
 79        """Create the Cache instance for a Config.
 80
 81        :meta private:
 82        """
 83        check_ispytest(_ispytest)
 84        cachedir = cls.cache_dir_from_config(config, _ispytest=True)
 85        if config.getoption("cacheclear") and cachedir.is_dir():
 86            cls.clear_cache(cachedir, _ispytest=True)
 87        return cls(cachedir, config, _ispytest=True)
 88
 89    @classmethod
 90    def clear_cache(cls, cachedir: Path, _ispytest: bool = False) -> None:
 91        """Clear the sub-directories used to hold cached directories and values.
 92
 93        :meta private:
 94        """
 95        check_ispytest(_ispytest)
 96        for prefix in (cls._CACHE_PREFIX_DIRS, cls._CACHE_PREFIX_VALUES):
 97            d = cachedir / prefix
 98            if d.is_dir():
 99                rm_rf(d)
100
101    @staticmethod
102    def cache_dir_from_config(config: Config, *, _ispytest: bool = False) -> Path:
103        """Get the path to the cache directory for a Config.
104
105        :meta private:
106        """
107        check_ispytest(_ispytest)
108        return resolve_from_str(config.getini("cache_dir"), config.rootpath)
109
110    def warn(self, fmt: str, *, _ispytest: bool = False, **args: object) -> None:
111        """Issue a cache warning.
112
113        :meta private:
114        """
115        check_ispytest(_ispytest)
116        import warnings
117
118        from _pytest.warning_types import PytestCacheWarning
119
120        warnings.warn(
121            PytestCacheWarning(fmt.format(**args) if args else fmt),
122            self._config.hook,
123            stacklevel=3,
124        )
125
126    def _mkdir(self, path: Path) -> None:
127        self._ensure_cache_dir_and_supporting_files()
128        path.mkdir(exist_ok=True, parents=True)
129
130    def mkdir(self, name: str) -> Path:
131        """Return a directory path object with the given name.
132
133        If the directory does not yet exist, it will be created. You can use
134        it to manage files to e.g. store/retrieve database dumps across test
135        sessions.
136
137        .. versionadded:: 7.0
138
139        :param name:
140            Must be a string not containing a ``/`` separator.
141            Make sure the name contains your plugin or application
142            identifiers to prevent clashes with other cache users.
143        """
144        path = Path(name)
145        if len(path.parts) > 1:
146            raise ValueError("name is not allowed to contain path separators")
147        res = self._cachedir.joinpath(self._CACHE_PREFIX_DIRS, path)
148        self._mkdir(res)
149        return res
150
151    def _getvaluepath(self, key: str) -> Path:
152        return self._cachedir.joinpath(self._CACHE_PREFIX_VALUES, Path(key))
153
154    def get(self, key: str, default):
155        """Return the cached value for the given key.
156
157        If no value was yet cached or the value cannot be read, the specified
158        default is returned.
159
160        :param key:
161            Must be a ``/`` separated value. Usually the first
162            name is the name of your plugin or your application.
163        :param default:
164            The value to return in case of a cache-miss or invalid cache value.
165        """
166        path = self._getvaluepath(key)
167        try:
168            with path.open("r", encoding="UTF-8") as f:
169                return json.load(f)
170        except (ValueError, OSError):
171            return default
172
173    def set(self, key: str, value: object) -> None:
174        """Save value for the given key.
175
176        :param key:
177            Must be a ``/`` separated value. Usually the first
178            name is the name of your plugin or your application.
179        :param value:
180            Must be of any combination of basic python types,
181            including nested types like lists of dictionaries.
182        """
183        path = self._getvaluepath(key)
184        try:
185            self._mkdir(path.parent)
186        except OSError as exc:
187            self.warn(
188                f"could not create cache path {path}: {exc}",
189                _ispytest=True,
190            )
191            return
192        data = json.dumps(value, ensure_ascii=False, indent=2)
193        try:
194            f = path.open("w", encoding="UTF-8")
195        except OSError as exc:
196            self.warn(
197                f"cache could not write path {path}: {exc}",
198                _ispytest=True,
199            )
200        else:
201            with f:
202                f.write(data)
203
204    def _ensure_cache_dir_and_supporting_files(self) -> None:
205        """Create the cache dir and its supporting files."""
206        if self._cachedir.is_dir():
207            return
208
209        self._cachedir.parent.mkdir(parents=True, exist_ok=True)
210        with tempfile.TemporaryDirectory(
211            prefix="pytest-cache-files-",
212            dir=self._cachedir.parent,
213        ) as newpath:
214            path = Path(newpath)
215
216            # Reset permissions to the default, see #12308.
217            # Note: there's no way to get the current umask atomically, eek.
218            umask = os.umask(0o022)
219            os.umask(umask)
220            path.chmod(0o777 - umask)
221
222            with open(path.joinpath("README.md"), "x", encoding="UTF-8") as f:
223                f.write(README_CONTENT)
224            with open(path.joinpath(".gitignore"), "x", encoding="UTF-8") as f:
225                f.write("# Created by pytest automatically.\n*\n")
226            with open(path.joinpath("CACHEDIR.TAG"), "xb") as f:
227                f.write(CACHEDIR_TAG_CONTENT)
228
229            try:
230                path.rename(self._cachedir)
231            except OSError as e:
232                # If 2 concurrent pytests both race to the rename, the loser
233                # gets "Directory not empty" from the rename. In this case,
234                # everything is handled so just continue (while letting the
235                # temporary directory be cleaned up).
236                # On Windows, the error is a FileExistsError which translates to EEXIST.
237                if e.errno not in (errno.ENOTEMPTY, errno.EEXIST):
238                    raise
239            else:
240                # Create a directory in place of the one we just moved so that
241                # `TemporaryDirectory`'s cleanup doesn't complain.
242                #
243                # TODO: pass ignore_cleanup_errors=True when we no longer support python < 3.10.
244                # See https://github.com/python/cpython/issues/74168. Note that passing
245                # delete=False would do the wrong thing in case of errors and isn't supported
246                # until python 3.12.
247                path.mkdir()

284@final
285@dataclasses.dataclass
286class CallInfo(Generic[TResult]):
287    """Result/Exception info of a function invocation."""
288
289    _result: TResult | None
290    #: The captured exception of the call, if it raised.
291    excinfo: ExceptionInfo[BaseException] | None
292    #: The system time when the call started, in seconds since the epoch.
293    start: float
294    #: The system time when the call ended, in seconds since the epoch.
295    stop: float
296    #: The call duration, in seconds.
297    duration: float
298    #: The context of invocation: "collect", "setup", "call" or "teardown".
299    when: Literal["collect", "setup", "call", "teardown"]
300
301    def __init__(
302        self,
303        result: TResult | None,
304        excinfo: ExceptionInfo[BaseException] | None,
305        start: float,
306        stop: float,
307        duration: float,
308        when: Literal["collect", "setup", "call", "teardown"],
309        *,
310        _ispytest: bool = False,
311    ) -> None:
312        check_ispytest(_ispytest)
313        self._result = result
314        self.excinfo = excinfo
315        self.start = start
316        self.stop = stop
317        self.duration = duration
318        self.when = when
319
320    @property
321    def result(self) -> TResult:
322        """The return value of the call, if it didn't raise.
323
324        Can only be accessed if excinfo is None.
325        """
326        if self.excinfo is not None:
327            raise AttributeError(f"{self!r} has no valid result")
328        # The cast is safe because an exception wasn't raised, hence
329        # _result has the expected function return type (which may be
330        #  None, that's why a cast and not an assert).
331        return cast(TResult, self._result)
332
333    @classmethod
334    def from_call(
335        cls,
336        func: Callable[[], TResult],
337        when: Literal["collect", "setup", "call", "teardown"],
338        reraise: type[BaseException] | tuple[type[BaseException], ...] | None = None,
339    ) -> CallInfo[TResult]:
340        """Call func, wrapping the result in a CallInfo.
341
342        :param func:
343            The function to call. Called without arguments.
344        :type func: Callable[[], _pytest.runner.TResult]
345        :param when:
346            The phase in which the function is called.
347        :param reraise:
348            Exception or exceptions that shall propagate if raised by the
349            function, instead of being wrapped in the CallInfo.
350        """
351        excinfo = None
352        instant = timing.Instant()
353        try:
354            result: TResult | None = func()
355        except BaseException:
356            excinfo = ExceptionInfo.from_current()
357            if reraise is not None and isinstance(excinfo.value, reraise):
358                raise
359            result = None
360        duration = instant.elapsed()
361        return cls(
362            start=duration.start.time,
363            stop=duration.stop.time,
364            duration=duration.seconds,
365            when=when,
366            result=result,
367            excinfo=excinfo,
368            _ispytest=True,
369        )
370
371    def __repr__(self) -> str:
372        if self.excinfo is None:
373            return f"<CallInfo when={self.when!r} result: {self._result!r}>"
374        return f"<CallInfo when={self.when!r} excinfo={self.excinfo!r}>"

917class CaptureFixture(Generic[AnyStr]):
918    """Object returned by the :fixture:`capsys`, :fixture:`capsysbinary`,
919    :fixture:`capfd` and :fixture:`capfdbinary` fixtures."""
920
921    def __init__(
922        self,
923        captureclass: type[CaptureBase[AnyStr]],
924        request: SubRequest,
925        *,
926        config: dict[str, Any] | None = None,
927        _ispytest: bool = False,
928    ) -> None:
929        check_ispytest(_ispytest)
930        self.captureclass: type[CaptureBase[AnyStr]] = captureclass
931        self.request = request
932        self._config = config if config else {}
933        self._capture: MultiCapture[AnyStr] | None = None
934        self._captured_out: AnyStr = self.captureclass.EMPTY_BUFFER
935        self._captured_err: AnyStr = self.captureclass.EMPTY_BUFFER
936
937    def _start(self) -> None:
938        if self._capture is None:
939            self._capture = MultiCapture(
940                in_=None,
941                out=self.captureclass(1, **self._config),
942                err=self.captureclass(2, **self._config),
943            )
944            self._capture.start_capturing()
945
946    def close(self) -> None:
947        if self._capture is not None:
948            out, err = self._capture.pop_outerr_to_orig()
949            self._captured_out += out
950            self._captured_err += err
951            self._capture.stop_capturing()
952            self._capture = None
953
954    def readouterr(self) -> CaptureResult[AnyStr]:
955        """Read and return the captured output so far, resetting the internal
956        buffer.
957
958        :returns:
959            The captured content as a namedtuple with ``out`` and ``err``
960            string attributes.
961        """
962        captured_out, captured_err = self._captured_out, self._captured_err
963        if self._capture is not None:
964            out, err = self._capture.readouterr()
965            captured_out += out
966            captured_err += err
967        self._captured_out = self.captureclass.EMPTY_BUFFER
968        self._captured_err = self.captureclass.EMPTY_BUFFER
969        return CaptureResult(captured_out, captured_err)
970
971    def _suspend(self) -> None:
972        """Suspend this fixture's own capturing temporarily."""
973        if self._capture is not None:
974            self._capture.suspend_capturing()
975
976    def _resume(self) -> None:
977        """Resume this fixture's own capturing temporarily."""
978        if self._capture is not None:
979            self._capture.resume_capturing()
980
981    def _is_started(self) -> bool:
982        """Whether actively capturing -- not disabled or closed."""
983        if self._capture is not None:
984            return self._capture.is_started()
985        return False
986
987    @contextlib.contextmanager
988    def disabled(self) -> Generator[None]:
989        """Temporarily disable capturing while inside the ``with`` block."""
990        capmanager: CaptureManager = self.request.config.pluginmanager.getplugin(
991            "capturemanager"
992        )
993        with capmanager.global_and_fixture_disabled():
994            yield

746class Class(PyCollector):
747    """Collector for test methods (and nested classes) in a Python class."""
748
749    @classmethod
750    def from_parent(cls, parent, *, name, obj=None, **kw) -> Self:  # type: ignore[override]
751        """The public constructor."""
752        return super().from_parent(name=name, parent=parent, **kw)
753
754    def newinstance(self):
755        return self.obj()
756
757    def collect(self) -> Iterable[nodes.Item | nodes.Collector]:
758        if not safe_getattr(self.obj, "__test__", True):
759            return []
760        if hasinit(self.obj):
761            assert self.parent is not None
762            self.warn(
763                PytestCollectionWarning(
764                    f"cannot collect test class {self.obj.__name__!r} because it has a "
765                    f"__init__ constructor (from: {self.parent.nodeid})"
766                )
767            )
768            return []
769        elif hasnew(self.obj):
770            assert self.parent is not None
771            self.warn(
772                PytestCollectionWarning(
773                    f"cannot collect test class {self.obj.__name__!r} because it has a "
774                    f"__new__ constructor (from: {self.parent.nodeid})"
775                )
776            )
777            return []
778
779        self._register_setup_class_fixture()
780        self._register_setup_method_fixture()
781
782        self.session._fixturemanager.parsefactories(self.newinstance(), self.nodeid)
783
784        return super().collect()
785
786    def _register_setup_class_fixture(self) -> None:
787        """Register an autouse, class scoped fixture into the collected class object
788        that invokes setup_class/teardown_class if either or both are available.
789
790        Using a fixture to invoke this methods ensures we play nicely and unsurprisingly with
791        other fixtures (#517).
792        """
793        setup_class = _get_first_non_fixture_func(self.obj, ("setup_class",))
794        teardown_class = _get_first_non_fixture_func(self.obj, ("teardown_class",))
795        if setup_class is None and teardown_class is None:
796            return
797
798        def xunit_setup_class_fixture(request) -> Generator[None]:
799            cls = request.cls
800            if setup_class is not None:
801                func = getimfunc(setup_class)
802                _call_with_optional_argument(func, cls)
803            yield
804            if teardown_class is not None:
805                func = getimfunc(teardown_class)
806                _call_with_optional_argument(func, cls)
807
808        self.session._fixturemanager._register_fixture(
809            # Use a unique name to speed up lookup.
810            name=f"_xunit_setup_class_fixture_{self.obj.__qualname__}",
811            func=xunit_setup_class_fixture,
812            nodeid=self.nodeid,
813            scope="class",
814            autouse=True,
815        )
816
817    def _register_setup_method_fixture(self) -> None:
818        """Register an autouse, function scoped fixture into the collected class object
819        that invokes setup_method/teardown_method if either or both are available.
820
821        Using a fixture to invoke these methods ensures we play nicely and unsurprisingly with
822        other fixtures (#517).
823        """
824        setup_name = "setup_method"
825        setup_method = _get_first_non_fixture_func(self.obj, (setup_name,))
826        teardown_name = "teardown_method"
827        teardown_method = _get_first_non_fixture_func(self.obj, (teardown_name,))
828        if setup_method is None and teardown_method is None:
829            return
830
831        def xunit_setup_method_fixture(request) -> Generator[None]:
832            instance = request.instance
833            method = request.function
834            if setup_method is not None:
835                func = getattr(instance, setup_name)
836                _call_with_optional_argument(func, method)
837            yield
838            if teardown_method is not None:
839                func = getattr(instance, teardown_name)
840                _call_with_optional_argument(func, method)
841
842        self.session._fixturemanager._register_fixture(
843            # Use a unique name to speed up lookup.
844            name=f"_xunit_setup_method_fixture_{self.obj.__qualname__}",
845            func=xunit_setup_method_fixture,
846            nodeid=self.nodeid,
847            scope="function",
848            autouse=True,
849        )

457@final
458class CollectReport(BaseReport):
459    """Collection report object.
460
461    Reports can contain arbitrary extra attributes.
462    """
463
464    when = "collect"
465
466    def __init__(
467        self,
468        nodeid: str,
469        outcome: Literal["passed", "failed", "skipped"],
470        longrepr: None
471        | ExceptionInfo[BaseException]
472        | tuple[str, int, str]
473        | str
474        | TerminalRepr,
475        result: list[Item | Collector] | None,
476        sections: Iterable[tuple[str, str]] = (),
477        **extra,
478    ) -> None:
479        #: Normalized collection nodeid.
480        self.nodeid = nodeid
481
482        #: Test outcome, always one of "passed", "failed", "skipped".
483        self.outcome = outcome
484
485        #: None or a failure representation.
486        self.longrepr = longrepr
487
488        #: The collected items and collection nodes.
489        self.result = result or []
490
491        #: Tuples of str ``(heading, content)`` with extra information
492        #: for the test report. Used by pytest to add text captured
493        #: from ``stdout``, ``stderr``, and intercepted logging events. May
494        #: be used by other plugins to add arbitrary information to reports.
495        self.sections = list(sections)
496
497        self.__dict__.update(extra)
498
499    @property
500    def location(  # type:ignore[override]
501        self,
502    ) -> tuple[str, int | None, str] | None:
503        return (self.fspath, None, self.fspath)
504
505    def __repr__(self) -> str:
506        return f"<CollectReport {self.nodeid!r} lenresult={len(self.result)} outcome={self.outcome!r}>"