Change to use atexit
This commit is contained in:
parent
a0b5dbc11b
commit
58c5d5089a
|
@ -1,16 +0,0 @@
|
|||
from typing import Any
|
||||
from typing import Callable
|
||||
from typing import List
|
||||
from typing import Tuple
|
||||
|
||||
|
||||
class Callback:
|
||||
def __init__(self) -> None:
|
||||
self._funcs: List[Tuple[Callable[..., Any], Any, Any]] = []
|
||||
|
||||
def register(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
|
||||
self._funcs.append((func, args, kwargs))
|
||||
|
||||
def execute(self) -> None:
|
||||
for func, args, kwargs in reversed(self._funcs):
|
||||
func(*args, **kwargs)
|
|
@ -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,7 +361,7 @@ def make_numbered_dir_with_cleanup(
|
|||
keep: int,
|
||||
lock_timeout: float,
|
||||
mode: int,
|
||||
register=atexit.register,
|
||||
register,
|
||||
) -> Path:
|
||||
"""Create a numbered dir with a cleanup lock and remove old ones."""
|
||||
e = None
|
||||
|
|
|
@ -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
|
||||
|
@ -37,7 +38,6 @@ from _pytest.deprecated import check_ispytest
|
|||
from _pytest.fixtures import fixture
|
||||
from _pytest.fixtures import FixtureRequest
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
from _pytest.callback import Callback
|
||||
|
||||
tmppath_result_key = StashKey[Dict[str, bool]]()
|
||||
|
||||
|
@ -78,7 +78,7 @@ class TempPathFactory:
|
|||
self._retention_count = retention_count
|
||||
self._retention_policy = retention_policy
|
||||
self._basetemp = basetemp
|
||||
self._lock_and_dir_removal_callback = Callback()
|
||||
self._exit_stack = ExitStack()
|
||||
|
||||
@classmethod
|
||||
def from_config(
|
||||
|
@ -198,7 +198,7 @@ class TempPathFactory:
|
|||
keep=keep,
|
||||
lock_timeout=LOCK_TIMEOUT,
|
||||
mode=0o700,
|
||||
register=self._lock_and_dir_removal_callback.register,
|
||||
register=self._exit_stack.callback,
|
||||
)
|
||||
assert basetemp is not None, basetemp
|
||||
self._basetemp = basetemp
|
||||
|
@ -304,6 +304,10 @@ 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
|
||||
|
||||
# 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
|
||||
|
@ -318,8 +322,6 @@ def pytest_sessionfinish(session, exitstatus: Union[int, ExitCode]):
|
|||
# permissions, etc, in which case we ignore it.
|
||||
rmtree(passed_dir, ignore_errors=True)
|
||||
|
||||
tmp_path_factory._lock_and_dir_removal_callback.execute()
|
||||
|
||||
|
||||
@hookimpl(tryfirst=True, hookwrapper=True)
|
||||
def pytest_runtest_makereport(item: Item, call):
|
||||
|
|
Loading…
Reference in New Issue