jpayne@69: # Pytest customization
jpayne@69: import os
jpayne@69: import pytest
jpayne@69: import warnings
jpayne@69: 
jpayne@69: import numpy as np
jpayne@69: import numpy.testing as npt
jpayne@69: from scipy._lib._fpumode import get_fpu_mode
jpayne@69: from scipy._lib._testutils import FPUModeChangeWarning
jpayne@69: from scipy._lib import _pep440
jpayne@69: 
jpayne@69: 
jpayne@69: def pytest_configure(config):
jpayne@69:     config.addinivalue_line("markers",
jpayne@69:         "slow: Tests that are very slow.")
jpayne@69:     config.addinivalue_line("markers",
jpayne@69:         "xslow: mark test as extremely slow (not run unless explicitly requested)")
jpayne@69:     config.addinivalue_line("markers",
jpayne@69:         "xfail_on_32bit: mark test as failing on 32-bit platforms")
jpayne@69:     try:
jpayne@69:         import pytest_timeout  # noqa:F401
jpayne@69:     except Exception:
jpayne@69:         config.addinivalue_line(
jpayne@69:             "markers", 'timeout: mark a test for a non-default timeout')
jpayne@69: 
jpayne@69: 
jpayne@69: def _get_mark(item, name):
jpayne@69:     if _pep440.parse(pytest.__version__) >= _pep440.Version("3.6.0"):
jpayne@69:         mark = item.get_closest_marker(name)
jpayne@69:     else:
jpayne@69:         mark = item.get_marker(name)
jpayne@69:     return mark
jpayne@69: 
jpayne@69: 
jpayne@69: def pytest_runtest_setup(item):
jpayne@69:     mark = _get_mark(item, "xslow")
jpayne@69:     if mark is not None:
jpayne@69:         try:
jpayne@69:             v = int(os.environ.get('SCIPY_XSLOW', '0'))
jpayne@69:         except ValueError:
jpayne@69:             v = False
jpayne@69:         if not v:
jpayne@69:             pytest.skip("very slow test; set environment variable SCIPY_XSLOW=1 to run it")
jpayne@69:     mark = _get_mark(item, 'xfail_on_32bit')
jpayne@69:     if mark is not None and np.intp(0).itemsize < 8:
jpayne@69:         pytest.xfail('Fails on our 32-bit test platform(s): %s' % (mark.args[0],))
jpayne@69: 
jpayne@69:     # Older versions of threadpoolctl have an issue that may lead to this
jpayne@69:     # warning being emitted, see gh-14441
jpayne@69:     with npt.suppress_warnings() as sup:
jpayne@69:         sup.filter(pytest.PytestUnraisableExceptionWarning)
jpayne@69: 
jpayne@69:         try:
jpayne@69:             from threadpoolctl import threadpool_limits
jpayne@69: 
jpayne@69:             HAS_THREADPOOLCTL = True
jpayne@69:         except Exception:  # observed in gh-14441: (ImportError, AttributeError)
jpayne@69:             # Optional dependency only. All exceptions are caught, for robustness
jpayne@69:             HAS_THREADPOOLCTL = False
jpayne@69: 
jpayne@69:         if HAS_THREADPOOLCTL:
jpayne@69:             # Set the number of openmp threads based on the number of workers
jpayne@69:             # xdist is using to prevent oversubscription. Simplified version of what
jpayne@69:             # sklearn does (it can rely on threadpoolctl and its builtin OpenMP helper
jpayne@69:             # functions)
jpayne@69:             try:
jpayne@69:                 xdist_worker_count = int(os.environ['PYTEST_XDIST_WORKER_COUNT'])
jpayne@69:             except KeyError:
jpayne@69:                 # raises when pytest-xdist is not installed
jpayne@69:                 return
jpayne@69: 
jpayne@69:             if not os.getenv('OMP_NUM_THREADS'):
jpayne@69:                 max_openmp_threads = os.cpu_count() // 2  # use nr of physical cores
jpayne@69:                 threads_per_worker = max(max_openmp_threads // xdist_worker_count, 1)
jpayne@69:                 try:
jpayne@69:                     threadpool_limits(threads_per_worker, user_api='blas')
jpayne@69:                 except Exception:
jpayne@69:                     # May raise AttributeError for older versions of OpenBLAS.
jpayne@69:                     # Catch any error for robustness.
jpayne@69:                     return
jpayne@69: 
jpayne@69: 
jpayne@69: @pytest.fixture(scope="function", autouse=True)
jpayne@69: def check_fpu_mode(request):
jpayne@69:     """
jpayne@69:     Check FPU mode was not changed during the test.
jpayne@69:     """
jpayne@69:     old_mode = get_fpu_mode()
jpayne@69:     yield
jpayne@69:     new_mode = get_fpu_mode()
jpayne@69: 
jpayne@69:     if old_mode != new_mode:
jpayne@69:         warnings.warn("FPU mode changed from {0:#x} to {1:#x} during "
jpayne@69:                       "the test".format(old_mode, new_mode),
jpayne@69:                       category=FPUModeChangeWarning, stacklevel=0)