Type annotate misc functions
This commit is contained in:
Ran Benita
parent
d95132178c
commit
e68a26199c
|
|
@ -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():
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Reference in New Issue