File: test_cpumask.py

package info (click to toggle)
drgn 0.1.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 7,852 kB
  • sloc: python: 74,992; ansic: 54,589; awk: 423; makefile: 351; sh: 99
file content (60 lines) | stat: -rw-r--r-- 2,163 bytes parent folder | download
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
# Copyright (c) Meta Platforms, Inc. and affiliates.
# SPDX-License-Identifier: LGPL-2.1-or-later

from pathlib import Path

import drgn.helpers.linux.cpumask
from drgn.helpers.linux.cpumask import cpumask_of, cpumask_to_cpulist, for_each_cpu
from tests.linux_kernel import LinuxKernelTestCase, parse_range_list, possible_cpus

CPU_PATH = Path("/sys/devices/system/cpu")


class TestCpuMask(LinuxKernelTestCase):
    _MASKS = ("online", "possible", "present")

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        for online_path in sorted(CPU_PATH.glob("cpu*/online")):
            if int(online_path.read_text()):
                cls.offlined_path = online_path
                online_path.write_text("0")
                cls.addClassCleanup(online_path.write_text, "1")
                break

    def test_for_each_cpu(self):
        for name in self._MASKS:
            with self.subTest(name=name):
                self.assertEqual(
                    list(
                        getattr(drgn.helpers.linux.cpumask, f"for_each_{name}_cpu")(
                            self.prog
                        )
                    ),
                    sorted(parse_range_list((CPU_PATH / name).read_text())),
                )

    def test_cpumask_to_cpulist(self):
        for name in self._MASKS:
            with self.subTest(name=name):
                self.assertEqual(
                    cpumask_to_cpulist(
                        getattr(drgn.helpers.linux.cpumask, f"cpu_{name}_mask")(
                            self.prog
                        )
                    ),
                    (CPU_PATH / name).read_text().strip(),
                )

    def test_cpumask_weight(self):
        for name in self._MASKS:
            with self.subTest(name=name):
                self.assertEqual(
                    getattr(drgn.helpers.linux.cpumask, f"num_{name}_cpus")(self.prog),
                    len(parse_range_list((CPU_PATH / name).read_text())),
                )

    def test_cpumask_of(self):
        cpu = max(possible_cpus())
        self.assertEqual(list(for_each_cpu(cpumask_of(self.prog, cpu))), [cpu])