<commit_before>import os
import re
import pytest

from threadpoolctl import threadpool_limits, threadpool_info, _ThreadpoolInfo
from threadpoolctl import _ALL_PREFIXES, _ALL_USER_APIS

from .utils import cython_extensions_compiled
from .utils import libopenblas_paths
from .utils import scipy


def is_old_openblas(module):
    # Possible bug in getting maximum number of threads with OpenBLAS < 0.2.16
    # and OpenBLAS does not expose its version before 0.3.4.
    return module.internal_api == "openblas" and module.version is None


def effective_num_threads(nthreads, max_threads):
    if nthreads is None or nthreads > max_threads:
        return max_threads
    return nthreads


def _threadpool_info():
    # Like threadpool_info but return the object instead of the list of dicts
    return _ThreadpoolInfo(user_api=_ALL_USER_APIS)


def test_threadpool_limits_public_api():
    # Check consistency between threadpool_info and _ThreadpoolInfo
    public_info = threadpool_info()
    private_info = _threadpool_info()

    for module1, module2 in zip(public_info, private_info):
        assert module1 == module2.todict()


def test_ThreadpoolInfo_todicts():
    info = _threadpool_info()

    assert info.todicts() == [module.todict() for module in info]
    assert info.todicts() == [module.todict() for module in info.modules]

    for module in info:
        module_dict = module.todict()
        assert "user_api" in module_dict
        assert "internal_api" in module_dict
        assert "prefix" in module_dict
        assert "filepath" in module_dict
        assert "version" in module_dict
        assert "num_threads" in module_dict


@pytest.mark.parametrize("prefix", _ALL_PREFIXES)
@pytest.mark.parametrize("limit", [1, 3])
def test_threadpool_limits_by_prefix(prefix, limit):
    # Check that the maximum number of threads can be set by prefix
    original_infos = _threadpool_info()

    modules_matching_prefix = original_infos.get_modules("prefix", prefix)
    if not modules_matching_prefix:
        pytest.skip("Requires {} runtime".format(prefix))

    with threadpool_limits(limits={prefix: limit}):
        for module in modules_matching_prefix:
            if is_old_openblas(module):
                continue<commit_msg>better tests for todict(s) methods<commit_after>import os
import re
import pytest

from threadpoolctl import threadpool_limits, threadpool_info, _ThreadpoolInfo
from threadpoolctl import _ALL_PREFIXES, _ALL_USER_APIS

from .utils import cython_extensions_compiled
from .utils import libopenblas_paths
from .utils import scipy


def is_old_openblas(module):
    # Possible bug in getting maximum number of threads with OpenBLAS < 0.2.16
    # and OpenBLAS does not expose its version before 0.3.4.
    return module.internal_api == "openblas" and module.version is None


def effective_num_threads(nthreads, max_threads):
    if nthreads is None or nthreads > max_threads:
        return max_threads
    return nthreads


def _threadpool_info():
    # Like threadpool_info but return the object instead of the list of dicts
    return _ThreadpoolInfo(user_api=_ALL_USER_APIS)


def test_threadpool_limits_public_api():
    # Check consistency between threadpool_info and _ThreadpoolInfo
    public_info = threadpool_info()
    private_info = _threadpool_info()

    for module1, module2 in zip(public_info, private_info):
        assert module1 == module2.todict()


def test_ThreadpoolInfo_todicts():
    # Check all keys expected for the public api are in the dicts returned by
    # the .todict(s) methods
    info = _threadpool_info()

    assert threadpool_info() == [module.todict() for module in info.modules]
    assert info.todicts() == [module.todict() for module in info]
    assert info.todicts() == [module.todict() for module in info.modules]

    for module in info:
        module_dict = module.todict()
        assert "user_api" in module_dict
        assert "internal_api" in module_dict
        assert "prefix" in module_dict
        assert "filepath" in module_dict
        assert "version" in module_dict
        assert "num_threads" in module_dict

        if module.internal_api == "mkl":
            assert "threading_layer" in module_dict


@pytest.mark.parametrize("prefix", _ALL_PREFIXES)
@pytest.mark.parametrize("limit", [1, 3])
def test_threadpool_limits_by_prefix(prefix, limit):
    # Check that the maximum number of threads can be set by prefix
    original_infos = _threadpool_info()

    modules_matching_prefix = original_infos.get_modules("prefix", prefix)
    if not modules_matching_prefix:
        pytest.skip("Requires {} runtime".format(prefix))

    with threadpool_limits(limits={prefix: limit}):
        for module in modules_matching_prefix:
            if is_old_openblas(module):
                continue
{'fraction_non_alphanumeric': 0.07138397502601457, 'fraction_numerical': 0.005411030176899064, 'mean_word_length': 3.8891149542217702, 'pattern_counts': {'":': 1, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '93875', 'n_tokens_mistral': 1458, 'n_tokens_neox': 1429, 'n_words': 428}