| #!/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 datetime |
| import enum |
| import errno |
| import os |
| import pprint |
| import shutil |
| import signal |
| import socket |
| import sys |
| import time |
| import warnings |
| from unittest import mock |
| |
| 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._common import broadcast_addr |
| |
| from . import AARCH64 |
| from . import ASCII_FS |
| from . import CI_TESTING |
| from . import GITHUB_ACTIONS |
| from . import GLOBAL_TIMEOUT |
| from . import HAS_BATTERY |
| from . import HAS_CPU_FREQ |
| from . import HAS_HEAP_INFO |
| from . import HAS_NET_IO_COUNTERS |
| from . import HAS_SENSORS_BATTERY |
| from . import HAS_SENSORS_FANS |
| from . import HAS_SENSORS_TEMPERATURES |
| from . import MACOS_12PLUS |
| from . import PYPY |
| from . import UNICODE_SUFFIX |
| from . import PsutilTestCase |
| from . import check_net_address |
| from . import pytest |
| from . 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_subproc() |
| 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 = list(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 not 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._prefetch.keys()) == ['pid'] |
| # yield again |
| for p in psutil.process_iter(attrs=['pid']): |
| assert list(p._prefetch.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._prefetch['cpu_times'] is None |
| assert p._prefetch['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._prefetch['cpu_times'] is flag |
| assert p._prefetch['pid'] >= 0 |
| assert m.called |
| |
| def test_prefetch(self): |
| for p in psutil.process_iter(attrs=["name", "status"]): |
| assert p.name() == p._prefetch["name"] |
| assert p.status() == p._prefetch["status"] |
| |
| def test_info_deprecation(self): |
| for p in psutil.process_iter(attrs=["name"]): |
| with warnings.catch_warnings(record=True) as ws: |
| warnings.simplefilter("always") |
| p.info # noqa: B018 |
| assert len(ws) == 1 |
| assert issubclass(ws[0].category, DeprecationWarning) |
| assert "deprecated" in str(ws[0].message) |
| break |
| |
| def test_prefetch_ad_value(self): |
| with mock.patch( |
| "psutil._psplatform.Process.cpu_times", |
| side_effect=psutil.AccessDenied(0, ""), |
| ): |
| for p in psutil.process_iter(attrs=["cpu_times"]): |
| assert p.cpu_times() is None |
| |
| def test_prefetch_cleared(self): |
| # Prefetch cache is cleared when attrs is not specified. |
| for p in psutil.process_iter(attrs=["name"]): |
| assert p._prefetch |
| for p in psutil.process_iter(): |
| assert not p._prefetch |
| |
| def test_prefetch_with_args_bypasses_cache(self): |
| # Methods called with args should bypass the prefetch cache |
| # (e.g. nice(10) is a setter, not a getter). |
| p = psutil.Process() |
| p._prefetch["nice"] = -999 |
| # Without args: returns cached value. |
| assert p.nice() == -999 |
| # With args: bypasses cache, calls real method. |
| real_nice = psutil.Process().nice() |
| assert p.nice(real_nice) is None |
| assert p.nice() == -999 |
| |
| def test_prefetch_double_call(self): |
| # Multiple calls should return the same cached value. |
| for p in psutil.process_iter(attrs=["name"]): |
| name1 = p.name() |
| name2 = p.name() |
| assert name1 == name2 |
| assert name1 == p._prefetch["name"] |
| break |
| |
| def test_deprecated_prefetch_empty_attrs(self): |
| # attrs=[] should prefetch all methods. |
| with pytest.warns(DeprecationWarning): |
| p = next(psutil.process_iter(attrs=[])) |
| assert p._prefetch.keys() == psutil.Process.attrs |
| |
| def test_prefetch_with_non_prefetched(self): |
| # A method not in attrs should still do a live syscall. |
| for p in psutil.process_iter(attrs=["name"]): |
| assert "status" not in p._prefetch |
| # status() should still work via syscall |
| assert p.status() |
| break |
| |
| def test_zombie_process_is_not_skipped(self): |
| # ZombieProcess is a subclass of NoSuchProcess; make sure |
| # process_iter() yields the process rather than removing it from |
| # the cache as if it had disappeared. |
| list(psutil.process_iter()) # populate the pmap cache |
| p = psutil._pmap[os.getpid()] |
| with mock.patch.object( |
| p, "as_dict", side_effect=psutil.ZombieProcess(p.pid) |
| ): |
| pids = [x.pid for x in psutil.process_iter(attrs=["name"])] |
| assert p.pid in pids |
| |
| 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_subproc() unreliable on PYPY + WINDOWS", |
| ) |
| def test_wait_procs(self): |
| def callback(p): |
| pids.append(p.pid) |
| |
| pids = [] |
| sproc1 = self.spawn_subproc() |
| sproc2 = self.spawn_subproc() |
| sproc3 = self.spawn_subproc() |
| 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 not gone |
| assert len(alive) == 3 |
| assert not 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) == {sproc1.pid, sproc2.pid, sproc3.pid} |
| for p in gone: |
| assert hasattr(p, 'returncode') |
| |
| @pytest.mark.skipif( |
| PYPY and WINDOWS, |
| reason="spawn_subproc() unreliable on PYPY + WINDOWS", |
| ) |
| def test_wait_procs_no_timeout(self): |
| sproc1 = self.spawn_subproc() |
| sproc2 = self.spawn_subproc() |
| sproc3 = self.spawn_subproc() |
| 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_subproc() |
| 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=str(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: B018 |
| user.host # noqa: B018 |
| assert user.started > 0.0 |
| datetime.datetime.fromtimestamp(user.started) |
| if WINDOWS or OPENBSD: |
| assert user.pid is None |
| else: |
| psutil.Process(user.pid) |
| |
| @pytest.mark.skipif(not HAS_HEAP_INFO, reason="not supported") |
| def test_heap_info(self): |
| m = psutil.heap_info() |
| assert m.heap_used > 0 |
| if MACOS: |
| assert m.mmap_used == 0 # not supported |
| else: |
| assert m.mmap_used > 0 |
| if WINDOWS: |
| assert m.heap_count >= 0 |
| |
| @pytest.mark.skipif(not HAS_HEAP_INFO, reason="not supported") |
| def test_heap_trim(self): |
| psutil.heap_trim() |
| |
| 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) |
| if name != 'total': |
| if not value >= 0: |
| return pytest.fail(f"{name!r} < 0 ({value})") |
| if value > mem.total: |
| return pytest.fail( |
| f"{name!r} > total (total={mem.total}, {name}={value})" |
| ) |
| |
| def test_virtual_memory_fields_order(self): |
| mem = psutil.virtual_memory() |
| common = ("total", "available", "percent", "used", "free") |
| assert mem._fields[:5] == common |
| if LINUX: |
| assert mem._fields[5:] == ( |
| "active", |
| "inactive", |
| "buffers", |
| "cached", |
| "shared", |
| "slab", |
| ) |
| elif MACOS: |
| assert mem._fields[5:] == ( |
| "active", |
| "inactive", |
| "wired", |
| ) |
| elif BSD: |
| assert mem._fields[5:] == ( |
| "active", |
| "inactive", |
| "buffers", |
| "cached", |
| "shared", |
| "wired", |
| ) |
| elif WINDOWS: |
| assert mem._fields[5:] == ( |
| "cached", |
| "wired", |
| ) |
| elif SUNOS or AIX: |
| assert mem._fields[5:] == () |
| |
| 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: |
| return 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: |
| return 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) |
| # assert 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 None |
| return pytest.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 seemed 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) |
| # assert new_t >= last_t |
| # 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 pytest.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 None |
| |
| @pytest.mark.skipif( |
| (CI_TESTING and OPENBSD) or MACOS or SUNOS, |
| reason="unreliable on OPENBSD + CI", |
| ) |
| @retry_on_failure(30) |
| 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=str(base), per_cpu=str(per_cpu) |
| ): |
| assert ( |
| abs(getattr(base, field) - getattr(summed_values, field)) |
| < 2 |
| ) |
| |
| 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{}\nlast={}\nnew={}".format( |
| 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 |
| |
| @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') |
| for name in nt._fields: |
| value = getattr(nt, name) |
| assert isinstance(value, (int, float)) |
| assert value >= 0 |
| |
| ls = psutil.cpu_freq(percpu=True) |
| if (FREEBSD or AARCH64) and not ls: |
| return pytest.skip( |
| "returns empty list on FreeBSD and Linux aarch64" |
| ) |
| |
| assert ls, ls |
| check_ls([psutil.cpu_freq(percpu=False)]) |
| |
| if LINUX: |
| assert len(ls) == psutil.cpu_count() |
| |
| 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): |
| 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 |
| |
| 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 FileNotFoundError is expected across |
| # all platforms |
| fname = self.get_testfn() |
| with pytest.raises(FileNotFoundError): |
| psutil.disk_usage(fname) |
| |
| # we should also be able to use a file path |
| psutil.disk_usage(__file__) |
| |
| @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: B018 |
| # 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 |
| |
| 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). |
| # assert sorted(nics.keys()) == sorted( |
| # psutil.net_io_counters(pernic=True).keys() |
| # ) |
| |
| families = { |
| socket.AF_INET, |
| socket.AF_INET6, |
| socket.AF_UNSPEC, |
| 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, type(None))) |
| if addr.address is None: # virtual NIC |
| assert addr.family == socket.AF_UNSPEC |
| assert isinstance(addr.netmask, (str, type(None))) |
| assert isinstance(addr.broadcast, (str, type(None))) |
| assert addr.family in families |
| 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: |
| with socket.socket(addr.family) as 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 |
| with socket.socket(af, socktype, proto) as 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 |
| |
| # check broadcast address |
| if ( |
| addr.broadcast |
| and addr.netmask |
| and addr.family in {socket.AF_INET, socket.AF_INET6} |
| ): |
| assert addr.broadcast == broadcast_addr(addr) |
| |
| 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' |
| |
| 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._psplatform.cext.net_if_mtu', |
| side_effect=OSError(errno.ENODEV, ""), |
| ) as m: |
| ret = psutil.net_if_stats() |
| assert ret == {} |
| assert m.called |
| |
| @pytest.mark.skipif(not POSIX, reason="POSIX only") |
| def test_nic_names(self): |
| stdlib_names = {name for _, name in socket.if_nameindex()} |
| assert stdlib_names == set(psutil.net_io_counters(pernic=True).keys()) |
| assert stdlib_names == set(psutil.net_if_addrs().keys()) |
| assert stdlib_names == set(psutil.net_if_stats().keys()) |
| |
| |
| 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) |
| assert entry.current >= 0 |
