floraachy
/
pytest2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge 5e6ce8f433 into b31db4809b

This commit is contained in:
Yusuke Kadowaki 2022-12-18 23:36:31 +00:00 committed by GitHub
parent b31db4809b 5e6ce8f433
commit 72686992a0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 93 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/_pytest/pathlib.py
View File

@ -1,4 +1,3 @@
import atexit
import contextlib
import fnmatch
import importlib.util
@ -244,7 +243,7 @@ def create_cleanup_lock(p: Path) -> Path:
return lock_path
def register_cleanup_lock_removal(lock_path: Path, register=atexit.register):
def register_cleanup_lock_removal(lock_path: Path, register):
"""Register a cleanup function for removing a lock, by default on atexit."""
pid = os.getpid()
@ -362,6 +361,7 @@ def make_numbered_dir_with_cleanup(
keep: int,
lock_timeout: float,
mode: int,
register,
) -> Path:
"""Create a numbered dir with a cleanup lock and remove old ones."""
e = None
@ -371,13 +371,13 @@ def make_numbered_dir_with_cleanup(
# Only lock the current dir when keep is not 0
if keep != 0:
lock_path = create_cleanup_lock(p)
register_cleanup_lock_removal(lock_path)
register_cleanup_lock_removal(lock_path, register=register)
except Exception as exc:
e = exc
else:
consider_lock_dead_if_created_before = p.stat().st_mtime - lock_timeout
# Register a cleanup for program exit
atexit.register(
register(
cleanup_numbered_dir,
root,
prefix,

33
src/_pytest/tmpdir.py
View File

@ -3,6 +3,7 @@ import os
import re
import sys
import tempfile
from contextlib import ExitStack
from pathlib import Path
from shutil import rmtree
from typing import Dict
@ -77,6 +78,7 @@ class TempPathFactory:
self._retention_count = retention_count
self._retention_policy = retention_policy
self._basetemp = basetemp
self._exit_stack = ExitStack()
@classmethod
def from_config(
@ -196,6 +198,7 @@ class TempPathFactory:
keep=keep,
lock_timeout=LOCK_TIMEOUT,
mode=0o700,
register=self._exit_stack.callback,
)
assert basetemp is not None, basetemp
self._basetemp = basetemp
@ -303,19 +306,23 @@ def pytest_sessionfinish(session, exitstatus: Union[int, ExitCode]):
the policy is "failed", and the basetemp is not specified by a user.
"""
tmp_path_factory: TempPathFactory = session.config._tmp_path_factory
if tmp_path_factory._basetemp is None:
return
policy = tmp_path_factory._retention_policy
if (
exitstatus == 0
and policy == "failed"
and tmp_path_factory._given_basetemp is None
):
passed_dir = tmp_path_factory._basetemp
if passed_dir.exists():
# We do a "best effort" to remove files, but it might not be possible due to some leaked resource,
# permissions, etc, in which case we ignore it.
rmtree(passed_dir, ignore_errors=True)
# tmporal directory cleanup, which is registered to
# this ExitStack, will be executed at the end of this scope
with tmp_path_factory._exit_stack:
if tmp_path_factory._basetemp is None:
return
policy = tmp_path_factory._retention_policy
if (
exitstatus == 0
and policy == "failed"
and tmp_path_factory._given_basetemp is None
):
passed_dir = tmp_path_factory._basetemp
if passed_dir.exists():
# We do a "best effort" to remove files, but it might not be possible due to some leaked resource,
# permissions, etc, in which case we ignore it.
rmtree(passed_dir, ignore_errors=True)
@hookimpl(tryfirst=True, hookwrapper=True)

93
testing/test_tmpdir.py
View File

@ -92,6 +92,73 @@ class TestConfigTmpPath:
assert mytemp.exists()
assert not mytemp.joinpath("hello").exists()
def test_policy_none_delete_all(self, pytester: Pytester) -> None:
p = pytester.makepyfile(
"""
def test_1(tmp_path):
assert 0 == 0
"""
)
p_failed = pytester.makepyfile(
another_file_name="""
def test_1(tmp_path):
assert 0 == 1
"""
)
pytester.makepyprojecttoml(
"""
[tool.pytest.ini_options]
tmp_path_retention_policy = "none"
"""
)
pytester.inline_run(p)
pytester.inline_run(p_failed)
root = pytester._test_tmproot
for child in root.iterdir():
base_dir = list(child.iterdir())
# Check the base dir itself is gone without depending on test results
assert base_dir == []
@pytest.mark.parametrize("policy", ['"failed"', '"all"'])
@pytest.mark.parametrize("count", [0, 1, 3])
def test_retention_count(self, pytester: Pytester, policy, count) -> None:
p = pytester.makepyfile(
"""
def test_1(tmp_path):
assert 0 == 0
"""
)
p_failed = pytester.makepyfile(
another_file_name="""
def test_1(tmp_path):
assert 0 == 1
"""
)
pytester.makepyprojecttoml(
f"""
[tool.pytest.ini_options]
tmp_path_retention_policy = {policy}
tmp_path_retention_count = {count}
"""
)
pytester.inline_run(p)
pytester.inline_run(p_failed)
pytester.inline_run(p)
pytester.inline_run(p_failed)
pytester.inline_run(p)
pytester.inline_run(p_failed)
pytester.inline_run(p)
pytester.inline_run(p_failed)
root = pytester._test_tmproot
for child in root.iterdir():
base_dir = filter(lambda x: not x.is_symlink(), child.iterdir())
assert len(list(base_dir)) == count
def test_policy_failed_removes_only_passed_dir(self, pytester: Pytester) -> None:
p = pytester.makepyfile(
"""
@ -119,26 +186,6 @@ class TestConfigTmpPath:
assert len(test_dir) == 1
assert test_dir[0].name == "test_20"
def test_policy_failed_removes_basedir_when_all_passed(
self, pytester: Pytester
) -> None:
p = pytester.makepyfile(
"""
def test_1(tmp_path):
assert 0 == 0
"""
)
pytester.inline_run(p)
root = pytester._test_tmproot
for child in root.iterdir():
# This symlink will be deleted by cleanup_numbered_dir **after**
# the test finishes because it's triggered by atexit.
# So it has to be ignored here.
base_dir = filter(lambda x: not x.is_symlink(), child.iterdir())
# Check the base dir itself is gone
assert len(list(base_dir)) == 0
# issue #10502
def test_policy_failed_removes_dir_when_skipped_from_fixture(
self, pytester: Pytester
@ -160,10 +207,8 @@ class TestConfigTmpPath:
# Check if the whole directory is removed
root = pytester._test_tmproot
for child in root.iterdir():
base_dir = list(
filter(lambda x: x.is_dir() and not x.is_symlink(), child.iterdir())
)
assert len(base_dir) == 0
base_dir = list(child.iterdir())
assert base_dir == []
# issue #10502
def test_policy_all_keeps_dir_when_skipped_from_fixture(