Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Create MypyConfigStash #174

Merged
merged 1 commit into from
Aug 24, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 29 additions & 11 deletions src/pytest_mypy.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,24 @@
import pytest


@dataclass(frozen=True) # compat python < 3.10 (kw_only=True)
class MypyConfigStash:
"""Plugin data stored in the pytest.Config stash."""

mypy_results_path: Path

@classmethod
def from_serialized(cls, serialized):
return cls(mypy_results_path=Path(serialized))

def serialized(self):
return str(self.mypy_results_path)


mypy_argv = []
nodeid_name = "mypy"
stash_keys = {
"mypy_results_path": pytest.StashKey[Path](),
stash_key = {
"config": pytest.StashKey[MypyConfigStash](),
}
terminal_summary_title = "mypy"

Expand Down Expand Up @@ -83,7 +97,9 @@ def pytest_configure(config):
# Subsequent MypyItems will see the file exists,
# and they will read the parsed results.
with NamedTemporaryFile(delete=True) as tmp_f:
config.stash[stash_keys["mypy_results_path"]] = Path(tmp_f.name)
config.stash[stash_key["config"]] = MypyConfigStash(
mypy_results_path=Path(tmp_f.name),
)

# If xdist is enabled, then the results path should be exposed to
# the workers so that they know where to read parsed results from.
Expand All @@ -92,8 +108,8 @@ def pytest_configure(config):
class _MypyXdistPlugin:
def pytest_configure_node(self, node): # xdist hook
"""Pass the mypy results path to workers."""
_get_xdist_workerinput(node)["_mypy_results_path"] = str(
node.config.stash[stash_keys["mypy_results_path"]]
_get_xdist_workerinput(node)["mypy_config_stash_serialized"] = (
node.config.stash[stash_key["config"]].serialized()
)

config.pluginmanager.register(_MypyXdistPlugin())
Expand Down Expand Up @@ -262,11 +278,13 @@ def from_mypy(
@classmethod
def from_session(cls, session) -> "MypyResults":
"""Load (or generate) cached mypy results for a pytest session."""
mypy_results_path = Path(
session.config.stash[stash_keys["mypy_results_path"]]
if _is_xdist_controller(session.config)
else _get_xdist_workerinput(session.config)["_mypy_results_path"]
)
if _is_xdist_controller(session.config):
mypy_config_stash = session.config.stash[stash_key["config"]]
else:
mypy_config_stash = MypyConfigStash.from_serialized(
_get_xdist_workerinput(session.config)["mypy_config_stash_serialized"]
)
mypy_results_path = mypy_config_stash.mypy_results_path
with FileLock(str(mypy_results_path) + ".lock"):
try:
with open(mypy_results_path, mode="r") as results_f:
Expand Down Expand Up @@ -299,7 +317,7 @@ def pytest_terminal_summary(terminalreporter, config):
"""Report stderr and unrecognized lines from stdout."""
if not _is_xdist_controller(config):
return
mypy_results_path = config.stash[stash_keys["mypy_results_path"]]
mypy_results_path = config.stash[stash_key["config"]].mypy_results_path
try:
with open(mypy_results_path, mode="r") as results_f:
results = MypyResults.load(results_f)
Expand Down
5 changes: 2 additions & 3 deletions tests/test_pytest_mypy.py
Original file line number Diff line number Diff line change
Expand Up @@ -521,13 +521,12 @@ def test_mypy_no_output(testdir, xdist_args):
@pytest.hookimpl(hookwrapper=True)
def pytest_terminal_summary(config):
pytest_mypy = config.pluginmanager.getplugin("mypy")
stash_key = pytest_mypy.stash_keys["mypy_results_path"]
try:
mypy_results_path = config.stash[stash_key]
mypy_config_stash = config.stash[pytest_mypy.stash_key["config"]]
except KeyError:
# xdist worker
return
with open(mypy_results_path, mode="w") as results_f:
with open(mypy_config_stash.mypy_results_path, mode="w") as results_f:
pytest_mypy.MypyResults(
opts=[],
stdout="",
Expand Down
Loading