diff --git a/tests/test_pylint_runners.py b/tests/test_pylint_runners.py index d334b08a496..5fc2da73f35 100644 --- a/tests/test_pylint_runners.py +++ b/tests/test_pylint_runners.py @@ -71,36 +71,38 @@ def test_pylint_argument_deduplication( assert err.value.code == 0 -def test_pylint_run_jobs_equal_zero_dont_crash_with_cpu_fraction( +@pytest.mark.parametrize( + "quota,shares,period", + [ + # Shares path + ("-1", "2", ""), + ("-1", "1023", ""), + ("-1", "1024", ""), + # Periods path + ("100", "", "200"), + ("999", "", "1000"), + ("1000", "", "1000"), + ], +) +def test_pylint_run_dont_crash_with_cgroupv1( tmp_path: pathlib.Path, + quota: str, + shares: str, + period: str, ) -> None: """Check that the pylint runner does not crash if `pylint.lint.run._query_cpu` determines only a fraction of a CPU core to be available. """ - builtin_open = open - - def _mock_open(*args: Any, **kwargs: Any) -> BufferedReader: - if args[0] == "/sys/fs/cgroup/cpu/cpu.cfs_quota_us": - return mock_open(read_data=b"-1")(*args, **kwargs) # type: ignore[no-any-return] - if args[0] == "/sys/fs/cgroup/cpu/cpu.shares": - return mock_open(read_data=b"2")(*args, **kwargs) # type: ignore[no-any-return] - return builtin_open(*args, **kwargs) # type: ignore[no-any-return] - - pathlib_path = pathlib.Path - - def _mock_path(*args: str, **kwargs: Any) -> pathlib.Path: - if args[0] == "/sys/fs/cgroup/cpu/cpu.shares": - return MagicMock(is_file=lambda: True) - if args[0] == "/sys/fs/cgroup/cpu.max": - return MagicMock(is_file=lambda: False) - return pathlib_path(*args, **kwargs) - filepath = os.path.abspath(__file__) testargs = [filepath, "--jobs=0"] + with _test_cwd(tmp_path): with pytest.raises(SystemExit) as err: - with patch("builtins.open", _mock_open): - with patch("pylint.lint.run.Path", _mock_path): + with patch( + "builtins.open", + mock_cgroup_fs(quota=quota, shares=shares, period=period), + ): + with patch("pylint.lint.run.Path", mock_cgroup_path(v2=False)): Run(testargs, reporter=Reporter()) assert err.value.code == 0 @@ -112,37 +114,20 @@ def _mock_path(*args: str, **kwargs: Any) -> pathlib.Path: "max 100000", ], ) -def test_pylint_run_jobs_equal_zero_dont_crash_with_cgroupv2( +def test_pylint_run_dont_crash_with_cgroupv2( tmp_path: pathlib.Path, contents: str, ) -> None: """Check that the pylint runner does not crash if `pylint.lint.run._query_cpu` determines only a fraction of a CPU core to be available. """ - builtin_open = open - - def _mock_open(*args: Any, **kwargs: Any) -> BufferedReader: - if args[0] == "/sys/fs/cgroup/cpu.max": - return mock_open(read_data=contents)(*args, **kwargs) # type: ignore[no-any-return] - return builtin_open(*args, **kwargs) # type: ignore[no-any-return] - - pathlib_path = pathlib.Path - - def _mock_path(*args: str, **kwargs: Any) -> pathlib.Path: - if args[0] == "/sys/fs/cgroup/cpu/cpu.shares": - return MagicMock(is_file=lambda: False) - if args[0] == "/sys/fs/cgroup/cpu/cfs_quota_us": - return MagicMock(is_file=lambda: False) - if args[0] == "/sys/fs/cgroup/cpu.max": - return MagicMock(is_file=lambda: True) - return pathlib_path(*args, **kwargs) - filepath = os.path.abspath(__file__) testargs = [filepath, "--jobs=0"] + with _test_cwd(tmp_path): with pytest.raises(SystemExit) as err: - with patch("builtins.open", _mock_open): - with patch("pylint.lint.run.Path", _mock_path): + with patch("builtins.open", mock_cgroup_fs(max_v2=contents)): + with patch("pylint.lint.run.Path", mock_cgroup_path(v2=True)): Run(testargs, reporter=Reporter()) assert err.value.code == 0 @@ -167,26 +152,76 @@ def test_query_cpu_cgroupv2( """Check that `pylint.lint.run._query_cpu` generates realistic values in cgroupsv2 systems. """ - builtin_open = open + with _test_cwd(tmp_path): + with patch("builtins.open", mock_cgroup_fs(max_v2=contents)): + with patch("pylint.lint.run.Path", mock_cgroup_path(v2=True)): + cpus = _query_cpu() + assert cpus == expected - def _mock_open(*args: Any, **kwargs: Any) -> BufferedReader: - if args[0] == "/sys/fs/cgroup/cpu.max": - return mock_open(read_data=contents)(*args, **kwargs) # type: ignore[no-any-return] - return builtin_open(*args, **kwargs) # type: ignore[no-any-return] - pathlib_path = pathlib.Path +@pytest.mark.parametrize( + "quota,shares,period,expected", + [ + # Shares path + ("-1", "2", "", 1), + ("-1", "1023", "", 1), + ("-1", "1024", "", 1), + ("-1", "2048", "", 2), + # Periods path + ("100", "", "200", 1), + ("999", "", "1000", 1), + ("1000", "", "1000", 1), + ("2000", "", "1000", 2), + ], +) +def test_query_cpu_cgroupv1( + tmp_path: pathlib.Path, + quota: str, + shares: str, + period: str, + expected: int, +) -> None: + """Check that `pylint.lint.run._query_cpu` generates realistic values in cgroupsv1 + systems. + """ + with _test_cwd(tmp_path): + with patch( + "builtins.open", mock_cgroup_fs(quota=quota, shares=shares, period=period) + ): + with patch("pylint.lint.run.Path", mock_cgroup_path(v2=False)): + cpus = _query_cpu() + assert cpus == expected + +def mock_cgroup_path(v2: bool) -> Any: def _mock_path(*args: str, **kwargs: Any) -> pathlib.Path: + if args[0] == "/sys/fs/cgroup/cpu/cpu.cfs_period_us": + return MagicMock(is_file=lambda: not v2) if args[0] == "/sys/fs/cgroup/cpu/cpu.shares": - return MagicMock(is_file=lambda: False) - if args[0] == "/sys/fs/cgroup/cpu/cfs_quota_us": - return MagicMock(is_file=lambda: False) + return MagicMock(is_file=lambda: not v2) + if args[0] == "/sys/fs/cgroup/cpu/cpu.cfs_quota_us": + return MagicMock(is_file=lambda: not v2) if args[0] == "/sys/fs/cgroup/cpu.max": - return MagicMock(is_file=lambda: True) - return pathlib_path(*args, **kwargs) + return MagicMock(is_file=lambda: v2) + return pathlib.Path(*args, **kwargs) - with _test_cwd(tmp_path): - with patch("builtins.open", _mock_open): - with patch("pylint.lint.run.Path", _mock_path): - cpus = _query_cpu() - assert cpus == expected + return _mock_path + + +def mock_cgroup_fs( + quota: str = "", shares: str = "", period: str = "", max_v2: str = "" +) -> Any: + builtin_open = open + + def _mock_open(*args: Any, **kwargs: Any) -> BufferedReader: + if args[0] == "/sys/fs/cgroup/cpu/cpu.cfs_quota_us": + return mock_open(read_data=quota)(*args, **kwargs) # type: ignore[no-any-return] + if args[0] == "/sys/fs/cgroup/cpu/cpu.shares": + return mock_open(read_data=shares)(*args, **kwargs) # type: ignore[no-any-return] + if args[0] == "/sys/fs/cgroup/cpu/cpu.cfs_period_us": + return mock_open(read_data=period)(*args, **kwargs) # type: ignore[no-any-return] + if args[0] == "/sys/fs/cgroup/cpu.max": + return mock_open(read_data=max_v2)(*args, **kwargs) # type: ignore[no-any-return] + return builtin_open(*args, **kwargs) # type: ignore[no-any-return] + + return _mock_open