p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #6836 from bluetech/store 

Add a typing-compatible mechanism for ad-hoc attributes on various objects
This commit is contained in:
Bruno Oliveira 2020-03-01 09:30:10 -03:00 committed by GitHub
parent 5fc80d8bc3 d636fcd557
commit 92767fec51
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 261 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/_pytest/config/__init__.py
View File

@ -42,6 +42,7 @@ from _pytest.compat import TYPE_CHECKING
from _pytest.outcomes import fail
from _pytest.outcomes import Skipped
from _pytest.pathlib import Path
from _pytest.store import Store
from _pytest.warning_types import PytestConfigWarning
if TYPE_CHECKING:
@ -791,6 +792,9 @@ class Config:
self._override_ini = () # type: Sequence[str]
self._opt2dest = {} # type: Dict[str, str]
self._cleanup = [] # type: List[Callable[[], None]]
# A place where plugins can store information on the config for their
# own use. Currently only intended for internal plugins.
self._store = Store()
self.pluginmanager.register(self, "pytestconfig")
self._configured = False
self.hook.pytest_addoption.call_historic(

15
src/_pytest/faulthandler.py
View File

@ -1,8 +1,13 @@
import io
import os
import sys
from typing import TextIO
import pytest
from _pytest.store import StoreKey
fault_handler_stderr_key = StoreKey[TextIO]()
def pytest_addoption(parser):
@ -46,8 +51,8 @@ class FaultHandlerHooks:
import faulthandler
stderr_fd_copy = os.dup(self._get_stderr_fileno())
config.fault_handler_stderr = os.fdopen(stderr_fd_copy, "w")
faulthandler.enable(file=config.fault_handler_stderr)
config._store[fault_handler_stderr_key] = open(stderr_fd_copy, "w")
faulthandler.enable(file=config._store[fault_handler_stderr_key])
def pytest_unconfigure(self, config):
import faulthandler
@ -57,8 +62,8 @@ class FaultHandlerHooks:
# re-enable the faulthandler, attaching it to the default sys.stderr
# so we can see crashes after pytest has finished, usually during
# garbage collection during interpreter shutdown
config.fault_handler_stderr.close()
del config.fault_handler_stderr
config._store[fault_handler_stderr_key].close()
del config._store[fault_handler_stderr_key]
faulthandler.enable(file=self._get_stderr_fileno())
@staticmethod
@ -78,7 +83,7 @@ class FaultHandlerHooks:
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_protocol(self, item):
timeout = self.get_timeout_config_value(item.config)
stderr = item.config.fault_handler_stderr
stderr = item.config._store[fault_handler_stderr_key]
if timeout > 0 and stderr is not None:
import faulthandler

18
src/_pytest/junitxml.py
View File

@ -22,9 +22,13 @@ import pytest
from _pytest import deprecated
from _pytest import nodes
from _pytest.config import filename_arg
from _pytest.store import StoreKey
from _pytest.warnings import _issue_warning_captured
xml_key = StoreKey["LogXML"]()
class Junit(py.xml.Namespace):
pass
@ -260,7 +264,7 @@ def _warn_incompatibility_with_xunit2(request, fixture_name):
"""Emits a PytestWarning about the given fixture being incompatible with newer xunit revisions"""
from _pytest.warning_types import PytestWarning
xml = getattr(request.config, "_xml", None)
xml = request.config._store.get(xml_key, None)
if xml is not None and xml.family not in ("xunit1", "legacy"):
request.node.warn(
PytestWarning(
@ -312,7 +316,7 @@ def record_xml_attribute(request):
attr_func = add_attr_noop
xml = getattr(request.config, "_xml", None)
xml = request.config._store.get(xml_key, None)
if xml is not None:
node_reporter = xml.node_reporter(request.node.nodeid)
attr_func = node_reporter.add_attribute
@ -353,7 +357,7 @@ def record_testsuite_property(request):
__tracebackhide__ = True
_check_record_param_type("name", name)
xml = getattr(request.config, "_xml", None)
xml = request.config._store.get(xml_key, None)
if xml is not None:
record_func = xml.add_global_property # noqa
return record_func
@ -412,7 +416,7 @@ def pytest_configure(config):
if not junit_family:
_issue_warning_captured(deprecated.JUNIT_XML_DEFAULT_FAMILY, config.hook, 2)
junit_family = "xunit1"
config._xml = LogXML(
config._store[xml_key] = LogXML(
xmlpath,
config.option.junitprefix,
config.getini("junit_suite_name"),
@ -421,13 +425,13 @@ def pytest_configure(config):
junit_family,
config.getini("junit_log_passing_tests"),
)
config.pluginmanager.register(config._xml)
config.pluginmanager.register(config._store[xml_key])
def pytest_unconfigure(config):
xml = getattr(config, "_xml", None)
xml = config._store.get(xml_key, None)
if xml:
del config._xml
del config._store[xml_key]
config.pluginmanager.unregister(xml)

11
src/_pytest/mark/__init__.py
View File

@ -1,4 +1,6 @@
""" generic mechanism for marking and selecting python functions. """
from typing import Optional
from .legacy import matchkeyword
from .legacy import matchmark
from .structures import EMPTY_PARAMETERSET_OPTION
@ -8,12 +10,17 @@ from .structures import MARK_GEN
from .structures import MarkDecorator
from .structures import MarkGenerator
from .structures import ParameterSet
from _pytest.config import Config
from _pytest.config import hookimpl
from _pytest.config import UsageError
from _pytest.store import StoreKey
__all__ = ["Mark", "MarkDecorator", "MarkGenerator", "get_empty_parameterset_mark"]
old_mark_config_key = StoreKey[Optional[Config]]()
def param(*values, **kw):
"""Specify a parameter in `pytest.mark.parametrize`_ calls or
:ref:`parametrized fixtures <fixture-parametrize-marks>`.
@ -145,7 +152,7 @@ def pytest_collection_modifyitems(items, config):
def pytest_configure(config):
config._old_mark_config = MARK_GEN._config
config._store[old_mark_config_key] = MARK_GEN._config
MARK_GEN._config = config
empty_parameterset = config.getini(EMPTY_PARAMETERSET_OPTION)
@ -158,4 +165,4 @@ def pytest_configure(config):
def pytest_unconfigure(config):
MARK_GEN._config = getattr(config, "_old_mark_config", None)
MARK_GEN._config = config._store.get(old_mark_config_key, None)

5
src/_pytest/nodes.py
View File

@ -29,6 +29,7 @@ from _pytest.mark.structures import MarkDecorator
from _pytest.mark.structures import NodeKeywords
from _pytest.outcomes import fail
from _pytest.outcomes import Failed
from _pytest.store import Store
if TYPE_CHECKING:
# Imported here due to circular import.
@ -146,6 +147,10 @@ class Node(metaclass=NodeMeta):
if self.name != "()":
self._nodeid += "::" + self.name
# A place where plugins can store information on the node for their
# own use. Currently only intended for internal plugins.
self._store = Store()
@classmethod
def from_parent(cls, parent: "Node", **kw):
"""

20
src/_pytest/pastebin.py
View File

@ -1,7 +1,12 @@
""" submit failure or test session information to a pastebin service. """
import tempfile
from typing import IO
import pytest
from _pytest.store import StoreKey
pastebinfile_key = StoreKey[IO[bytes]]()
def pytest_addoption(parser):
@ -26,25 +31,26 @@ def pytest_configure(config):
# when using pytest-xdist, for example
if tr is not None:
# pastebin file will be utf-8 encoded binary file
config._pastebinfile = tempfile.TemporaryFile("w+b")
config._store[pastebinfile_key] = tempfile.TemporaryFile("w+b")
oldwrite = tr._tw.write
def tee_write(s, **kwargs):
oldwrite(s, **kwargs)
if isinstance(s, str):
s = s.encode("utf-8")
config._pastebinfile.write(s)
config._store[pastebinfile_key].write(s)
tr._tw.write = tee_write
def pytest_unconfigure(config):
if hasattr(config, "_pastebinfile"):
if pastebinfile_key in config._store:
pastebinfile = config._store[pastebinfile_key]
# get terminal contents and delete file
config._pastebinfile.seek(0)
sessionlog = config._pastebinfile.read()
config._pastebinfile.close()
del config._pastebinfile
pastebinfile.seek(0)
sessionlog = pastebinfile.read()
pastebinfile.close()
del config._store[pastebinfile_key]
# undo our patching in the terminal reporter
tr = config.pluginmanager.getplugin("terminalreporter")
del tr._tw.__dict__["write"]

13
src/_pytest/resultlog.py
View File

@ -5,6 +5,11 @@ import os
import py
from _pytest.store import StoreKey
resultlog_key = StoreKey["ResultLog"]()
def pytest_addoption(parser):
group = parser.getgroup("terminal reporting", "resultlog plugin options")
@ -26,8 +31,8 @@ def pytest_configure(config):
if not os.path.isdir(dirname):
os.makedirs(dirname)
logfile = open(resultlog, "w", 1) # line buffered
config._resultlog = ResultLog(config, logfile)
config.pluginmanager.register(config._resultlog)
config._store[resultlog_key] = ResultLog(config, logfile)
config.pluginmanager.register(config._store[resultlog_key])
from _pytest.deprecated import RESULT_LOG
from _pytest.warnings import _issue_warning_captured
@ -36,10 +41,10 @@ def pytest_configure(config):
def pytest_unconfigure(config):
resultlog = getattr(config, "_resultlog", None)
resultlog = config._store.get(resultlog_key, None)
if resultlog:
resultlog.logfile.close()
del config._resultlog
del config._store[resultlog_key]
config.pluginmanager.unregister(resultlog)

34
src/_pytest/skipping.py
View File

@ -4,6 +4,12 @@ from _pytest.mark.evaluate import MarkEvaluator
from _pytest.outcomes import fail
from _pytest.outcomes import skip
from _pytest.outcomes import xfail
from _pytest.store import StoreKey
skipped_by_mark_key = StoreKey[bool]()
evalxfail_key = StoreKey[MarkEvaluator]()
unexpectedsuccess_key = StoreKey[str]()
def pytest_addoption(parser):
@ -68,14 +74,14 @@ def pytest_configure(config):
@hookimpl(tryfirst=True)
def pytest_runtest_setup(item):
# Check if skip or skipif are specified as pytest marks
item._skipped_by_mark = False
item._store[skipped_by_mark_key] = False
eval_skipif = MarkEvaluator(item, "skipif")
if eval_skipif.istrue():
item._skipped_by_mark = True
item._store[skipped_by_mark_key] = True
skip(eval_skipif.getexplanation())
for skip_info in item.iter_markers(name="skip"):
item._skipped_by_mark = True
item._store[skipped_by_mark_key] = True
if "reason" in skip_info.kwargs:
skip(skip_info.kwargs["reason"])
elif skip_info.args:
@ -83,7 +89,7 @@ def pytest_runtest_setup(item):
else:
skip("unconditional skip")
item._evalxfail = MarkEvaluator(item, "xfail")
item._store[evalxfail_key] = MarkEvaluator(item, "xfail")
check_xfail_no_run(item)
@ -99,7 +105,7 @@ def pytest_pyfunc_call(pyfuncitem):
def check_xfail_no_run(item):
"""check xfail(run=False)"""
if not item.config.option.runxfail:
evalxfail = item._evalxfail
evalxfail = item._store[evalxfail_key]
if evalxfail.istrue():
if not evalxfail.get("run", True):
xfail("[NOTRUN] " + evalxfail.getexplanation())
@ -107,12 +113,12 @@ def check_xfail_no_run(item):
def check_strict_xfail(pyfuncitem):
"""check xfail(strict=True) for the given PASSING test"""
evalxfail = pyfuncitem._evalxfail
evalxfail = pyfuncitem._store[evalxfail_key]
if evalxfail.istrue():
strict_default = pyfuncitem.config.getini("xfail_strict")
is_strict_xfail = evalxfail.get("strict", strict_default)
if is_strict_xfail:
del pyfuncitem._evalxfail
del pyfuncitem._store[evalxfail_key]
explanation = evalxfail.getexplanation()
fail("[XPASS(strict)] " + explanation, pytrace=False)
@ -121,12 +127,12 @@ def check_strict_xfail(pyfuncitem):
def pytest_runtest_makereport(item, call):
outcome = yield
rep = outcome.get_result()
evalxfail = getattr(item, "_evalxfail", None)
# unittest special case, see setting of _unexpectedsuccess
if hasattr(item, "_unexpectedsuccess") and rep.when == "call":
if item._unexpectedsuccess:
rep.longrepr = "Unexpected success: {}".format(item._unexpectedsuccess)
evalxfail = item._store.get(evalxfail_key, None)
# unittest special case, see setting of unexpectedsuccess_key
if unexpectedsuccess_key in item._store and rep.when == "call":
reason = item._store[unexpectedsuccess_key]
if reason:
rep.longrepr = "Unexpected success: {}".format(reason)
else:
rep.longrepr = "Unexpected success"
rep.outcome = "failed"
@ -154,7 +160,7 @@ def pytest_runtest_makereport(item, call):
rep.outcome = "passed"
rep.wasxfail = explanation
elif (
getattr(item, "_skipped_by_mark", False)
item._store.get(skipped_by_mark_key, True)
and rep.skipped
and type(rep.longrepr) is tuple
):

116
src/_pytest/store.py Normal file
View File

@ -0,0 +1,116 @@
from typing import Any
from typing import cast
from typing import Dict
from typing import Generic
from typing import TypeVar
from typing import Union
__all__ = ["Store", "StoreKey"]
T = TypeVar("T")
D = TypeVar("D")
class StoreKey(Generic[T]):
"""StoreKey is an object used as a key to a Store.
A StoreKey is associated with the type T of the value of the key.
A StoreKey is unique and cannot conflict with another key.
"""
__slots__ = ()
class Store:
"""Store is a type-safe heterogenous mutable mapping that
allows keys and value types to be defined separately from
where it is defined.
Usually you will be given an object which has a ``Store``:
.. code-block:: python
store: Store = some_object.store
If a module wants to store data in this Store, it creates StoreKeys
for its keys (at the module level):
.. code-block:: python
some_str_key = StoreKey[str]()
some_bool_key = StoreKey[bool]()
To store information:
.. code-block:: python
# Value type must match the key.
store[some_str_key] = "value"
store[some_bool_key] = True
To retrieve the information:
.. code-block:: python
# The static type of some_str is str.
some_str = store[some_str_key]
# The static type of some_bool is bool.
some_bool = store[some_bool_key]
Why use this?
-------------
Problem: module Internal defines an object. Module External, which
module Internal doesn't know about, receives the object and wants to
attach information to it, to be retrieved later given the object.
Bad solution 1: Module External assigns private attributes directly on
the object. This doesn't work well because the type checker doesn't
know about these attributes and it complains about undefined attributes.
Bad solution 2: module Internal adds a ``Dict[str, Any]`` attribute to
the object. Module External stores its data in private keys of this dict.
This doesn't work well because retrieved values are untyped.
Good solution: module Internal adds a ``Store`` to the object. Module
External mints StoreKeys for its own keys. Module External stores and
retrieves its data using its keys.
"""
__slots__ = ("_store",)
def __init__(self) -> None:
self._store = {} # type: Dict[StoreKey[Any], object]
def __setitem__(self, key: StoreKey[T], value: T) -> None:
"""Set a value for key."""
self._store[key] = value
def __getitem__(self, key: StoreKey[T]) -> T:
"""Get the value for key.
Raises KeyError if the key wasn't set before.
"""
return cast(T, self._store[key])
def get(self, key: StoreKey[T], default: D) -> Union[T, D]:
"""Get the value for key, or return default if the key wasn't set
before."""
try:
return self[key]
except KeyError:
return default
def __delitem__(self, key: StoreKey[T]) -> None:
"""Delete the value for key.
Raises KeyError if the key wasn't set before.
"""
del self._store[key]
def __contains__(self, key: StoreKey[T]) -> bool:
"""Returns whether key was set."""
return key in self._store

6
src/_pytest/unittest.py
View File

@ -14,6 +14,8 @@ from _pytest.outcomes import xfail
from _pytest.python import Class
from _pytest.python import Function
from _pytest.runner import CallInfo
from _pytest.skipping import skipped_by_mark_key
from _pytest.skipping import unexpectedsuccess_key
def pytest_pycollect_makeitem(collector, name, obj):
@ -174,7 +176,7 @@ class TestCaseFunction(Function):
try:
skip(reason)
except skip.Exception:
self._skipped_by_mark = True
self._store[skipped_by_mark_key] = True
self._addexcinfo(sys.exc_info())
def addExpectedFailure(self, testcase, rawexcinfo, reason=""):
@ -184,7 +186,7 @@ class TestCaseFunction(Function):
self._addexcinfo(sys.exc_info())
def addUnexpectedSuccess(self, testcase, reason=""):
self._unexpectedsuccess = reason
self._store[unexpectedsuccess_key] = reason
def addSuccess(self, testcase):
pass

2
testing/test_junitxml.py
View File

@ -10,6 +10,7 @@ import pytest
from _pytest.junitxml import LogXML
from _pytest.pathlib import Path
from _pytest.reports import BaseReport
from _pytest.store import Store
@pytest.fixture(scope="session")
@ -865,6 +866,7 @@ def test_dont_configure_on_slaves(tmpdir):
def __init__(self):
self.pluginmanager = self
self.option = self
self._store = Store()
def getini(self, name):
return "pytest"

11
testing/test_resultlog.py
View File

@ -6,6 +6,7 @@ import pytest
from _pytest.resultlog import pytest_configure
from _pytest.resultlog import pytest_unconfigure
from _pytest.resultlog import ResultLog
from _pytest.resultlog import resultlog_key
pytestmark = pytest.mark.filterwarnings("ignore:--result-log is deprecated")
@ -179,17 +180,17 @@ def test_makedir_for_resultlog(testdir, LineMatcher):
def test_no_resultlog_on_slaves(testdir):
config = testdir.parseconfig("-p", "resultlog", "--resultlog=resultlog")
assert not hasattr(config, "_resultlog")
assert resultlog_key not in config._store
pytest_configure(config)
assert hasattr(config, "_resultlog")
assert resultlog_key in config._store
pytest_unconfigure(config)
assert not hasattr(config, "_resultlog")
assert resultlog_key not in config._store
config.slaveinput = {}
pytest_configure(config)
assert not hasattr(config, "_resultlog")
assert resultlog_key not in config._store
pytest_unconfigure(config)
assert not hasattr(config, "_resultlog")
assert resultlog_key not in config._store
def test_failure_issue380(testdir):

52
testing/test_store.py Normal file
View File

@ -0,0 +1,52 @@
import pytest
from _pytest.store import Store
from _pytest.store import StoreKey
def test_store() -> None:
store = Store()
key1 = StoreKey[str]()
key2 = StoreKey[int]()
# Basic functionality - single key.
assert key1 not in store
store[key1] = "hello"
assert key1 in store
assert store[key1] == "hello"
assert store.get(key1, None) == "hello"
store[key1] = "world"
assert store[key1] == "world"
# Has correct type (no mypy error).
store[key1] + "string"
# No interaction with another key.
assert key2 not in store
assert store.get(key2, None) is None
with pytest.raises(KeyError):
store[key2]
with pytest.raises(KeyError):
del store[key2]
store[key2] = 1
assert store[key2] == 1
# Has correct type (no mypy error).
store[key2] + 20
del store[key1]
with pytest.raises(KeyError):
del store[key1]
with pytest.raises(KeyError):
store[key1]
# Can't accidentally add attributes to store object itself.
with pytest.raises(AttributeError):
store.foo = "nope" # type: ignore[attr-defined] # noqa: F821
# No interaction with anoter store.
store2 = Store()
key3 = StoreKey[int]()
assert key2 not in store2
store2[key2] = 100
store2[key3] = 200
assert store2[key2] + store2[key3] == 300
assert store[key2] == 1
assert key3 not in store