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()

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

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

400@final
401class CollectReport(BaseReport):
402    """Collection report object.
403
404    Reports can contain arbitrary extra attributes.
405    """
406
407    when = "collect"
408
409    def __init__(
410        self,
411        nodeid: str,
412        outcome: Literal["passed", "failed", "skipped"],
413        longrepr: None
414        | ExceptionInfo[BaseException]
415        | tuple[str, int, str]
416        | str
417        | TerminalRepr,
418        result: list[Item | Collector] | None,
419        sections: Iterable[tuple[str, str]] = (),
420        **extra,
421    ) -> None:
422        #: Normalized collection nodeid.
423        self.nodeid = nodeid
424
425        #: Test outcome, always one of "passed", "failed", "skipped".
426        self.outcome = outcome
427
428        #: None or a failure representation.
429        self.longrepr = longrepr
430
431        #: The collected items and collection nodes.
432        self.result = result or []
433
434        #: Tuples of str ``(heading, content)`` with extra information
435        #: for the test report. Used by pytest to add text captured
436        #: from ``stdout``, ``stderr``, and intercepted logging events. May
437        #: be used by other plugins to add arbitrary information to reports.
438        self.sections = list(sections)
439
440        self.__dict__.update(extra)
441
442    @property
443    def location(  # type:ignore[override]
444        self,
445    ) -> tuple[str, int | None, str] | None:
446        return (self.fspath, None, self.fspath)
447
448    def __repr__(self) -> str:
449        return f"<CollectReport {self.nodeid!r} lenresult={len(self.result)} outcome={self.outcome!r}>"