summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorGiampaolo Rodola <g.rodola@gmail.com>2020-02-14 03:24:03 +0100
committerGiampaolo Rodola <g.rodola@gmail.com>2020-02-14 03:24:03 +0100
commit8bc5a180e96445fcd9e96fa6aebf8bc698910d96 (patch)
tree9f2c04e53d3d0698a79885676a42ed40508f76e5
parent17843980f8edf665915f09cc9abfa3a32c322fbf (diff)
downloadpsutil-8bc5a180e96445fcd9e96fa6aebf8bc698910d96.tar.gz
divide test_system.py unit tests in multiple classes
-rwxr-xr-xpsutil/tests/test_system.py383
1 files changed, 200 insertions, 183 deletions
diff --git a/psutil/tests/test_system.py b/psutil/tests/test_system.py
index 2d606be4..0d3f4375 100755
--- a/psutil/tests/test_system.py
+++ b/psutil/tests/test_system.py
@@ -60,8 +60,7 @@ from psutil.tests import unittest
# ===================================================================
-class TestSystemAPIs(unittest.TestCase):
- """Tests for system-related APIs."""
+class TestProcessAPIs(unittest.TestCase):
def setUp(self):
safe_rmpath(TESTFN)
@@ -190,12 +189,67 @@ class TestSystemAPIs(unittest.TestCase):
p.terminate()
gone, alive = psutil.wait_procs(procs)
+ def test_pid_exists(self):
+ sproc = get_test_subprocess()
+ self.assertTrue(psutil.pid_exists(sproc.pid))
+ p = psutil.Process(sproc.pid)
+ p.kill()
+ p.wait()
+ self.assertFalse(psutil.pid_exists(sproc.pid))
+ self.assertFalse(psutil.pid_exists(-1))
+ self.assertEqual(psutil.pid_exists(0), 0 in psutil.pids())
+
+ def test_pid_exists_2(self):
+ reap_children()
+ pids = psutil.pids()
+ for pid in pids:
+ try:
+ assert psutil.pid_exists(pid)
+ except AssertionError:
+ # in case the process disappeared in meantime fail only
+ # if it is no longer in psutil.pids()
+ time.sleep(.1)
+ if pid in psutil.pids():
+ self.fail(pid)
+ pids = range(max(pids) + 5000, max(pids) + 6000)
+ for pid in pids:
+ self.assertFalse(psutil.pid_exists(pid), msg=pid)
+
+ def test_pids(self):
+ pidslist = psutil.pids()
+ procslist = [x.pid for x in psutil.process_iter()]
+ # make sure every pid is unique
+ self.assertEqual(sorted(set(pidslist)), pidslist)
+ self.assertEqual(pidslist, procslist)
+
+
+class TestMiscAPIs(unittest.TestCase):
+
def test_boot_time(self):
bt = psutil.boot_time()
self.assertIsInstance(bt, float)
self.assertGreater(bt, 0)
self.assertLess(bt, time.time())
+ @unittest.skipIf(CI_TESTING and not psutil.users(), "unreliable on CI")
+ def test_users(self):
+ users = psutil.users()
+ self.assertNotEqual(users, [])
+ for user in users:
+ assert user.name, user
+ self.assertIsInstance(user.name, str)
+ self.assertIsInstance(user.terminal, (str, type(None)))
+ if user.host is not None:
+ self.assertIsInstance(user.host, (str, type(None)))
+ user.terminal
+ user.host
+ assert user.started > 0.0, user
+ datetime.datetime.fromtimestamp(user.started)
+ if WINDOWS or OPENBSD:
+ self.assertIsNone(user.pid)
+ else:
+ psutil.Process(user.pid)
+
@unittest.skipIf(not POSIX, 'POSIX only')
def test_PAGESIZE(self):
# pagesize is used internally to perform different calculations
@@ -204,6 +258,55 @@ class TestSystemAPIs(unittest.TestCase):
import resource
self.assertEqual(os.sysconf("SC_PAGE_SIZE"), resource.getpagesize())
+ def test_test(self):
+ # test for psutil.test() function
+ stdout = sys.stdout
+ sys.stdout = DEVNULL
+ try:
+ psutil.test()
+ finally:
+ sys.stdout = stdout
+
+ def test_os_constants(self):
+ names = ["POSIX", "WINDOWS", "LINUX", "MACOS", "FREEBSD", "OPENBSD",
+ "NETBSD", "BSD", "SUNOS"]
+ for name in names:
+ self.assertIsInstance(getattr(psutil, name), bool, msg=name)
+
+ if os.name == 'posix':
+ assert psutil.POSIX
+ assert not psutil.WINDOWS
+ names.remove("POSIX")
+ if "linux" in sys.platform.lower():
+ assert psutil.LINUX
+ names.remove("LINUX")
+ elif "bsd" in sys.platform.lower():
+ assert psutil.BSD
+ self.assertEqual([psutil.FREEBSD, psutil.OPENBSD,
+ psutil.NETBSD].count(True), 1)
+ names.remove("BSD")
+ names.remove("FREEBSD")
+ names.remove("OPENBSD")
+ names.remove("NETBSD")
+ elif "sunos" in sys.platform.lower() or \
+ "solaris" in sys.platform.lower():
+ assert psutil.SUNOS
+ names.remove("SUNOS")
+ elif "darwin" in sys.platform.lower():
+ assert psutil.MACOS
+ names.remove("MACOS")
+ else:
+ assert psutil.WINDOWS
+ assert not psutil.POSIX
+ names.remove("WINDOWS")
+
+ # assert all other constants are set to False
+ for name in names:
+ self.assertIs(getattr(psutil, name), False, msg=name)
+
+
+class TestMemoryAPIs(unittest.TestCase):
+
def test_virtual_memory(self):
mem = psutil.virtual_memory()
assert mem.total > 0, mem
@@ -238,47 +341,8 @@ class TestSystemAPIs(unittest.TestCase):
assert mem.sin >= 0, mem
assert mem.sout >= 0, mem
- def test_pid_exists(self):
- sproc = get_test_subprocess()
- self.assertTrue(psutil.pid_exists(sproc.pid))
- p = psutil.Process(sproc.pid)
- p.kill()
- p.wait()
- self.assertFalse(psutil.pid_exists(sproc.pid))
- self.assertFalse(psutil.pid_exists(-1))
- self.assertEqual(psutil.pid_exists(0), 0 in psutil.pids())
- def test_pid_exists_2(self):
- reap_children()
- pids = psutil.pids()
- for pid in pids:
- try:
- assert psutil.pid_exists(pid)
- except AssertionError:
- # in case the process disappeared in meantime fail only
- # if it is no longer in psutil.pids()
- time.sleep(.1)
- if pid in psutil.pids():
- self.fail(pid)
- pids = range(max(pids) + 5000, max(pids) + 6000)
- for pid in pids:
- self.assertFalse(psutil.pid_exists(pid), msg=pid)
-
- def test_pids(self):
- pidslist = psutil.pids()
- procslist = [x.pid for x in psutil.process_iter()]
- # make sure every pid is unique
- self.assertEqual(sorted(set(pidslist)), pidslist)
- self.assertEqual(pidslist, procslist)
-
- def test_test(self):
- # test for psutil.test() function
- stdout = sys.stdout
- sys.stdout = DEVNULL
- try:
- psutil.test()
- finally:
- sys.stdout = stdout
+class TestCpuAPIs(unittest.TestCase):
def test_cpu_count_logical(self):
logical = psutil.cpu_count()
@@ -472,6 +536,55 @@ class TestSystemAPIs(unittest.TestCase):
for percent in cpu:
self._test_cpu_percent(percent, None, None)
+ def test_cpu_stats(self):
+ # Tested more extensively in per-platform test modules.
+ infos = psutil.cpu_stats()
+ self.assertEqual(
+ infos._fields,
+ ('ctx_switches', 'interrupts', 'soft_interrupts', 'syscalls'))
+ for name in infos._fields:
+ value = getattr(infos, name)
+ self.assertGreaterEqual(value, 0)
+ # on AIX, ctx_switches is always 0
+ if not AIX and name in ('ctx_switches', 'interrupts'):
+ self.assertGreater(value, 0)
+
+ @unittest.skipIf(not HAS_CPU_FREQ, "not suported")
+ def test_cpu_freq(self):
+ def check_ls(ls):
+ for nt in ls:
+ self.assertEqual(nt._fields, ('current', 'min', 'max'))
+ if nt.max != 0.0:
+ self.assertLessEqual(nt.current, nt.max)
+ for name in nt._fields:
+ value = getattr(nt, name)
+ self.assertIsInstance(value, (int, long, float))
+ self.assertGreaterEqual(value, 0)
+
+ ls = psutil.cpu_freq(percpu=True)
+ if TRAVIS and not ls:
+ raise self.skipTest("skipped on Travis")
+ if FREEBSD and not ls:
+ raise self.skipTest("returns empty list on FreeBSD")
+
+ assert ls, ls
+ check_ls([psutil.cpu_freq(percpu=False)])
+
+ if LINUX:
+ self.assertEqual(len(ls), psutil.cpu_count())
+
+ @unittest.skipIf(not HAS_GETLOADAVG, "not supported")
+ def test_getloadavg(self):
+ loadavg = psutil.getloadavg()
+ assert len(loadavg) == 3
+
+ for load in loadavg:
+ self.assertIsInstance(load, float)
+ self.assertGreaterEqual(load, 0.0)
+
+
+class TestDiskAPIs(unittest.TestCase):
+
def test_disk_usage(self):
usage = psutil.disk_usage(os.getcwd())
self.assertEqual(usage._fields, ('total', 'used', 'free', 'percent'))
@@ -562,6 +675,50 @@ class TestSystemAPIs(unittest.TestCase):
self.assertIn(mount, mounts)
psutil.disk_usage(mount)
+ @unittest.skipIf(LINUX and not os.path.exists('/proc/diskstats'),
+ '/proc/diskstats not available on this linux version')
+ @unittest.skipIf(CI_TESTING and not psutil.disk_io_counters(),
+ "unreliable on CI") # no visible disks
+ def test_disk_io_counters(self):
+ def check_ntuple(nt):
+ self.assertEqual(nt[0], nt.read_count)
+ self.assertEqual(nt[1], nt.write_count)
+ self.assertEqual(nt[2], nt.read_bytes)
+ self.assertEqual(nt[3], nt.write_bytes)
+ if not (OPENBSD or NETBSD):
+ self.assertEqual(nt[4], nt.read_time)
+ self.assertEqual(nt[5], nt.write_time)
+ if LINUX:
+ self.assertEqual(nt[6], nt.read_merged_count)
+ self.assertEqual(nt[7], nt.write_merged_count)
+ self.assertEqual(nt[8], nt.busy_time)
+ elif FREEBSD:
+ self.assertEqual(nt[6], nt.busy_time)
+ for name in nt._fields:
+ assert getattr(nt, name) >= 0, nt
+
+ ret = psutil.disk_io_counters(perdisk=False)
+ assert ret is not None, "no disks on this system?"
+ check_ntuple(ret)
+ ret = psutil.disk_io_counters(perdisk=True)
+ # make sure there are no duplicates
+ self.assertEqual(len(ret), len(set(ret)))
+ for key in ret:
+ assert key, key
+ check_ntuple(ret[key])
+
+ def test_disk_io_counters_no_disks(self):
+ # Emulate a case where no disks are installed, see:
+ # https://github.com/giampaolo/psutil/issues/1062
+ with mock.patch('psutil._psplatform.disk_io_counters',
+ return_value={}) as m:
+ self.assertIsNone(psutil.disk_io_counters(perdisk=False))
+ self.assertEqual(psutil.disk_io_counters(perdisk=True), {})
+ assert m.called
+
+
+class TestNetAPIs(unittest.TestCase):
+
@unittest.skipIf(not HAS_NET_IO_COUNTERS, 'not supported')
def test_net_io_counters(self):
def check_ntuple(nt):
@@ -704,148 +861,8 @@ class TestSystemAPIs(unittest.TestCase):
self.assertEqual(ret, {})
assert m.called
- @unittest.skipIf(LINUX and not os.path.exists('/proc/diskstats'),
- '/proc/diskstats not available on this linux version')
- @unittest.skipIf(CI_TESTING and not psutil.disk_io_counters(),
- "unreliable on CI") # no visible disks
- def test_disk_io_counters(self):
- def check_ntuple(nt):
- self.assertEqual(nt[0], nt.read_count)
- self.assertEqual(nt[1], nt.write_count)
- self.assertEqual(nt[2], nt.read_bytes)
- self.assertEqual(nt[3], nt.write_bytes)
- if not (OPENBSD or NETBSD):
- self.assertEqual(nt[4], nt.read_time)
- self.assertEqual(nt[5], nt.write_time)
- if LINUX:
- self.assertEqual(nt[6], nt.read_merged_count)
- self.assertEqual(nt[7], nt.write_merged_count)
- self.assertEqual(nt[8], nt.busy_time)
- elif FREEBSD:
- self.assertEqual(nt[6], nt.busy_time)
- for name in nt._fields:
- assert getattr(nt, name) >= 0, nt
-
- ret = psutil.disk_io_counters(perdisk=False)
- assert ret is not None, "no disks on this system?"
- check_ntuple(ret)
- ret = psutil.disk_io_counters(perdisk=True)
- # make sure there are no duplicates
- self.assertEqual(len(ret), len(set(ret)))
- for key in ret:
- assert key, key
- check_ntuple(ret[key])
- def test_disk_io_counters_no_disks(self):
- # Emulate a case where no disks are installed, see:
- # https://github.com/giampaolo/psutil/issues/1062
- with mock.patch('psutil._psplatform.disk_io_counters',
- return_value={}) as m:
- self.assertIsNone(psutil.disk_io_counters(perdisk=False))
- self.assertEqual(psutil.disk_io_counters(perdisk=True), {})
- assert m.called
-
- @unittest.skipIf(CI_TESTING and not psutil.users(), "unreliable on CI")
- def test_users(self):
- users = psutil.users()
- self.assertNotEqual(users, [])
- for user in users:
- assert user.name, user
- self.assertIsInstance(user.name, str)
- self.assertIsInstance(user.terminal, (str, type(None)))
- if user.host is not None:
- self.assertIsInstance(user.host, (str, type(None)))
- user.terminal
- user.host
- assert user.started > 0.0, user
- datetime.datetime.fromtimestamp(user.started)
- if WINDOWS or OPENBSD:
- self.assertIsNone(user.pid)
- else:
- psutil.Process(user.pid)
-
- def test_cpu_stats(self):
- # Tested more extensively in per-platform test modules.
- infos = psutil.cpu_stats()
- self.assertEqual(
- infos._fields,
- ('ctx_switches', 'interrupts', 'soft_interrupts', 'syscalls'))
- for name in infos._fields:
- value = getattr(infos, name)
- self.assertGreaterEqual(value, 0)
- # on AIX, ctx_switches is always 0
- if not AIX and name in ('ctx_switches', 'interrupts'):
- self.assertGreater(value, 0)
-
- @unittest.skipIf(not HAS_CPU_FREQ, "not suported")
- def test_cpu_freq(self):
- def check_ls(ls):
- for nt in ls:
- self.assertEqual(nt._fields, ('current', 'min', 'max'))
- if nt.max != 0.0:
- self.assertLessEqual(nt.current, nt.max)
- for name in nt._fields:
- value = getattr(nt, name)
- self.assertIsInstance(value, (int, long, float))
- self.assertGreaterEqual(value, 0)
-
- ls = psutil.cpu_freq(percpu=True)
- if TRAVIS and not ls:
- raise self.skipTest("skipped on Travis")
- if FREEBSD and not ls:
- raise self.skipTest("returns empty list on FreeBSD")
-
- assert ls, ls
- check_ls([psutil.cpu_freq(percpu=False)])
-
- if LINUX:
- self.assertEqual(len(ls), psutil.cpu_count())
-
- @unittest.skipIf(not HAS_GETLOADAVG, "not supported")
- def test_getloadavg(self):
- loadavg = psutil.getloadavg()
- assert len(loadavg) == 3
-
- for load in loadavg:
- self.assertIsInstance(load, float)
- self.assertGreaterEqual(load, 0.0)
-
- def test_os_constants(self):
- names = ["POSIX", "WINDOWS", "LINUX", "MACOS", "FREEBSD", "OPENBSD",
- "NETBSD", "BSD", "SUNOS"]
- for name in names:
- self.assertIsInstance(getattr(psutil, name), bool, msg=name)
-
- if os.name == 'posix':
- assert psutil.POSIX
- assert not psutil.WINDOWS
- names.remove("POSIX")
- if "linux" in sys.platform.lower():
- assert psutil.LINUX
- names.remove("LINUX")
- elif "bsd" in sys.platform.lower():
- assert psutil.BSD
- self.assertEqual([psutil.FREEBSD, psutil.OPENBSD,
- psutil.NETBSD].count(True), 1)
- names.remove("BSD")
- names.remove("FREEBSD")
- names.remove("OPENBSD")
- names.remove("NETBSD")
- elif "sunos" in sys.platform.lower() or \
- "solaris" in sys.platform.lower():
- assert psutil.SUNOS
- names.remove("SUNOS")
- elif "darwin" in sys.platform.lower():
- assert psutil.MACOS
- names.remove("MACOS")
- else:
- assert psutil.WINDOWS
- assert not psutil.POSIX
- names.remove("WINDOWS")
-
- # assert all other constants are set to False
- for name in names:
- self.assertIs(getattr(psutil, name), False, msg=name)
+class TestSensorsAPIs(unittest.TestCase):
@unittest.skipIf(not HAS_SENSORS_TEMPERATURES, "not supported")
def test_sensors_temperatures(self):