Add extra cgroup tests

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