finalizare 1.0
This commit is contained in:
andrei-mihnea-cerbu
@@ -0,0 +1,238 @@
|
||||
# Pytest customization
|
||||
import json
|
||||
import os
|
||||
import warnings
|
||||
import tempfile
|
||||
|
||||
import numpy as np
|
||||
import numpy.testing as npt
|
||||
import pytest
|
||||
import hypothesis
|
||||
|
||||
from scipy._lib._fpumode import get_fpu_mode
|
||||
from scipy._lib._testutils import FPUModeChangeWarning
|
||||
from scipy._lib import _pep440
|
||||
from scipy._lib._array_api import SCIPY_ARRAY_API, SCIPY_DEVICE
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
config.addinivalue_line("markers",
|
||||
"slow: Tests that are very slow.")
|
||||
config.addinivalue_line("markers",
|
||||
"xslow: mark test as extremely slow (not run unless explicitly requested)")
|
||||
config.addinivalue_line("markers",
|
||||
"xfail_on_32bit: mark test as failing on 32-bit platforms")
|
||||
try:
|
||||
import pytest_timeout # noqa:F401
|
||||
except Exception:
|
||||
config.addinivalue_line(
|
||||
"markers", 'timeout: mark a test for a non-default timeout')
|
||||
config.addinivalue_line("markers",
|
||||
"skip_if_array_api(*backends, reasons=None, np_only=False, cpu_only=False): "
|
||||
"mark the desired skip configuration for the `skip_if_array_api` fixture.")
|
||||
|
||||
|
||||
def _get_mark(item, name):
|
||||
if _pep440.parse(pytest.__version__) >= _pep440.Version("3.6.0"):
|
||||
mark = item.get_closest_marker(name)
|
||||
else:
|
||||
mark = item.get_marker(name)
|
||||
return mark
|
||||
|
||||
|
||||
def pytest_runtest_setup(item):
|
||||
mark = _get_mark(item, "xslow")
|
||||
if mark is not None:
|
||||
try:
|
||||
v = int(os.environ.get('SCIPY_XSLOW', '0'))
|
||||
except ValueError:
|
||||
v = False
|
||||
if not v:
|
||||
pytest.skip("very slow test; "
|
||||
"set environment variable SCIPY_XSLOW=1 to run it")
|
||||
mark = _get_mark(item, 'xfail_on_32bit')
|
||||
if mark is not None and np.intp(0).itemsize < 8:
|
||||
pytest.xfail(f'Fails on our 32-bit test platform(s): {mark.args[0]}')
|
||||
|
||||
# Older versions of threadpoolctl have an issue that may lead to this
|
||||
# warning being emitted, see gh-14441
|
||||
with npt.suppress_warnings() as sup:
|
||||
sup.filter(pytest.PytestUnraisableExceptionWarning)
|
||||
|
||||
try:
|
||||
from threadpoolctl import threadpool_limits
|
||||
|
||||
HAS_THREADPOOLCTL = True
|
||||
except Exception: # observed in gh-14441: (ImportError, AttributeError)
|
||||
# Optional dependency only. All exceptions are caught, for robustness
|
||||
HAS_THREADPOOLCTL = False
|
||||
|
||||
if HAS_THREADPOOLCTL:
|
||||
# Set the number of openmp threads based on the number of workers
|
||||
# xdist is using to prevent oversubscription. Simplified version of what
|
||||
# sklearn does (it can rely on threadpoolctl and its builtin OpenMP helper
|
||||
# functions)
|
||||
try:
|
||||
xdist_worker_count = int(os.environ['PYTEST_XDIST_WORKER_COUNT'])
|
||||
except KeyError:
|
||||
# raises when pytest-xdist is not installed
|
||||
return
|
||||
|
||||
if not os.getenv('OMP_NUM_THREADS'):
|
||||
max_openmp_threads = os.cpu_count() // 2 # use nr of physical cores
|
||||
threads_per_worker = max(max_openmp_threads // xdist_worker_count, 1)
|
||||
try:
|
||||
threadpool_limits(threads_per_worker, user_api='blas')
|
||||
except Exception:
|
||||
# May raise AttributeError for older versions of OpenBLAS.
|
||||
# Catch any error for robustness.
|
||||
return
|
||||
|
||||
|
||||
@pytest.fixture(scope="function", autouse=True)
|
||||
def check_fpu_mode(request):
|
||||
"""
|
||||
Check FPU mode was not changed during the test.
|
||||
"""
|
||||
old_mode = get_fpu_mode()
|
||||
yield
|
||||
new_mode = get_fpu_mode()
|
||||
|
||||
if old_mode != new_mode:
|
||||
warnings.warn(f"FPU mode changed from {old_mode:#x} to {new_mode:#x} during "
|
||||
"the test",
|
||||
category=FPUModeChangeWarning, stacklevel=0)
|
||||
|
||||
|
||||
# Array API backend handling
|
||||
xp_available_backends = {'numpy': np}
|
||||
|
||||
if SCIPY_ARRAY_API and isinstance(SCIPY_ARRAY_API, str):
|
||||
# fill the dict of backends with available libraries
|
||||
try:
|
||||
import array_api_strict
|
||||
xp_available_backends.update({'array_api_strict': array_api_strict})
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
try:
|
||||
import torch # type: ignore[import]
|
||||
xp_available_backends.update({'pytorch': torch})
|
||||
# can use `mps` or `cpu`
|
||||
torch.set_default_device(SCIPY_DEVICE)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
try:
|
||||
import cupy # type: ignore[import]
|
||||
xp_available_backends.update({'cupy': cupy})
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
# by default, use all available backends
|
||||
if SCIPY_ARRAY_API.lower() not in ("1", "true"):
|
||||
SCIPY_ARRAY_API_ = json.loads(SCIPY_ARRAY_API)
|
||||
|
||||
if 'all' in SCIPY_ARRAY_API_:
|
||||
pass # same as True
|
||||
else:
|
||||
# only select a subset of backend by filtering out the dict
|
||||
try:
|
||||
xp_available_backends = {
|
||||
backend: xp_available_backends[backend]
|
||||
for backend in SCIPY_ARRAY_API_
|
||||
}
|
||||
except KeyError:
|
||||
msg = f"'--array-api-backend' must be in {xp_available_backends.keys()}"
|
||||
raise ValueError(msg)
|
||||
|
||||
if 'cupy' in xp_available_backends:
|
||||
SCIPY_DEVICE = 'cuda'
|
||||
|
||||
array_api_compatible = pytest.mark.parametrize("xp", xp_available_backends.values())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def skip_if_array_api(xp, request):
|
||||
"""
|
||||
Skip based on the ``skip_if_array_api`` marker.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
*backends : tuple
|
||||
Backends to skip, e.g. ``("array_api_strict", "torch")``.
|
||||
These are overriden when ``np_only`` is ``True``, and are not
|
||||
necessary to provide for non-CPU backends when ``cpu_only`` is ``True``.
|
||||
reasons : list, optional
|
||||
A list of reasons for each skip. When ``np_only`` is ``True``,
|
||||
this should be a singleton list. Otherwise, this should be a list
|
||||
of reasons, one for each corresponding backend in ``backends``.
|
||||
If unprovided, default reasons are used. Note that it is not possible
|
||||
to specify a custom reason with ``cpu_only``. Default: ``None``.
|
||||
np_only : bool, optional
|
||||
When ``True``, the test is skipped for all backends other
|
||||
than the default NumPy backend. There is no need to provide
|
||||
any ``backends`` in this case. To specify a reason, pass a
|
||||
singleton list to ``reasons``. Default: ``False``.
|
||||
cpu_only : bool, optional
|
||||
When ``True``, the test is skipped on non-CPU devices.
|
||||
There is no need to provide any ``backends`` in this case,
|
||||
but any ``backends`` will also be skipped on the CPU.
|
||||
Default: ``False``.
|
||||
"""
|
||||
if "skip_if_array_api" not in request.keywords:
|
||||
return
|
||||
backends = request.keywords["skip_if_array_api"].args
|
||||
kwargs = request.keywords["skip_if_array_api"].kwargs
|
||||
np_only = kwargs.get("np_only", False)
|
||||
cpu_only = kwargs.get("cpu_only", False)
|
||||
if np_only:
|
||||
reasons = kwargs.get("reasons", ["do not run with non-NumPy backends."])
|
||||
reason = reasons[0]
|
||||
if xp.__name__ != 'numpy':
|
||||
pytest.skip(reason=reason)
|
||||
return
|
||||
if cpu_only:
|
||||
reason = "do not run with `SCIPY_ARRAY_API` set and not on CPU"
|
||||
if SCIPY_ARRAY_API and SCIPY_DEVICE != 'cpu':
|
||||
if xp.__name__ == 'cupy':
|
||||
pytest.skip(reason=reason)
|
||||
elif xp.__name__ == 'torch':
|
||||
if 'cpu' not in torch.empty(0).device.type:
|
||||
pytest.skip(reason=reason)
|
||||
if backends is not None:
|
||||
reasons = kwargs.get("reasons", False)
|
||||
for i, backend in enumerate(backends):
|
||||
if xp.__name__ == backend:
|
||||
if not reasons:
|
||||
reason = f"do not run with array API backend: {backend}"
|
||||
else:
|
||||
reason = reasons[i]
|
||||
pytest.skip(reason=reason)
|
||||
|
||||
|
||||
# Following the approach of NumPy's conftest.py...
|
||||
# Use a known and persistent tmpdir for hypothesis' caches, which
|
||||
# can be automatically cleared by the OS or user.
|
||||
hypothesis.configuration.set_hypothesis_home_dir(
|
||||
os.path.join(tempfile.gettempdir(), ".hypothesis")
|
||||
)
|
||||
|
||||
# We register two custom profiles for SciPy - for details see
|
||||
# https://hypothesis.readthedocs.io/en/latest/settings.html
|
||||
# The first is designed for our own CI runs; the latter also
|
||||
# forces determinism and is designed for use via scipy.test()
|
||||
hypothesis.settings.register_profile(
|
||||
name="nondeterministic", deadline=None, print_blob=True,
|
||||
)
|
||||
hypothesis.settings.register_profile(
|
||||
name="deterministic",
|
||||
deadline=None, print_blob=True, database=None, derandomize=True,
|
||||
suppress_health_check=list(hypothesis.HealthCheck),
|
||||
)
|
||||
|
||||
# Profile is currently set by environment variable `SCIPY_HYPOTHESIS_PROFILE`
|
||||
# In the future, it would be good to work the choice into dev.py.
|
||||
SCIPY_HYPOTHESIS_PROFILE = os.environ.get("SCIPY_HYPOTHESIS_PROFILE",
|
||||
"deterministic")
|
||||
hypothesis.settings.load_profile(SCIPY_HYPOTHESIS_PROFILE)
|
||||
Reference in New Issue
Block a user