aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/psutil/tests/test_system.py
diff options
context:
space:
mode:
authorS. Solomon Darnell2025-03-28 21:52:21 -0500
committerS. Solomon Darnell2025-03-28 21:52:21 -0500
commit4a52a71956a8d46fcb7294ac71734504bb09bcc2 (patch)
treeee3dc5af3b6313e921cd920906356f5d4febc4ed /.venv/lib/python3.12/site-packages/psutil/tests/test_system.py
parentcc961e04ba734dd72309fb548a2f97d67d578813 (diff)
downloadgn-ai-master.tar.gz
two version of R2R are hereHEADmaster
Diffstat (limited to '.venv/lib/python3.12/site-packages/psutil/tests/test_system.py')
-rw-r--r--.venv/lib/python3.12/site-packages/psutil/tests/test_system.py985
1 files changed, 985 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/psutil/tests/test_system.py b/.venv/lib/python3.12/site-packages/psutil/tests/test_system.py
new file mode 100644
index 00000000..0b69ada7
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/psutil/tests/test_system.py
@@ -0,0 +1,985 @@
+#!/usr/bin/env python3
+
+# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Tests for system APIS."""
+
+import contextlib
+import datetime
+import errno
+import os
+import platform
+import pprint
+import shutil
+import signal
+import socket
+import sys
+import time
+
+import psutil
+from psutil import AIX
+from psutil import BSD
+from psutil import FREEBSD
+from psutil import LINUX
+from psutil import MACOS
+from psutil import NETBSD
+from psutil import OPENBSD
+from psutil import POSIX
+from psutil import SUNOS
+from psutil import WINDOWS
+from psutil._compat import PY3
+from psutil._compat import FileNotFoundError
+from psutil._compat import long
+from psutil.tests import ASCII_FS
+from psutil.tests import CI_TESTING
+from psutil.tests import DEVNULL
+from psutil.tests import GITHUB_ACTIONS
+from psutil.tests import GLOBAL_TIMEOUT
+from psutil.tests import HAS_BATTERY
+from psutil.tests import HAS_CPU_FREQ
+from psutil.tests import HAS_GETLOADAVG
+from psutil.tests import HAS_NET_IO_COUNTERS
+from psutil.tests import HAS_SENSORS_BATTERY
+from psutil.tests import HAS_SENSORS_FANS
+from psutil.tests import HAS_SENSORS_TEMPERATURES
+from psutil.tests import IS_64BIT
+from psutil.tests import MACOS_12PLUS
+from psutil.tests import PYPY
+from psutil.tests import QEMU_USER
+from psutil.tests import UNICODE_SUFFIX
+from psutil.tests import PsutilTestCase
+from psutil.tests import check_net_address
+from psutil.tests import enum
+from psutil.tests import mock
+from psutil.tests import pytest
+from psutil.tests import retry_on_failure
+
+
+# ===================================================================
+# --- System-related API tests
+# ===================================================================
+
+
+class TestProcessIter(PsutilTestCase):
+ def test_pid_presence(self):
+ assert os.getpid() in [x.pid for x in psutil.process_iter()]
+ sproc = self.spawn_testproc()
+ assert sproc.pid in [x.pid for x in psutil.process_iter()]
+ p = psutil.Process(sproc.pid)
+ p.kill()
+ p.wait()
+ assert sproc.pid not in [x.pid for x in psutil.process_iter()]
+
+ def test_no_duplicates(self):
+ ls = [x for x in psutil.process_iter()]
+ assert sorted(ls, key=lambda x: x.pid) == sorted(
+ set(ls), key=lambda x: x.pid
+ )
+
+ def test_emulate_nsp(self):
+ list(psutil.process_iter()) # populate cache
+ for x in range(2):
+ with mock.patch(
+ 'psutil.Process.as_dict',
+ side_effect=psutil.NoSuchProcess(os.getpid()),
+ ):
+ assert list(psutil.process_iter(attrs=["cpu_times"])) == []
+ psutil.process_iter.cache_clear() # repeat test without cache
+
+ def test_emulate_access_denied(self):
+ list(psutil.process_iter()) # populate cache
+ for x in range(2):
+ with mock.patch(
+ 'psutil.Process.as_dict',
+ side_effect=psutil.AccessDenied(os.getpid()),
+ ):
+ with pytest.raises(psutil.AccessDenied):
+ list(psutil.process_iter(attrs=["cpu_times"]))
+ psutil.process_iter.cache_clear() # repeat test without cache
+
+ def test_attrs(self):
+ for p in psutil.process_iter(attrs=['pid']):
+ assert list(p.info.keys()) == ['pid']
+ # yield again
+ for p in psutil.process_iter(attrs=['pid']):
+ assert list(p.info.keys()) == ['pid']
+ with pytest.raises(ValueError):
+ list(psutil.process_iter(attrs=['foo']))
+ with mock.patch(
+ "psutil._psplatform.Process.cpu_times",
+ side_effect=psutil.AccessDenied(0, ""),
+ ) as m:
+ for p in psutil.process_iter(attrs=["pid", "cpu_times"]):
+ assert p.info['cpu_times'] is None
+ assert p.info['pid'] >= 0
+ assert m.called
+ with mock.patch(
+ "psutil._psplatform.Process.cpu_times",
+ side_effect=psutil.AccessDenied(0, ""),
+ ) as m:
+ flag = object()
+ for p in psutil.process_iter(
+ attrs=["pid", "cpu_times"], ad_value=flag
+ ):
+ assert p.info['cpu_times'] is flag
+ assert p.info['pid'] >= 0
+ assert m.called
+
+ def test_cache_clear(self):
+ list(psutil.process_iter()) # populate cache
+ assert psutil._pmap
+ psutil.process_iter.cache_clear()
+ assert not psutil._pmap
+
+
+class TestProcessAPIs(PsutilTestCase):
+ @pytest.mark.skipif(
+ PYPY and WINDOWS,
+ reason="spawn_testproc() unreliable on PYPY + WINDOWS",
+ )
+ def test_wait_procs(self):
+ def callback(p):
+ pids.append(p.pid)
+
+ pids = []
+ sproc1 = self.spawn_testproc()
+ sproc2 = self.spawn_testproc()
+ sproc3 = self.spawn_testproc()
+ procs = [psutil.Process(x.pid) for x in (sproc1, sproc2, sproc3)]
+ with pytest.raises(ValueError):
+ psutil.wait_procs(procs, timeout=-1)
+ with pytest.raises(TypeError):
+ psutil.wait_procs(procs, callback=1)
+ t = time.time()
+ gone, alive = psutil.wait_procs(procs, timeout=0.01, callback=callback)
+
+ assert time.time() - t < 0.5
+ assert gone == []
+ assert len(alive) == 3
+ assert pids == []
+ for p in alive:
+ assert not hasattr(p, 'returncode')
+
+ @retry_on_failure(30)
+ def test_1(procs, callback):
+ gone, alive = psutil.wait_procs(
+ procs, timeout=0.03, callback=callback
+ )
+ assert len(gone) == 1
+ assert len(alive) == 2
+ return gone, alive
+
+ sproc3.terminate()
+ gone, alive = test_1(procs, callback)
+ assert sproc3.pid in [x.pid for x in gone]
+ if POSIX:
+ assert gone.pop().returncode == -signal.SIGTERM
+ else:
+ assert gone.pop().returncode == 1
+ assert pids == [sproc3.pid]
+ for p in alive:
+ assert not hasattr(p, 'returncode')
+
+ @retry_on_failure(30)
+ def test_2(procs, callback):
+ gone, alive = psutil.wait_procs(
+ procs, timeout=0.03, callback=callback
+ )
+ assert len(gone) == 3
+ assert len(alive) == 0
+ return gone, alive
+
+ sproc1.terminate()
+ sproc2.terminate()
+ gone, alive = test_2(procs, callback)
+ assert set(pids) == set([sproc1.pid, sproc2.pid, sproc3.pid])
+ for p in gone:
+ assert hasattr(p, 'returncode')
+
+ @pytest.mark.skipif(
+ PYPY and WINDOWS,
+ reason="spawn_testproc() unreliable on PYPY + WINDOWS",
+ )
+ def test_wait_procs_no_timeout(self):
+ sproc1 = self.spawn_testproc()
+ sproc2 = self.spawn_testproc()
+ sproc3 = self.spawn_testproc()
+ procs = [psutil.Process(x.pid) for x in (sproc1, sproc2, sproc3)]
+ for p in procs:
+ p.terminate()
+ psutil.wait_procs(procs)
+
+ def test_pid_exists(self):
+ sproc = self.spawn_testproc()
+ assert psutil.pid_exists(sproc.pid)
+ p = psutil.Process(sproc.pid)
+ p.kill()
+ p.wait()
+ assert not psutil.pid_exists(sproc.pid)
+ assert not psutil.pid_exists(-1)
+ assert psutil.pid_exists(0) == (0 in psutil.pids())
+
+ def test_pid_exists_2(self):
+ 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(0.1)
+ assert pid not in psutil.pids()
+ pids = range(max(pids) + 15000, max(pids) + 16000)
+ for pid in pids:
+ assert not psutil.pid_exists(pid)
+
+
+class TestMiscAPIs(PsutilTestCase):
+ def test_boot_time(self):
+ bt = psutil.boot_time()
+ assert isinstance(bt, float)
+ assert bt > 0
+ assert bt < time.time()
+
+ @pytest.mark.skipif(
+ CI_TESTING and not psutil.users(), reason="unreliable on CI"
+ )
+ def test_users(self):
+ users = psutil.users()
+ assert users != []
+ for user in users:
+ with self.subTest(user=user):
+ assert user.name
+ assert isinstance(user.name, str)
+ assert isinstance(user.terminal, (str, type(None)))
+ if user.host is not None:
+ assert isinstance(user.host, (str, type(None)))
+ user.terminal # noqa
+ user.host # noqa
+ assert user.started > 0.0
+ datetime.datetime.fromtimestamp(user.started)
+ if WINDOWS or OPENBSD:
+ assert user.pid is None
+ else:
+ psutil.Process(user.pid)
+
+ 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:
+ assert isinstance(getattr(psutil, name), bool), 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
+ assert [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:
+ assert not getattr(psutil, name), name
+
+
+class TestMemoryAPIs(PsutilTestCase):
+ def test_virtual_memory(self):
+ mem = psutil.virtual_memory()
+ assert mem.total > 0, mem
+ assert mem.available > 0, mem
+ assert 0 <= mem.percent <= 100, mem
+ assert mem.used > 0, mem
+ assert mem.free >= 0, mem
+ for name in mem._fields:
+ value = getattr(mem, name)
+ if name != 'percent':
+ assert isinstance(value, (int, long))
+ if name != 'total':
+ if not value >= 0:
+ raise self.fail("%r < 0 (%s)" % (name, value))
+ if value > mem.total:
+ raise self.fail(
+ "%r > total (total=%s, %s=%s)"
+ % (name, mem.total, name, value)
+ )
+
+ def test_swap_memory(self):
+ mem = psutil.swap_memory()
+ assert mem._fields == (
+ 'total',
+ 'used',
+ 'free',
+ 'percent',
+ 'sin',
+ 'sout',
+ )
+
+ assert mem.total >= 0, mem
+ assert mem.used >= 0, mem
+ if mem.total > 0:
+ # likely a system with no swap partition
+ assert mem.free > 0, mem
+ else:
+ assert mem.free == 0, mem
+ assert 0 <= mem.percent <= 100, mem
+ assert mem.sin >= 0, mem
+ assert mem.sout >= 0, mem
+
+
+class TestCpuAPIs(PsutilTestCase):
+ def test_cpu_count_logical(self):
+ logical = psutil.cpu_count()
+ assert logical is not None
+ assert logical == len(psutil.cpu_times(percpu=True))
+ assert 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 pytest.skip("cpuinfo doesn't include physical id")
+
+ def test_cpu_count_cores(self):
+ logical = psutil.cpu_count()
+ cores = psutil.cpu_count(logical=False)
+ if cores is None:
+ raise pytest.skip("cpu_count_cores() is None")
+ if WINDOWS and sys.getwindowsversion()[:2] <= (6, 1): # <= Vista
+ assert cores is None
+ else:
+ assert cores >= 1
+ assert logical >= cores
+
+ def test_cpu_count_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:
+ assert psutil.cpu_count() is None
+ assert m.called
+ with mock.patch(
+ 'psutil._psplatform.cpu_count_cores', return_value=val
+ ) as m:
+ assert psutil.cpu_count(logical=False) is None
+ assert m.called
+
+ def test_cpu_times(self):
+ # Check type, value >= 0, str().
+ total = 0
+ times = psutil.cpu_times()
+ sum(times)
+ for cp_time in times:
+ assert isinstance(cp_time, float)
+ assert cp_time >= 0.0
+ total += cp_time
+ assert round(abs(total - sum(times)), 6) == 0
+ str(times)
+ # CPU times are always supposed to increase over time
+ # or at least remain the same and that's because time
+ # cannot go backwards.
+ # Surprisingly sometimes this might not be the case (at
+ # least on Windows and Linux), see:
+ # https://github.com/giampaolo/psutil/issues/392
+ # https://github.com/giampaolo/psutil/issues/645
+ # if not WINDOWS:
+ # last = psutil.cpu_times()
+ # for x in range(100):
+ # new = psutil.cpu_times()
+ # for field in new._fields:
+ # new_t = getattr(new, field)
+ # last_t = getattr(last, field)
+ # self.assertGreaterEqual(new_t, last_t,
+ # msg="%s %s" % (new_t, last_t))
+ # last = new
+
+ def test_cpu_times_time_increases(self):
+ # Make sure time increases between calls.
+ t1 = sum(psutil.cpu_times())
+ stop_at = time.time() + GLOBAL_TIMEOUT
+ while time.time() < stop_at:
+ t2 = sum(psutil.cpu_times())
+ if t2 > t1:
+ return
+ raise self.fail("time remained the same")
+
+ def test_per_cpu_times(self):
+ # Check type, value >= 0, str().
+ for times in psutil.cpu_times(percpu=True):
+ total = 0
+ sum(times)
+ for cp_time in times:
+ assert isinstance(cp_time, float)
+ assert cp_time >= 0.0
+ total += cp_time
+ assert round(abs(total - sum(times)), 6) == 0
+ str(times)
+ assert len(psutil.cpu_times(percpu=True)[0]) == len(
+ psutil.cpu_times(percpu=False)
+ )
+
+ # Note: in theory CPU times are always supposed to increase over
+ # time or remain the same but never go backwards. In practice
+ # sometimes this is not the case.
+ # This issue seemd to be afflict Windows:
+ # https://github.com/giampaolo/psutil/issues/392
+ # ...but it turns out also Linux (rarely) behaves the same.
+ # last = psutil.cpu_times(percpu=True)
+ # for x in range(100):
+ # new = psutil.cpu_times(percpu=True)
+ # for index in range(len(new)):
+ # newcpu = new[index]
+ # lastcpu = last[index]
+ # for field in newcpu._fields:
+ # new_t = getattr(newcpu, field)
+ # last_t = getattr(lastcpu, field)
+ # self.assertGreaterEqual(
+ # new_t, last_t, msg="%s %s" % (lastcpu, newcpu))
+ # last = new
+
+ def test_per_cpu_times_2(self):
+ # Simulate some work load then make sure time have increased
+ # between calls.
+ tot1 = psutil.cpu_times(percpu=True)
+ giveup_at = time.time() + GLOBAL_TIMEOUT
+ while True:
+ if time.time() >= giveup_at:
+ return self.fail("timeout")
+ tot2 = psutil.cpu_times(percpu=True)
+ for t1, t2 in zip(tot1, tot2):
+ t1, t2 = psutil._cpu_busy_time(t1), psutil._cpu_busy_time(t2)
+ difference = t2 - t1
+ if difference >= 0.05:
+ return
+
+ @pytest.mark.skipif(
+ CI_TESTING and OPENBSD, reason="unreliable on OPENBSD + CI"
+ )
+ def test_cpu_times_comparison(self):
+ # Make sure the sum of all per cpu times is almost equal to
+ # base "one cpu" times. On OpenBSD the sum of per-CPUs is
+ # higher for some reason.
+ base = psutil.cpu_times()
+ per_cpu = psutil.cpu_times(percpu=True)
+ summed_values = base._make([sum(num) for num in zip(*per_cpu)])
+ for field in base._fields:
+ with self.subTest(field=field, base=base, per_cpu=per_cpu):
+ assert (
+ abs(getattr(base, field) - getattr(summed_values, field))
+ < 1
+ )
+
+ def _test_cpu_percent(self, percent, last_ret, new_ret):
+ try:
+ assert isinstance(percent, float)
+ assert percent >= 0.0
+ assert percent <= 100.0 * psutil.cpu_count()
+ except AssertionError as err:
+ raise AssertionError(
+ "\n%s\nlast=%s\nnew=%s"
+ % (err, pprint.pformat(last_ret), pprint.pformat(new_ret))
+ )
+
+ def test_cpu_percent(self):
+ last = psutil.cpu_percent(interval=0.001)
+ for _ in range(100):
+ new = psutil.cpu_percent(interval=None)
+ self._test_cpu_percent(new, last, new)
+ last = new
+ with pytest.raises(ValueError):
+ psutil.cpu_percent(interval=-1)
+
+ def test_per_cpu_percent(self):
+ last = psutil.cpu_percent(interval=0.001, percpu=True)
+ assert len(last) == psutil.cpu_count()
+ for _ in range(100):
+ new = psutil.cpu_percent(interval=None, percpu=True)
+ for percent in new:
+ self._test_cpu_percent(percent, last, new)
+ last = new
+ with pytest.raises(ValueError):
+ psutil.cpu_percent(interval=-1, percpu=True)
+
+ def test_cpu_times_percent(self):
+ last = psutil.cpu_times_percent(interval=0.001)
+ for _ in range(100):
+ new = psutil.cpu_times_percent(interval=None)
+ for percent in new:
+ self._test_cpu_percent(percent, last, new)
+ self._test_cpu_percent(sum(new), last, new)
+ last = new
+ with pytest.raises(ValueError):
+ psutil.cpu_times_percent(interval=-1)
+
+ def test_per_cpu_times_percent(self):
+ last = psutil.cpu_times_percent(interval=0.001, percpu=True)
+ assert len(last) == psutil.cpu_count()
+ for _ in range(100):
+ new = psutil.cpu_times_percent(interval=None, percpu=True)
+ for cpu in new:
+ for percent in cpu:
+ self._test_cpu_percent(percent, last, new)
+ self._test_cpu_percent(sum(cpu), last, new)
+ last = new
+
+ def test_per_cpu_times_percent_negative(self):
+ # see: https://github.com/giampaolo/psutil/issues/645
+ psutil.cpu_times_percent(percpu=True)
+ zero_times = [
+ x._make([0 for x in range(len(x._fields))])
+ for x in psutil.cpu_times(percpu=True)
+ ]
+ with mock.patch('psutil.cpu_times', return_value=zero_times):
+ for cpu in psutil.cpu_times_percent(percpu=True):
+ 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()
+ assert infos._fields == (
+ 'ctx_switches',
+ 'interrupts',
+ 'soft_interrupts',
+ 'syscalls',
+ )
+ for name in infos._fields:
+ value = getattr(infos, name)
+ assert value >= 0
+ # on AIX, ctx_switches is always 0
+ if not AIX and name in {'ctx_switches', 'interrupts'}:
+ assert value > 0
+
+ # TODO: remove this once 1892 is fixed
+ @pytest.mark.skipif(
+ MACOS and platform.machine() == 'arm64', reason="skipped due to #1892"
+ )
+ @pytest.mark.skipif(not HAS_CPU_FREQ, reason="not supported")
+ def test_cpu_freq(self):
+ def check_ls(ls):
+ for nt in ls:
+ assert nt._fields == ('current', 'min', 'max')
+ if nt.max != 0.0:
+ assert nt.current <= nt.max
+ for name in nt._fields:
+ value = getattr(nt, name)
+ assert isinstance(value, (int, long, float))
+ assert value >= 0
+
+ ls = psutil.cpu_freq(percpu=True)
+ if FREEBSD and not ls:
+ raise pytest.skip("returns empty list on FreeBSD")
+
+ assert ls, ls
+ check_ls([psutil.cpu_freq(percpu=False)])
+
+ if LINUX:
+ assert len(ls) == psutil.cpu_count()
+
+ @pytest.mark.skipif(not HAS_GETLOADAVG, reason="not supported")
+ def test_getloadavg(self):
+ loadavg = psutil.getloadavg()
+ assert len(loadavg) == 3
+ for load in loadavg:
+ assert isinstance(load, float)
+ assert load >= 0.0
+
+
+class TestDiskAPIs(PsutilTestCase):
+ @pytest.mark.skipif(
+ PYPY and not IS_64BIT, reason="unreliable on PYPY32 + 32BIT"
+ )
+ def test_disk_usage(self):
+ usage = psutil.disk_usage(os.getcwd())
+ assert usage._fields == ('total', 'used', 'free', 'percent')
+
+ assert usage.total > 0, usage
+ assert usage.used > 0, usage
+ assert usage.free > 0, usage
+ assert usage.total > usage.used, usage
+ assert usage.total > usage.free, usage
+ assert 0 <= usage.percent <= 100, usage.percent
+ if hasattr(shutil, 'disk_usage'):
+ # py >= 3.3, see: http://bugs.python.org/issue12442
+ shutil_usage = shutil.disk_usage(os.getcwd())
+ tolerance = 5 * 1024 * 1024 # 5MB
+ assert usage.total == shutil_usage.total
+ assert abs(usage.free - shutil_usage.free) < tolerance
+ if not MACOS_12PLUS:
+ # see https://github.com/giampaolo/psutil/issues/2147
+ assert abs(usage.used - shutil_usage.used) < tolerance
+
+ # if path does not exist OSError ENOENT is expected across
+ # all platforms
+ fname = self.get_testfn()
+ with pytest.raises(FileNotFoundError):
+ psutil.disk_usage(fname)
+
+ @pytest.mark.skipif(not ASCII_FS, reason="not an ASCII fs")
+ def test_disk_usage_unicode(self):
+ # See: https://github.com/giampaolo/psutil/issues/416
+ with pytest.raises(UnicodeEncodeError):
+ psutil.disk_usage(UNICODE_SUFFIX)
+
+ def test_disk_usage_bytes(self):
+ psutil.disk_usage(b'.')
+
+ def test_disk_partitions(self):
+ def check_ntuple(nt):
+ assert isinstance(nt.device, str)
+ assert isinstance(nt.mountpoint, str)
+ assert isinstance(nt.fstype, str)
+ assert isinstance(nt.opts, str)
+
+ # all = False
+ ls = psutil.disk_partitions(all=False)
+ assert ls
+ for disk in ls:
+ check_ntuple(disk)
+ if WINDOWS and 'cdrom' in disk.opts:
+ continue
+ if not POSIX:
+ assert os.path.exists(disk.device), disk
+ else:
+ # we cannot make any assumption about this, see:
+ # http://goo.gl/p9c43
+ disk.device # noqa
+ # on modern systems mount points can also be files
+ assert os.path.exists(disk.mountpoint), disk
+ assert disk.fstype, disk
+
+ # all = True
+ ls = psutil.disk_partitions(all=True)
+ assert ls
+ for disk in psutil.disk_partitions(all=True):
+ check_ntuple(disk)
+ if not WINDOWS and disk.mountpoint:
+ try:
+ os.stat(disk.mountpoint)
+ except OSError as err:
+ if GITHUB_ACTIONS and MACOS and err.errno == errno.EIO:
+ continue
+ # http://mail.python.org/pipermail/python-dev/
+ # 2012-June/120787.html
+ if err.errno not in {errno.EPERM, errno.EACCES}:
+ raise
+ else:
+ assert os.path.exists(disk.mountpoint), disk
+
+ # ---
+
+ def find_mount_point(path):
+ path = os.path.abspath(path)
+ while not os.path.ismount(path):
+ path = os.path.dirname(path)
+ return path.lower()
+
+ mount = find_mount_point(__file__)
+ mounts = [
+ x.mountpoint.lower()
+ for x in psutil.disk_partitions(all=True)
+ if x.mountpoint
+ ]
+ assert mount in mounts
+
+ @pytest.mark.skipif(
+ LINUX and not os.path.exists('/proc/diskstats'),
+ reason="/proc/diskstats not available on this linux version",
+ )
+ @pytest.mark.skipif(
+ CI_TESTING and not psutil.disk_io_counters(), reason="unreliable on CI"
+ ) # no visible disks
+ def test_disk_io_counters(self):
+ def check_ntuple(nt):
+ assert nt[0] == nt.read_count
+ assert nt[1] == nt.write_count
+ assert nt[2] == nt.read_bytes
+ assert nt[3] == nt.write_bytes
+ if not (OPENBSD or NETBSD):
+ assert nt[4] == nt.read_time
+ assert nt[5] == nt.write_time
+ if LINUX:
+ assert nt[6] == nt.read_merged_count
+ assert nt[7] == nt.write_merged_count
+ assert nt[8] == nt.busy_time
+ elif FREEBSD:
+ assert 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
+ assert 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:
+ assert psutil.disk_io_counters(perdisk=False) is None
+ assert psutil.disk_io_counters(perdisk=True) == {}
+ assert m.called
+
+
+class TestNetAPIs(PsutilTestCase):
+ @pytest.mark.skipif(not HAS_NET_IO_COUNTERS, reason="not supported")
+ def test_net_io_counters(self):
+ def check_ntuple(nt):
+ assert nt[0] == nt.bytes_sent
+ assert nt[1] == nt.bytes_recv
+ assert nt[2] == nt.packets_sent
+ assert nt[3] == nt.packets_recv
+ assert nt[4] == nt.errin
+ assert nt[5] == nt.errout
+ assert nt[6] == nt.dropin
+ assert nt[7] == nt.dropout
+ assert nt.bytes_sent >= 0, nt
+ assert nt.bytes_recv >= 0, nt
+ assert nt.packets_sent >= 0, nt
+ assert nt.packets_recv >= 0, nt
+ assert nt.errin >= 0, nt
+ assert nt.errout >= 0, nt
+ assert nt.dropin >= 0, nt
+ assert nt.dropout >= 0, nt
+
+ ret = psutil.net_io_counters(pernic=False)
+ check_ntuple(ret)
+ ret = psutil.net_io_counters(pernic=True)
+ assert ret != []
+ for key in ret:
+ assert key
+ assert isinstance(key, str)
+ check_ntuple(ret[key])
+
+ @pytest.mark.skipif(not HAS_NET_IO_COUNTERS, reason="not supported")
+ def test_net_io_counters_no_nics(self):
+ # Emulate a case where no NICs are installed, see:
+ # https://github.com/giampaolo/psutil/issues/1062
+ with mock.patch(
+ 'psutil._psplatform.net_io_counters', return_value={}
+ ) as m:
+ assert psutil.net_io_counters(pernic=False) is None
+ assert psutil.net_io_counters(pernic=True) == {}
+ assert m.called
+
+ @pytest.mark.skipif(QEMU_USER, reason="QEMU user not supported")
+ def test_net_if_addrs(self):
+ nics = psutil.net_if_addrs()
+ assert nics, nics
+
+ nic_stats = psutil.net_if_stats()
+
+ # Not reliable on all platforms (net_if_addrs() reports more
+ # interfaces).
+ # self.assertEqual(sorted(nics.keys()),
+ # sorted(psutil.net_io_counters(pernic=True).keys()))
+
+ families = set([socket.AF_INET, socket.AF_INET6, psutil.AF_LINK])
+ for nic, addrs in nics.items():
+ assert isinstance(nic, str)
+ assert len(set(addrs)) == len(addrs)
+ for addr in addrs:
+ assert isinstance(addr.family, int)
+ assert isinstance(addr.address, str)
+ assert isinstance(addr.netmask, (str, type(None)))
+ assert isinstance(addr.broadcast, (str, type(None)))
+ assert addr.family in families
+ if PY3 and not PYPY:
+ assert isinstance(addr.family, enum.IntEnum)
+ if nic_stats[nic].isup:
+ # Do not test binding to addresses of interfaces
+ # that are down
+ if addr.family == socket.AF_INET:
+ s = socket.socket(addr.family)
+ with contextlib.closing(s):
+ s.bind((addr.address, 0))
+ elif addr.family == socket.AF_INET6:
+ info = socket.getaddrinfo(
+ addr.address,
+ 0,
+ socket.AF_INET6,
+ socket.SOCK_STREAM,
+ 0,
+ socket.AI_PASSIVE,
+ )[0]
+ af, socktype, proto, _canonname, sa = info
+ s = socket.socket(af, socktype, proto)
+ with contextlib.closing(s):
+ s.bind(sa)
+ for ip in (
+ addr.address,
+ addr.netmask,
+ addr.broadcast,
+ addr.ptp,
+ ):
+ if ip is not None:
+ # TODO: skip AF_INET6 for now because I get:
+ # AddressValueError: Only hex digits permitted in
+ # u'c6f3%lxcbr0' in u'fe80::c8e0:fff:fe54:c6f3%lxcbr0'
+ if addr.family != socket.AF_INET6:
+ check_net_address(ip, addr.family)
+ # broadcast and ptp addresses are mutually exclusive
+ if addr.broadcast:
+ assert addr.ptp is None
+ elif addr.ptp:
+ assert addr.broadcast is None
+
+ if BSD or MACOS or SUNOS:
+ if hasattr(socket, "AF_LINK"):
+ assert psutil.AF_LINK == socket.AF_LINK
+ elif LINUX:
+ assert psutil.AF_LINK == socket.AF_PACKET
+ elif WINDOWS:
+ assert psutil.AF_LINK == -1
+
+ def test_net_if_addrs_mac_null_bytes(self):
+ # Simulate that the underlying C function returns an incomplete
+ # MAC address. psutil is supposed to fill it with null bytes.
+ # https://github.com/giampaolo/psutil/issues/786
+ if POSIX:
+ ret = [('em1', psutil.AF_LINK, '06:3d:29', None, None, None)]
+ else:
+ ret = [('em1', -1, '06-3d-29', None, None, None)]
+ with mock.patch(
+ 'psutil._psplatform.net_if_addrs', return_value=ret
+ ) as m:
+ addr = psutil.net_if_addrs()['em1'][0]
+ assert m.called
+ if POSIX:
+ assert addr.address == '06:3d:29:00:00:00'
+ else:
+ assert addr.address == '06-3d-29-00-00-00'
+
+ @pytest.mark.skipif(QEMU_USER, reason="QEMU user not supported")
+ def test_net_if_stats(self):
+ nics = psutil.net_if_stats()
+ assert nics, nics
+ all_duplexes = (
+ psutil.NIC_DUPLEX_FULL,
+ psutil.NIC_DUPLEX_HALF,
+ psutil.NIC_DUPLEX_UNKNOWN,
+ )
+ for name, stats in nics.items():
+ assert isinstance(name, str)
+ isup, duplex, speed, mtu, flags = stats
+ assert isinstance(isup, bool)
+ assert duplex in all_duplexes
+ assert duplex in all_duplexes
+ assert speed >= 0
+ assert mtu >= 0
+ assert isinstance(flags, str)
+
+ @pytest.mark.skipif(
+ not (LINUX or BSD or MACOS), reason="LINUX or BSD or MACOS specific"
+ )
+ def test_net_if_stats_enodev(self):
+ # See: https://github.com/giampaolo/psutil/issues/1279
+ with mock.patch(
+ 'psutil._psutil_posix.net_if_mtu',
+ side_effect=OSError(errno.ENODEV, ""),
+ ) as m:
+ ret = psutil.net_if_stats()
+ assert ret == {}
+ assert m.called
+
+
+class TestSensorsAPIs(PsutilTestCase):
+ @pytest.mark.skipif(not HAS_SENSORS_TEMPERATURES, reason="not supported")
+ def test_sensors_temperatures(self):
+ temps = psutil.sensors_temperatures()
+ for name, entries in temps.items():
+ assert isinstance(name, str)
+ for entry in entries:
+ assert isinstance(entry.label, str)
+ if entry.current is not None:
+ assert entry.current >= 0
+ if entry.high is not None:
+ assert entry.high >= 0
+ if entry.critical is not None:
+ assert entry.critical >= 0
+
+ @pytest.mark.skipif(not HAS_SENSORS_TEMPERATURES, reason="not supported")
+ def test_sensors_temperatures_fahreneit(self):
+ d = {'coretemp': [('label', 50.0, 60.0, 70.0)]}
+ with mock.patch(
+ "psutil._psplatform.sensors_temperatures", return_value=d
+ ) as m:
+ temps = psutil.sensors_temperatures(fahrenheit=True)['coretemp'][0]
+ assert m.called
+ assert temps.current == 122.0
+ assert temps.high == 140.0
+ assert temps.critical == 158.0
+
+ @pytest.mark.skipif(not HAS_SENSORS_BATTERY, reason="not supported")
+ @pytest.mark.skipif(not HAS_BATTERY, reason="no battery")
+ def test_sensors_battery(self):
+ ret = psutil.sensors_battery()
+ assert ret.percent >= 0
+ assert ret.percent <= 100
+ if ret.secsleft not in {
+ psutil.POWER_TIME_UNKNOWN,
+ psutil.POWER_TIME_UNLIMITED,
+ }:
+ assert ret.secsleft >= 0
+ elif ret.secsleft == psutil.POWER_TIME_UNLIMITED:
+ assert ret.power_plugged
+ assert isinstance(ret.power_plugged, bool)
+
+ @pytest.mark.skipif(not HAS_SENSORS_FANS, reason="not supported")
+ def test_sensors_fans(self):
+ fans = psutil.sensors_fans()
+ for name, entries in fans.items():
+ assert isinstance(name, str)
+ for entry in entries:
+ assert isinstance(entry.label, str)
+ assert isinstance(entry.current, (int, long))
+ assert entry.current >= 0