1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
import contextlib
import os
import typing
import unittest
import unittest.mock
from typing import Iterator, Optional, Sequence
import tools.setup_helpers.cmake
import tools.setup_helpers.env # noqa: F401 unused but resolves circular import
T = typing.TypeVar("T")
class TestCMake(unittest.TestCase):
@unittest.mock.patch("multiprocessing.cpu_count")
def test_build_jobs(self, mock_cpu_count: unittest.mock.MagicMock) -> None:
"""Tests that the number of build jobs comes out correctly."""
mock_cpu_count.return_value = 13
cases = [
# MAX_JOBS, USE_NINJA, IS_WINDOWS, want
(("8", True, False), ["-j", "8"]), # noqa: E201,E241
((None, True, False), None), # noqa: E201,E241
(("7", False, False), ["-j", "7"]), # noqa: E201,E241
((None, False, False), ["-j", "13"]), # noqa: E201,E241
(("6", True, True), ["-j", "6"]), # noqa: E201,E241
((None, True, True), None), # noqa: E201,E241
(("11", False, True), ["/p:CL_MPCount=11"]), # noqa: E201,E241
((None, False, True), ["/p:CL_MPCount=13"]), # noqa: E201,E241
]
for (max_jobs, use_ninja, is_windows), want in cases:
with self.subTest(
MAX_JOBS=max_jobs, USE_NINJA=use_ninja, IS_WINDOWS=is_windows
):
with contextlib.ExitStack() as stack:
stack.enter_context(env_var("MAX_JOBS", max_jobs))
stack.enter_context(
unittest.mock.patch.object(
tools.setup_helpers.cmake, "USE_NINJA", use_ninja
)
)
stack.enter_context(
unittest.mock.patch.object(
tools.setup_helpers.cmake, "IS_WINDOWS", is_windows
)
)
cmake = tools.setup_helpers.cmake.CMake()
with unittest.mock.patch.object(cmake, "run") as cmake_run:
cmake.build({})
cmake_run.assert_called_once()
(call,) = cmake_run.mock_calls
build_args, _ = call.args
if want is None:
self.assertNotIn("-j", build_args)
else:
self.assert_contains_sequence(build_args, want)
@staticmethod
def assert_contains_sequence(
sequence: Sequence[T], subsequence: Sequence[T]
) -> None:
"""Raises an assertion if the subsequence is not contained in the sequence."""
if len(subsequence) == 0:
return # all sequences contain the empty subsequence
# Iterate over all windows of len(subsequence). Stop if the
# window matches.
for i in range(len(sequence) - len(subsequence) + 1):
candidate = sequence[i : i + len(subsequence)]
assert len(candidate) == len(subsequence) # sanity check
if candidate == subsequence:
return # found it
raise AssertionError(f"{subsequence} not found in {sequence}")
@contextlib.contextmanager
def env_var(key: str, value: Optional[str]) -> Iterator[None]:
"""Sets/clears an environment variable within a Python context."""
# Get the previous value and then override it.
previous_value = os.environ.get(key)
set_env_var(key, value)
try:
yield
finally:
# Restore to previous value.
set_env_var(key, previous_value)
def set_env_var(key: str, value: Optional[str]) -> None:
"""Sets/clears an environment variable."""
if value is None:
os.environ.pop(key, None)
else:
os.environ[key] = value
if __name__ == "__main__":
unittest.main()
|
