Refactor target_concurrency tests

This commit is contained in:
Max R 2023-09-09 21:54:47 -04:00
parent e2c6a822c7
commit 9ac229dad8
2 changed files with 51 additions and 46 deletions

View file

@ -1,6 +1,7 @@
from __future__ import annotations
import concurrent.futures
import multiprocessing
import os
import sys
import time
@ -12,6 +13,40 @@ from pre_commit import parse_shebang
from pre_commit import xargs
def test_cpu_count_sched_getaffinity_exists():
with mock.patch.object(
os, 'sched_getaffinity', create=True, return_value=set(range(345)),
):
assert xargs.cpu_count() == 345
@pytest.fixture
def no_sched_getaffinity():
# Simulates an OS without os.sched_getaffinity available (mac/windows)
# https://docs.python.org/3/library/os.html#interface-to-the-scheduler
with mock.patch.object(
os,
'sched_getaffinity',
create=True,
side_effect=AttributeError,
):
yield
def test_cpu_count_multiprocessing_cpu_count_implemented(no_sched_getaffinity):
with mock.patch.object(multiprocessing, 'cpu_count', return_value=123):
assert xargs.cpu_count() == 123
def test_cpu_count_multiprocessing_cpu_count_not_implemented(
no_sched_getaffinity,
):
with mock.patch.object(
multiprocessing, 'cpu_count', side_effect=NotImplementedError,
):
assert xargs.cpu_count() == 1
@pytest.mark.parametrize(
('env', 'expected'),
(