diff options
Diffstat (limited to 'psutil/tests')
| -rwxr-xr-x | psutil/tests/test_aix.py | 2 | ||||
| -rwxr-xr-x | psutil/tests/test_bsd.py | 4 | ||||
| -rwxr-xr-x | psutil/tests/test_contracts.py | 7 | ||||
| -rwxr-xr-x | psutil/tests/test_linux.py | 96 | ||||
| -rwxr-xr-x | psutil/tests/test_memleaks.py | 18 | ||||
| -rwxr-xr-x | psutil/tests/test_osx.py | 28 | ||||
| -rwxr-xr-x | psutil/tests/test_system.py | 98 | ||||
| -rwxr-xr-x | psutil/tests/test_windows.py | 15 |
8 files changed, 184 insertions, 84 deletions
diff --git a/psutil/tests/test_aix.py b/psutil/tests/test_aix.py index a32c3f6a..b1133833 100755 --- a/psutil/tests/test_aix.py +++ b/psutil/tests/test_aix.py @@ -107,7 +107,7 @@ class AIXSpecificTestCase(PsutilTestCase): def test_cpu_count_logical(self): out = sh('/usr/bin/mpstat -a') mpstat_lcpu = int(re.search(r"lcpu=(\d+)", out).group(1)) - psutil_lcpu = psutil.cpu_count(logical=True) + psutil_lcpu = psutil.cpu_count("logical") self.assertEqual(mpstat_lcpu, psutil_lcpu) def test_net_if_addrs_names(self): diff --git a/psutil/tests/test_bsd.py b/psutil/tests/test_bsd.py index b0bff87f..48b118ab 100755 --- a/psutil/tests/test_bsd.py +++ b/psutil/tests/test_bsd.py @@ -124,7 +124,7 @@ class BSDTestCase(PsutilTestCase): @unittest.skipIf(not which('sysctl'), "sysctl cmd not available") def test_cpu_count_logical(self): syst = sysctl("hw.ncpu") - self.assertEqual(psutil.cpu_count(logical=True), syst) + self.assertEqual(psutil.cpu_count("logical"), syst) @unittest.skipIf(not which('sysctl'), "sysctl cmd not available") def test_virtual_memory_total(self): @@ -457,7 +457,7 @@ class FreeBSDSystemTestCase(PsutilTestCase): # --- sensors_temperatures def test_sensors_temperatures_against_sysctl(self): - num_cpus = psutil.cpu_count(True) + num_cpus = psutil.cpu_count() for cpu in range(num_cpus): sensor = "dev.cpu.%s.temperature" % cpu # sysctl returns a string in the format 46.0C diff --git a/psutil/tests/test_contracts.py b/psutil/tests/test_contracts.py index 32c75fd7..3cb2c97f 100755 --- a/psutil/tests/test_contracts.py +++ b/psutil/tests/test_contracts.py @@ -233,6 +233,13 @@ class TestSystemAPITypes(PsutilTestCase): def test_cpu_count(self): self.assertIsInstance(psutil.cpu_count(), int) + self.assertIsNotNone(psutil.cpu_count("logical")) + if not OPENBSD or NETBSD: + self.assertIsNotNone(psutil.cpu_count("cores")) + if LINUX or MACOS or WINDOWS or FREEBSD: + self.assertIsNotNone(psutil.cpu_count("sockets")) + if LINUX or WINDOWS: + self.assertIsNotNone(psutil.cpu_count("numa")) @unittest.skipIf(not HAS_CPU_FREQ, "not supported") def test_cpu_freq(self): diff --git a/psutil/tests/test_linux.py b/psutil/tests/test_linux.py index 0c6d498c..e03490b3 100755 --- a/psutil/tests/test_linux.py +++ b/psutil/tests/test_linux.py @@ -58,6 +58,7 @@ if LINUX: SECTOR_SIZE = 512 EMPTY_TEMPERATURES = not glob.glob('/sys/class/hwmon/hwmon*') + # ===================================================================== # --- utils # ===================================================================== @@ -182,6 +183,14 @@ def vmstat(stat): raise ValueError("can't find %r in 'vmstat' output" % stat) +def lscpu(field): + out = sh("lscpu") + for line in out.splitlines(): + key, _, value = line.partition(':') + if field.lower() == key.lower(): + return value.strip() + + def get_free_version_info(): out = sh("free -V").strip() if 'UNKNOWN' in out: @@ -625,32 +634,11 @@ class TestSystemSwapMemory(PsutilTestCase): # ===================================================================== -# --- system CPU +# --- system CPU counts # ===================================================================== @unittest.skipIf(not LINUX, "LINUX only") -class TestSystemCPUTimes(PsutilTestCase): - - def test_fields(self): - fields = psutil.cpu_times()._fields - kernel_ver = re.findall(r'\d+\.\d+\.\d+', os.uname()[2])[0] - kernel_ver_info = tuple(map(int, kernel_ver.split('.'))) - if kernel_ver_info >= (2, 6, 11): - self.assertIn('steal', fields) - else: - self.assertNotIn('steal', fields) - if kernel_ver_info >= (2, 6, 24): - self.assertIn('guest', fields) - else: - self.assertNotIn('guest', fields) - if kernel_ver_info >= (3, 2, 0): - self.assertIn('guest_nice', fields) - else: - self.assertNotIn('guest_nice', fields) - - -@unittest.skipIf(not LINUX, "LINUX only") class TestSystemCPUCountLogical(PsutilTestCase): @unittest.skipIf(not os.path.exists("/sys/devices/system/cpu/online"), @@ -660,25 +648,24 @@ class TestSystemCPUCountLogical(PsutilTestCase): value = f.read().strip() if "-" in str(value): value = int(value.split('-')[1]) + 1 - self.assertEqual(psutil.cpu_count(), value) + self.assertEqual(psutil.cpu_count("logical"), value) @unittest.skipIf(not os.path.exists("/sys/devices/system/cpu"), "/sys/devices/system/cpu does not exist") def test_against_sysdev_cpu_num(self): ls = os.listdir("/sys/devices/system/cpu") count = len([x for x in ls if re.search(r"cpu\d+$", x) is not None]) - self.assertEqual(psutil.cpu_count(), count) + self.assertEqual(psutil.cpu_count("logical"), count) @unittest.skipIf(not which("nproc"), "nproc utility not available") def test_against_nproc(self): num = int(sh("nproc --all")) - self.assertEqual(psutil.cpu_count(logical=True), num) + self.assertEqual(psutil.cpu_count("logical"), num) @unittest.skipIf(not which("lscpu"), "lscpu utility not available") def test_against_lscpu(self): - out = sh("lscpu -p") - num = len([x for x in out.split('\n') if not x.startswith('#')]) - self.assertEqual(psutil.cpu_count(logical=True), num) + num = int(lscpu("cpu(s)")) + self.assertEqual(psutil.cpu_count("logical"), num) def test_emulate_fallbacks(self): import psutil._pslinux @@ -719,13 +706,8 @@ class TestSystemCPUCountCores(PsutilTestCase): @unittest.skipIf(not which("lscpu"), "lscpu utility not available") def test_against_lscpu(self): - out = sh("lscpu -p") - core_ids = set() - for line in out.split('\n'): - if not line.startswith('#'): - fields = line.split(',') - core_ids.add(fields[1]) - self.assertEqual(psutil.cpu_count(logical=False), len(core_ids)) + num = int(lscpu("core(s) per socket")) + self.assertEqual(psutil.cpu_count("cores"), num) def test_method_2(self): meth_1 = psutil._pslinux.cpu_count_cores() @@ -744,6 +726,50 @@ class TestSystemCPUCountCores(PsutilTestCase): @unittest.skipIf(not LINUX, "LINUX only") +class TestSystemCPUCountSockets(PsutilTestCase): + + @unittest.skipIf(not which("lscpu"), "lscpu utility not available") + def test_against_lscpu(self): + num = int(lscpu("socket(s)")) + self.assertEqual(psutil.cpu_count(kind="sockets"), num) + + +@unittest.skipIf(not LINUX, "LINUX only") +class TestSystemCPUCountNuma(PsutilTestCase): + + @unittest.skipIf(not which("lscpu"), "lscpu utility not available") + def test_against_lscpu(self): + num = int(lscpu("numa node(s)")) + self.assertEqual(psutil.cpu_count(kind="numa"), num) + + +# ===================================================================== +# --- system CPU (others) +# ===================================================================== + + +@unittest.skipIf(not LINUX, "LINUX only") +class TestSystemCPUTimes(PsutilTestCase): + + def test_fields(self): + fields = psutil.cpu_times()._fields + kernel_ver = re.findall(r'\d+\.\d+\.\d+', os.uname()[2])[0] + kernel_ver_info = tuple(map(int, kernel_ver.split('.'))) + if kernel_ver_info >= (2, 6, 11): + self.assertIn('steal', fields) + else: + self.assertNotIn('steal', fields) + if kernel_ver_info >= (2, 6, 24): + self.assertIn('guest', fields) + else: + self.assertNotIn('guest', fields) + if kernel_ver_info >= (3, 2, 0): + self.assertIn('guest_nice', fields) + else: + self.assertNotIn('guest_nice', fields) + + +@unittest.skipIf(not LINUX, "LINUX only") class TestSystemCPUFrequency(PsutilTestCase): @unittest.skipIf(not HAS_CPU_FREQ, "not supported") diff --git a/psutil/tests/test_memleaks.py b/psutil/tests/test_memleaks.py index e6940a30..38628478 100755 --- a/psutil/tests/test_memleaks.py +++ b/psutil/tests/test_memleaks.py @@ -342,13 +342,25 @@ class TestModuleFunctionsLeaks(TestMemoryLeak): # --- cpu + def test_cpu_count(self): + # here just to make ns.test_class_coverage happy + pass + @fewtimes_if_linux() - def test_cpu_count(self): # logical - self.execute(lambda: psutil.cpu_count(logical=True)) + def test_cpu_count_logical(self): + self.execute(lambda: psutil.cpu_count("logical")) @fewtimes_if_linux() def test_cpu_count_cores(self): - self.execute(lambda: psutil.cpu_count(logical=False)) + self.execute(lambda: psutil.cpu_count("cores")) + + @fewtimes_if_linux() + def test_cpu_count_sockets(self): + self.execute(lambda: psutil.cpu_count("sockets")) + + @fewtimes_if_linux() + def test_cpu_count_numa(self): + self.execute(lambda: psutil.cpu_count("numa")) @fewtimes_if_linux() def test_cpu_times(self): diff --git a/psutil/tests/test_osx.py b/psutil/tests/test_osx.py index b7a0b088..a3d97c31 100755 --- a/psutil/tests/test_osx.py +++ b/psutil/tests/test_osx.py @@ -22,12 +22,10 @@ from psutil.tests import TOLERANCE_SYS_MEM from psutil.tests import unittest -def sysctl(cmdline): - """Expects a sysctl command with an argument and parse the result - returning only the value of interest. - """ - out = sh(cmdline) - result = out.split()[1] +def sysctl(param): + """Expects a sysctl sub-command and return its stripped result.""" + out = sh("sysctl " + param) + result = out.partition(' ')[2] try: return int(result) except ValueError: @@ -134,21 +132,25 @@ class TestSystemAPIs(PsutilTestCase): # --- cpu def test_cpu_count_logical(self): - num = sysctl("sysctl hw.logicalcpu") - self.assertEqual(num, psutil.cpu_count(logical=True)) + num = sysctl("hw.logicalcpu") + self.assertEqual(num, psutil.cpu_count("logical")) def test_cpu_count_cores(self): - num = sysctl("sysctl hw.physicalcpu") - self.assertEqual(num, psutil.cpu_count(logical=False)) + num = sysctl("hw.physicalcpu") + self.assertEqual(num, psutil.cpu_count("cores")) + + def test_cpu_count_sockets(self): + num = sysctl("hw.packages") + self.assertEqual(num, psutil.cpu_count("sockets")) def test_cpu_freq(self): freq = psutil.cpu_freq() self.assertEqual( - freq.current * 1000 * 1000, sysctl("sysctl hw.cpufrequency")) + freq.current * 1000 * 1000, sysctl("hw.cpufrequency")) self.assertEqual( - freq.min * 1000 * 1000, sysctl("sysctl hw.cpufrequency_min")) + freq.min * 1000 * 1000, sysctl("hw.cpufrequency_min")) self.assertEqual( - freq.max * 1000 * 1000, sysctl("sysctl hw.cpufrequency_max")) + freq.max * 1000 * 1000, sysctl("hw.cpufrequency_max")) # --- virtual mem diff --git a/psutil/tests/test_system.py b/psutil/tests/test_system.py index 4e3ac3e4..145d0ffe 100755 --- a/psutil/tests/test_system.py +++ b/psutil/tests/test_system.py @@ -16,6 +16,7 @@ import signal import socket import sys import time +import warnings import psutil from psutil import AIX @@ -302,23 +303,69 @@ class TestMemoryAPIs(PsutilTestCase): assert mem.sout >= 0, mem -class TestCpuAPIs(PsutilTestCase): - - def test_cpu_count_logical(self): - logical = psutil.cpu_count() - self.assertIsNotNone(logical) - self.assertEqual(logical, len(psutil.cpu_times(percpu=True))) - self.assertGreaterEqual(logical, 1) - # - if os.path.exists("/proc/cpuinfo"): - with open("/proc/cpuinfo") as fd: - cpuinfo_data = fd.read() - if "physical id" not in cpuinfo_data: - raise unittest.SkipTest("cpuinfo doesn't include physical id") - - def test_cpu_count_cores(self): +class TestCpuCount(PsutilTestCase): + + def test_base(self): + kinds = ("logical", "cores", "sockets", "numa") + for kind in kinds: + n = psutil.cpu_count(kind=kind) + if n is not None: + self.assertIsInstance(n, int) + with self.subTest(kind): + self.assertGreaterEqual(n, 1) + + self.assertEqual(psutil.cpu_count(), psutil.cpu_count("logical")) + + with self.assertRaises(ValueError) as cm: + psutil.cpu_count(kind='xxx') + self.assertIn(str(kinds), str(cm.exception)) + + def test_consistency(self): + logical = psutil.cpu_count("logical") + cores = psutil.cpu_count("cores") + sockets = psutil.cpu_count("sockets") + numa = psutil.cpu_count(kind='numa') + + # logical (always supposed to be the highest) + if logical is not None: + if cores is not None: + self.assertGreaterEqual(logical, cores) + if sockets is not None: + self.assertGreaterEqual(logical, sockets) + + # cores (at least >= sockets) + if cores is not None: + if sockets is not None: + self.assertGreaterEqual(cores, sockets) + + if numa is not None: + self.assertGreaterEqual(numa, 0) + + def test_deprecation(self): + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + self.assertEqual(psutil.cpu_count(logical=True), + psutil.cpu_count(kind="logical")) + self.assertEqual(psutil.cpu_count(logical=False), + psutil.cpu_count(kind="cores")) + self.assertEqual(psutil.cpu_count(True), + psutil.cpu_count("logical")) + self.assertEqual(psutil.cpu_count(False), + psutil.cpu_count("cores")) + + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + psutil.cpu_count(logical=True) + assert ws + + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + psutil.cpu_count(True) + assert ws + + def test_cores(self): logical = psutil.cpu_count() - cores = psutil.cpu_count(logical=False) + cores = psutil.cpu_count("cores") if cores is None: raise self.skipTest("cpu_count_cores() is None") if WINDOWS and sys.getwindowsversion()[:2] <= (6, 1): # <= Vista @@ -327,19 +374,18 @@ class TestCpuAPIs(PsutilTestCase): self.assertGreaterEqual(cores, 1) self.assertGreaterEqual(logical, cores) - def test_cpu_count_none(self): + def test_return_none(self): # https://github.com/giampaolo/psutil/issues/1085 for val in (-1, 0, None): with mock.patch('psutil._psplatform.cpu_count_logical', return_value=val) as m: self.assertIsNone(psutil.cpu_count()) assert m.called - with mock.patch('psutil._psplatform.cpu_count_cores', - return_value=val) as m: - self.assertIsNone(psutil.cpu_count(logical=False)) - assert m.called - def test_cpu_times(self): + +class TestCpuTimes(PsutilTestCase): + + def test_base(self): # Check type, value >= 0, str(). total = 0 times = psutil.cpu_times() @@ -368,7 +414,7 @@ class TestCpuAPIs(PsutilTestCase): # msg="%s %s" % (new_t, last_t)) # last = new - def test_cpu_times_time_increases(self): + def test_time_increases(self): # Make sure time increases between calls. t1 = sum(psutil.cpu_times()) stop_at = time.time() + GLOBAL_TIMEOUT @@ -436,6 +482,9 @@ class TestCpuAPIs(PsutilTestCase): self.assertAlmostEqual( getattr(base, field), getattr(summed_values, field), delta=1) + +class TestCpuPercent(PsutilTestCase): + def _test_cpu_percent(self, percent, last_ret, new_ret): try: self.assertIsInstance(percent, float) @@ -498,6 +547,9 @@ class TestCpuAPIs(PsutilTestCase): for percent in cpu: self._test_cpu_percent(percent, None, None) + +class TestOtherCPUFunctions(PsutilTestCase): + def test_cpu_stats(self): # Tested more extensively in per-platform test modules. infos = psutil.cpu_stats() diff --git a/psutil/tests/test_windows.py b/psutil/tests/test_windows.py index aeb282c8..815771bf 100755 --- a/psutil/tests/test_windows.py +++ b/psutil/tests/test_windows.py @@ -82,28 +82,29 @@ class TestCpuAPIs(WindowsTestCase): @unittest.skipIf('NUMBER_OF_PROCESSORS' not in os.environ, 'NUMBER_OF_PROCESSORS env var is not available') - def test_cpu_count_vs_NUMBER_OF_PROCESSORS(self): + def test_cpu_count_logical_vs_NUMBER_OF_PROCESSORS(self): # Will likely fail on many-cores systems: # https://stackoverflow.com/questions/31209256 num_cpus = int(os.environ['NUMBER_OF_PROCESSORS']) - self.assertEqual(num_cpus, psutil.cpu_count()) + self.assertEqual(num_cpus, psutil.cpu_count("logical")) - def test_cpu_count_vs_GetSystemInfo(self): + def test_cpu_count_logical_vs_GetSystemInfo(self): # Will likely fail on many-cores systems: # https://stackoverflow.com/questions/31209256 sys_value = win32api.GetSystemInfo()[5] - psutil_value = psutil.cpu_count() + psutil_value = psutil.cpu_count("logical") self.assertEqual(sys_value, psutil_value) def test_cpu_count_logical_vs_wmi(self): w = wmi.WMI() proc = w.Win32_Processor()[0] - self.assertEqual(psutil.cpu_count(), proc.NumberOfLogicalProcessors) + self.assertEqual(psutil.cpu_count("logical"), + proc.NumberOfLogicalProcessors) def test_cpu_count_cores_vs_wmi(self): w = wmi.WMI() proc = w.Win32_Processor()[0] - self.assertEqual(psutil.cpu_count(logical=False), proc.NumberOfCores) + self.assertEqual(psutil.cpu_count("cores"), proc.NumberOfCores) def test_cpu_count_vs_cpu_times(self): self.assertEqual(psutil.cpu_count(), @@ -725,7 +726,7 @@ class RemoteProcessTestCase(PsutilTestCase): p = psutil.Process(self.proc32.pid) e = p.environ() self.assertIn("THINK_OF_A_NUMBER", e) - self.assertEquals(e["THINK_OF_A_NUMBER"], str(os.getpid())) + self.assertEqual(e["THINK_OF_A_NUMBER"], str(os.getpid())) def test_environ_64(self): p = psutil.Process(self.proc64.pid) |