diff --git a/src/_pytest/cacheprovider.py b/src/_pytest/cacheprovider.py index c95f17152..bb08c5a6e 100755 --- a/src/_pytest/cacheprovider.py +++ b/src/_pytest/cacheprovider.py @@ -8,6 +8,7 @@ import json import os from typing import Dict from typing import Generator +from typing import Iterable from typing import List from typing import Optional from typing import Set @@ -27,10 +28,12 @@ from _pytest.compat import order_preserving_dict from _pytest.config import Config from _pytest.config import ExitCode from _pytest.config.argparsing import Parser +from _pytest.fixtures import FixtureRequest from _pytest.main import Session from _pytest.python import Module from _pytest.reports import TestReport + README_CONTENT = """\ # pytest cache directory # @@ -52,8 +55,8 @@ Signature: 8a477f597d28d172789f06886806bc55 @attr.s class Cache: - _cachedir = attr.ib(repr=False) - _config = attr.ib(repr=False) + _cachedir = attr.ib(type=Path, repr=False) + _config = attr.ib(type=Config, repr=False) # sub-directory under cache-dir for directories created by "makedir" _CACHE_PREFIX_DIRS = "d" @@ -62,14 +65,14 @@ class Cache: _CACHE_PREFIX_VALUES = "v" @classmethod - def for_config(cls, config): + def for_config(cls, config: Config) -> "Cache": cachedir = cls.cache_dir_from_config(config) if config.getoption("cacheclear") and cachedir.is_dir(): cls.clear_cache(cachedir) return cls(cachedir, config) @classmethod - def clear_cache(cls, cachedir: Path): + def clear_cache(cls, cachedir: Path) -> None: """Clears the sub-directories used to hold cached directories and values.""" for prefix in (cls._CACHE_PREFIX_DIRS, cls._CACHE_PREFIX_VALUES): d = cachedir / prefix @@ -77,10 +80,10 @@ class Cache: rm_rf(d) @staticmethod - def cache_dir_from_config(config): + def cache_dir_from_config(config: Config): return resolve_from_str(config.getini("cache_dir"), config.rootdir) - def warn(self, fmt, **args): + def warn(self, fmt: str, **args: object) -> None: import warnings from _pytest.warning_types import PytestCacheWarning @@ -90,7 +93,7 @@ class Cache: stacklevel=3, ) - def makedir(self, name): + def makedir(self, name: str) -> py.path.local: """ return a directory path object with the given name. If the directory does not yet exist, it will be created. You can use it to manage files likes e. g. store/retrieve database @@ -100,14 +103,14 @@ class Cache: Make sure the name contains your plugin or application identifiers to prevent clashes with other cache users. """ - name = Path(name) - if len(name.parts) > 1: + path = Path(name) + if len(path.parts) > 1: raise ValueError("name is not allowed to contain path separators") - res = self._cachedir.joinpath(self._CACHE_PREFIX_DIRS, name) + res = self._cachedir.joinpath(self._CACHE_PREFIX_DIRS, path) res.mkdir(exist_ok=True, parents=True) return py.path.local(res) - def _getvaluepath(self, key): + def _getvaluepath(self, key: str) -> Path: return self._cachedir.joinpath(self._CACHE_PREFIX_VALUES, Path(key)) def get(self, key, default): @@ -128,7 +131,7 @@ class Cache: except (ValueError, OSError): return default - def set(self, key, value): + def set(self, key, value) -> None: """ save value for the given key. :param key: must be a ``/`` separated value. Usually the first @@ -158,7 +161,7 @@ class Cache: with f: f.write(data) - def _ensure_supporting_files(self): + def _ensure_supporting_files(self) -> None: """Create supporting files in the cache dir that are not really part of the cache.""" readme_path = self._cachedir / "README.md" readme_path.write_text(README_CONTENT) @@ -172,12 +175,12 @@ class Cache: class LFPluginCollWrapper: - def __init__(self, lfplugin: "LFPlugin"): + def __init__(self, lfplugin: "LFPlugin") -> None: self.lfplugin = lfplugin self._collected_at_least_one_failure = False @pytest.hookimpl(hookwrapper=True) - def pytest_make_collect_report(self, collector) -> Generator: + def pytest_make_collect_report(self, collector: nodes.Collector) -> Generator: if isinstance(collector, Session): out = yield res = out.get_result() # type: CollectReport @@ -220,11 +223,13 @@ class LFPluginCollWrapper: class LFPluginCollSkipfiles: - def __init__(self, lfplugin: "LFPlugin"): + def __init__(self, lfplugin: "LFPlugin") -> None: self.lfplugin = lfplugin @pytest.hookimpl - def pytest_make_collect_report(self, collector) -> Optional[CollectReport]: + def pytest_make_collect_report( + self, collector: nodes.Collector + ) -> Optional[CollectReport]: if isinstance(collector, Module): if Path(str(collector.fspath)) not in self.lfplugin._last_failed_paths: self.lfplugin._skipped_files += 1 @@ -262,9 +267,10 @@ class LFPlugin: result = {rootpath / nodeid.split("::")[0] for nodeid in self.lastfailed} return {x for x in result if x.exists()} - def pytest_report_collectionfinish(self): + def pytest_report_collectionfinish(self) -> Optional[str]: if self.active and self.config.getoption("verbose") >= 0: return "run-last-failure: %s" % self._report_status + return None def pytest_runtest_logreport(self, report: TestReport) -> None: if (report.when == "call" and report.passed) or report.skipped: @@ -347,9 +353,10 @@ class LFPlugin: class NFPlugin: """ Plugin which implements the --nf (run new-first) option """ - def __init__(self, config): + def __init__(self, config: Config) -> None: self.config = config self.active = config.option.newfirst + assert config.cache is not None self.cached_nodeids = set(config.cache.get("cache/nodeids", [])) @pytest.hookimpl(hookwrapper=True, tryfirst=True) @@ -374,7 +381,7 @@ class NFPlugin: else: self.cached_nodeids.update(item.nodeid for item in items) - def _get_increasing_order(self, items): + def _get_increasing_order(self, items: Iterable[nodes.Item]) -> List[nodes.Item]: return sorted(items, key=lambda item: item.fspath.mtime(), reverse=True) def pytest_sessionfinish(self) -> None: @@ -384,6 +391,8 @@ class NFPlugin: if config.getoption("collectonly"): return + + assert config.cache is not None config.cache.set("cache/nodeids", sorted(self.cached_nodeids)) @@ -462,7 +471,7 @@ def pytest_configure(config: Config) -> None: @pytest.fixture -def cache(request): +def cache(request: FixtureRequest) -> Cache: """ Return a cache object that can persist state between testing sessions. @@ -474,12 +483,14 @@ def cache(request): Values can be any object handled by the json stdlib module. """ + assert request.config.cache is not None return request.config.cache -def pytest_report_header(config): +def pytest_report_header(config: Config) -> Optional[str]: """Display cachedir with --cache-show and if non-default.""" if config.option.verbose > 0 or config.getini("cache_dir") != ".pytest_cache": + assert config.cache is not None cachedir = config.cache._cachedir # TODO: evaluate generating upward relative paths # starting with .., ../.. if sensible @@ -489,11 +500,14 @@ def pytest_report_header(config): except ValueError: displaypath = cachedir return "cachedir: {}".format(displaypath) + return None -def cacheshow(config, session): +def cacheshow(config: Config, session: Session) -> int: from pprint import pformat + assert config.cache is not None + tw = TerminalWriter() tw.line("cachedir: " + str(config.cache._cachedir)) if not config.cache._cachedir.is_dir(): diff --git a/src/_pytest/fixtures.py b/src/_pytest/fixtures.py index 4cd9a20ef..b45392ba5 100644 --- a/src/_pytest/fixtures.py +++ b/src/_pytest/fixtures.py @@ -57,6 +57,7 @@ if TYPE_CHECKING: from _pytest import nodes from _pytest.main import Session from _pytest.python import Metafunc + from _pytest.python import CallSpec2 _Scope = Literal["session", "package", "module", "class", "function"] @@ -217,10 +218,11 @@ def get_parametrized_fixture_keys(item, scopenum): the specified scope. """ assert scopenum < scopenum_function # function try: - cs = item.callspec + callspec = item.callspec # type: ignore[attr-defined] # noqa: F821 except AttributeError: pass else: + cs = callspec # type: CallSpec2 # cs.indices.items() is random order of argnames. Need to # sort this so that different calls to # get_parametrized_fixture_keys will be deterministic. @@ -434,9 +436,9 @@ class FixtureRequest: return fixturedefs[index] @property - def config(self): + def config(self) -> Config: """ the pytest config object associated with this request. """ - return self._pyfuncitem.config + return self._pyfuncitem.config # type: ignore[no-any-return] # noqa: F723 @scopeproperty() def function(self): @@ -1464,7 +1466,7 @@ class FixtureManager: else: continue # will raise FixtureLookupError at setup time - def pytest_collection_modifyitems(self, items): + def pytest_collection_modifyitems(self, items: "List[nodes.Item]") -> None: # separate parametrized setups items[:] = reorder_items(items) diff --git a/src/_pytest/hookspec.py b/src/_pytest/hookspec.py index ccdb0bde9..c5d5bdedd 100644 --- a/src/_pytest/hookspec.py +++ b/src/_pytest/hookspec.py @@ -223,7 +223,9 @@ def pytest_collection(session: "Session") -> Optional[Any]: """ -def pytest_collection_modifyitems(session: "Session", config: "Config", items): +def pytest_collection_modifyitems( + session: "Session", config: "Config", items: List["Item"] +) -> None: """ called after collection has been performed, may filter or re-order the items in-place. diff --git a/src/_pytest/main.py b/src/_pytest/main.py index d891335a7..a80097f5a 100644 --- a/src/_pytest/main.py +++ b/src/_pytest/main.py @@ -333,7 +333,7 @@ def pytest_ignore_collect( return None -def pytest_collection_modifyitems(items, config: Config) -> None: +def pytest_collection_modifyitems(items: List[nodes.Item], config: Config) -> None: deselect_prefixes = tuple(config.getoption("deselect") or []) if not deselect_prefixes: return @@ -487,18 +487,18 @@ class Session(nodes.FSCollector): @overload def _perform_collect( self, args: Optional[Sequence[str]], genitems: "Literal[True]" - ) -> Sequence[nodes.Item]: + ) -> List[nodes.Item]: raise NotImplementedError() @overload # noqa: F811 def _perform_collect( # noqa: F811 self, args: Optional[Sequence[str]], genitems: bool - ) -> Sequence[Union[nodes.Item, nodes.Collector]]: + ) -> Union[List[Union[nodes.Item]], List[Union[nodes.Item, nodes.Collector]]]: raise NotImplementedError() def _perform_collect( # noqa: F811 self, args: Optional[Sequence[str]], genitems: bool - ) -> Sequence[Union[nodes.Item, nodes.Collector]]: + ) -> Union[List[Union[nodes.Item]], List[Union[nodes.Item, nodes.Collector]]]: if args is None: args = self.config.args self.trace("perform_collect", self, args) diff --git a/src/_pytest/mark/__init__.py b/src/_pytest/mark/__init__.py index 05afb7749..16e821aee 100644 --- a/src/_pytest/mark/__init__.py +++ b/src/_pytest/mark/__init__.py @@ -2,6 +2,7 @@ import typing import warnings from typing import AbstractSet +from typing import List from typing import Optional from typing import Union @@ -173,7 +174,7 @@ class KeywordMatcher: return False -def deselect_by_keyword(items, config: Config) -> None: +def deselect_by_keyword(items: "List[Item]", config: Config) -> None: keywordexpr = config.option.keyword.lstrip() if not keywordexpr: return @@ -229,7 +230,7 @@ class MarkMatcher: return name in self.own_mark_names -def deselect_by_mark(items, config: Config) -> None: +def deselect_by_mark(items: "List[Item]", config: Config) -> None: matchexpr = config.option.markexpr if not matchexpr: return @@ -254,7 +255,7 @@ def deselect_by_mark(items, config: Config) -> None: items[:] = remaining -def pytest_collection_modifyitems(items, config: Config) -> None: +def pytest_collection_modifyitems(items: "List[Item]", config: Config) -> None: deselect_by_keyword(items, config) deselect_by_mark(items, config) diff --git a/src/_pytest/pathlib.py b/src/_pytest/pathlib.py index 90a7460b0..6878965e0 100644 --- a/src/_pytest/pathlib.py +++ b/src/_pytest/pathlib.py @@ -348,7 +348,7 @@ def make_numbered_dir_with_cleanup( raise e -def resolve_from_str(input, root): +def resolve_from_str(input: str, root): assert not isinstance(input, Path), "would break on py2" root = Path(root) input = expanduser(input) diff --git a/src/_pytest/pytester.py b/src/_pytest/pytester.py index 12ad0591c..60df17b90 100644 --- a/src/_pytest/pytester.py +++ b/src/_pytest/pytester.py @@ -647,8 +647,8 @@ class Testdir: for basename, value in items: p = self.tmpdir.join(basename).new(ext=ext) p.dirpath().ensure_dir() - source = Source(value) - source = "\n".join(to_text(line) for line in source.lines) + source_ = Source(value) + source = "\n".join(to_text(line) for line in source_.lines) p.write(source.strip().encode(encoding), "wb") if ret is None: ret = p @@ -839,7 +839,7 @@ class Testdir: config.hook.pytest_sessionfinish(session=session, exitstatus=ExitCode.OK) return res - def genitems(self, colitems): + def genitems(self, colitems: List[Union[Item, Collector]]) -> List[Item]: """Generate all test items from a collection node. This recurses into the collection node and returns a list of all the @@ -847,7 +847,7 @@ class Testdir: """ session = colitems[0].session - result = [] + result = [] # type: List[Item] for colitem in colitems: result.extend(session.genitems(colitem)) return result diff --git a/src/_pytest/stepwise.py b/src/_pytest/stepwise.py index 1921245dc..85cbe2931 100644 --- a/src/_pytest/stepwise.py +++ b/src/_pytest/stepwise.py @@ -1,4 +1,8 @@ +from typing import List +from typing import Optional + import pytest +from _pytest import nodes from _pytest.config import Config from _pytest.config.argparsing import Parser from _pytest.main import Session @@ -28,20 +32,23 @@ def pytest_configure(config: Config) -> None: class StepwisePlugin: - def __init__(self, config): + def __init__(self, config: Config) -> None: self.config = config self.active = config.getvalue("stepwise") - self.session = None + self.session = None # type: Optional[Session] self.report_status = "" if self.active: + assert config.cache is not None self.lastfailed = config.cache.get("cache/stepwise", None) self.skip = config.getvalue("stepwise_skip") def pytest_sessionstart(self, session: Session) -> None: self.session = session - def pytest_collection_modifyitems(self, session, config, items): + def pytest_collection_modifyitems( + self, session: Session, config: Config, items: List[nodes.Item] + ) -> None: if not self.active: return if not self.lastfailed: @@ -89,6 +96,7 @@ class StepwisePlugin: else: # Mark test as the last failing and interrupt the test session. self.lastfailed = report.nodeid + assert self.session is not None self.session.shouldstop = ( "Test failed, continuing from this test next run." ) @@ -100,11 +108,13 @@ class StepwisePlugin: if report.nodeid == self.lastfailed: self.lastfailed = None - def pytest_report_collectionfinish(self): + def pytest_report_collectionfinish(self) -> Optional[str]: if self.active and self.config.getoption("verbose") >= 0 and self.report_status: return "stepwise: %s" % self.report_status + return None def pytest_sessionfinish(self, session: Session) -> None: + assert self.config.cache is not None if self.active: self.config.cache.set("cache/stepwise", self.lastfailed) else: