77
88import pytest
99
10+ from threadpoolctl import threadpool_info
1011from threadpoolctl ._thread_tracer import ThreadTracerError , WindowsThreadSpawnTracer
1112
1213pytestmark = [
3132}
3233
3334
34- def _threadpool_info ():
35- try :
36- from threadpoolctl import threadpool_info
37-
38- return threadpool_info ()
39- except ImportError :
40- from threadpoolctl import get_threadpool_limits
41-
42- return get_threadpool_limits ()
43-
44-
45- def _module_num_threads (module ):
46- if "num_threads" in module :
47- return module ["num_threads" ]
48- return module .get ("n_thread" )
49-
50-
5135def _expected_pool_thread_count (user_api ):
5236 thread_counts = [
53- _module_num_threads ( module )
54- for module in _threadpool_info ()
55- if module .get ("user_api" ) == user_api and _module_num_threads ( module )
37+ module [ "num_threads" ]
38+ for module in threadpool_info ()
39+ if module .get ("user_api" ) == user_api and module . get ( "num_threads" )
5640 ]
5741 if not thread_counts :
5842 pytest .skip ("No {0} thread pool detected" .format (user_api ))
@@ -65,13 +49,15 @@ def _configure_blas_thread_env():
6549
6650
6751def _child_script (body ):
68- return textwrap .dedent ("""
52+ return textwrap .dedent (
53+ """
6954 import time
7055
7156 time.sleep({attach_delay})
7257 {body}
7358 time.sleep({flush_delay})
74- """ ).format (
59+ """
60+ ).format (
7561 attach_delay = TRACER_ATTACH_DELAY_SECONDS ,
7662 body = textwrap .indent (textwrap .dedent (body ).strip (), " " ),
7763 flush_delay = TRACER_FLUSH_DELAY_SECONDS ,
0 commit comments