summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorGiampaolo Rodola <g.rodola@gmail.com>2020-05-18 21:41:23 +0200
committerGiampaolo Rodola <g.rodola@gmail.com>2020-05-18 21:41:23 +0200
commit978296429c3eac20f25e6dff7c2e2ab59327221e (patch)
tree6dec1af2bcdb3c58e488ec4504a5e81b931c52a8
parent7130b3043800077b48c617325fb36358ee1d33ee (diff)
downloadpsutil-978296429c3eac20f25e6dff7c2e2ab59327221e.tar.gz
some fixes to help #1762
-rwxr-xr-xpsutil/tests/runner.py4
-rwxr-xr-xpsutil/tests/test_contracts.py5
-rwxr-xr-xpsutil/tests/test_process.py93
-rwxr-xr-xpsutil/tests/test_testutils.py4
-rwxr-xr-xpsutil/tests/test_windows.py5
5 files changed, 65 insertions, 46 deletions
diff --git a/psutil/tests/runner.py b/psutil/tests/runner.py
index 35fea42a..8b86a3e9 100755
--- a/psutil/tests/runner.py
+++ b/psutil/tests/runner.py
@@ -194,8 +194,6 @@ class ColouredTextRunner(unittest.TextTestRunner):
def run(self, suite):
result = self._run(suite)
- if CI_TESTING:
- print_sysinfo()
self._exit(result.wasSuccessful())
@@ -280,8 +278,6 @@ class ParallelRunner(ColouredTextRunner):
ser.testsRun, ser_fails, ser_errs, ser_skips, ser_elapsed)))
print("Ran %s tests in %.3fs using %s workers" % (
par.testsRun + ser.testsRun, par_elapsed + ser_elapsed, NWORKERS))
- if CI_TESTING:
- print_sysinfo()
ok = par.wasSuccessful() and ser.wasSuccessful()
self._exit(ok)
diff --git a/psutil/tests/test_contracts.py b/psutil/tests/test_contracts.py
index a57e19e9..57d11b43 100755
--- a/psutil/tests/test_contracts.py
+++ b/psutil/tests/test_contracts.py
@@ -14,6 +14,7 @@ import multiprocessing
import os
import signal
import stat
+import sys
import time
import traceback
@@ -672,6 +673,10 @@ class TestFetchAllProcesses(PsutilTestCase):
priorities = [getattr(psutil, x) for x in dir(psutil)
if x.endswith('_PRIORITY_CLASS')]
self.assertIn(ret, priorities)
+ if sys.version_info > (3, 4):
+ self.assertIsInstance(ret, enum.IntEnum)
+ else:
+ self.assertIsInstance(ret, int)
def num_ctx_switches(self, ret, info):
assert is_namedtuple(ret)
diff --git a/psutil/tests/test_process.py b/psutil/tests/test_process.py
index 59942f59..4a21cae5 100755
--- a/psutil/tests/test_process.py
+++ b/psutil/tests/test_process.py
@@ -37,10 +37,10 @@ from psutil._compat import PY3
from psutil._compat import super
from psutil.tests import APPVEYOR
from psutil.tests import call_until
+from psutil.tests import CI_TESTING
from psutil.tests import CIRRUS
from psutil.tests import copyload_shared_lib
from psutil.tests import create_exe
-from psutil.tests import enum
from psutil.tests import GITHUB_WHEELS
from psutil.tests import GLOBAL_TIMEOUT
from psutil.tests import HAS_CPU_AFFINITY
@@ -349,11 +349,13 @@ class TestProcess(PsutilTestCase):
@unittest.skipIf(not LINUX, "linux only")
def test_ionice_linux(self):
p = psutil.Process()
- self.assertEqual(p.ionice()[0], psutil.IOPRIO_CLASS_NONE)
+ if not CI_TESTING:
+ self.assertEqual(p.ionice()[0], psutil.IOPRIO_CLASS_NONE)
self.assertEqual(psutil.IOPRIO_CLASS_NONE, 0)
self.assertEqual(psutil.IOPRIO_CLASS_RT, 1) # high
self.assertEqual(psutil.IOPRIO_CLASS_BE, 2) # normal
self.assertEqual(psutil.IOPRIO_CLASS_IDLE, 3) # low
+ init = p.ionice()
try:
# low
p.ionice(psutil.IOPRIO_CLASS_IDLE)
@@ -367,6 +369,10 @@ class TestProcess(PsutilTestCase):
self.assertEqual(tuple(p.ionice()), (psutil.IOPRIO_CLASS_BE, 7))
with self.assertRaises(ValueError):
p.ionice(psutil.IOPRIO_CLASS_BE, value=8)
+ try:
+ p.ionice(psutil.IOPRIO_CLASS_RT, value=7)
+ except psutil.AccessDenied:
+ pass
# errs
self.assertRaisesRegex(
ValueError, "ioclass accepts no value",
@@ -378,13 +384,18 @@ class TestProcess(PsutilTestCase):
ValueError, "'ioclass' argument must be specified",
p.ionice, value=1)
finally:
- p.ionice(psutil.IOPRIO_CLASS_BE)
+ ioclass, value = init
+ if ioclass == psutil.IOPRIO_CLASS_NONE:
+ value = 0
+ p.ionice(ioclass, value)
@unittest.skipIf(not HAS_IONICE, "not supported")
@unittest.skipIf(not WINDOWS, 'not supported on this win version')
def test_ionice_win(self):
p = psutil.Process()
- self.assertEqual(p.ionice(), psutil.IOPRIO_NORMAL)
+ if not CI_TESTING:
+ self.assertEqual(p.ionice(), psutil.IOPRIO_NORMAL)
+ init = p.ionice()
try:
# base
p.ionice(psutil.IOPRIO_VERYLOW)
@@ -405,8 +416,7 @@ class TestProcess(PsutilTestCase):
ValueError, "is not a valid priority",
p.ionice, psutil.IOPRIO_HIGH + 1)
finally:
- p.ionice(psutil.IOPRIO_NORMAL)
- self.assertEqual(p.ionice(), psutil.IOPRIO_NORMAL)
+ p.ionice(init)
@unittest.skipIf(not HAS_RLIMIT, "not supported")
def test_rlimit_get(self):
@@ -784,43 +794,46 @@ class TestProcess(PsutilTestCase):
def test_nice(self):
p = psutil.Process()
self.assertRaises(TypeError, p.nice, "str")
- if WINDOWS:
- try:
- init = p.nice()
- if sys.version_info > (3, 4):
- self.assertIsInstance(init, enum.IntEnum)
- else:
- self.assertIsInstance(init, int)
- self.assertEqual(init, psutil.NORMAL_PRIORITY_CLASS)
- p.nice(psutil.HIGH_PRIORITY_CLASS)
- self.assertEqual(p.nice(), psutil.HIGH_PRIORITY_CLASS)
- p.nice(psutil.NORMAL_PRIORITY_CLASS)
- self.assertEqual(p.nice(), psutil.NORMAL_PRIORITY_CLASS)
- finally:
- p.nice(psutil.NORMAL_PRIORITY_CLASS)
- else:
- first_nice = p.nice()
- try:
- if hasattr(os, "getpriority"):
- self.assertEqual(
- os.getpriority(os.PRIO_PROCESS, os.getpid()), p.nice())
- p.nice(1)
- self.assertEqual(p.nice(), 1)
- if hasattr(os, "getpriority"):
- self.assertEqual(
- os.getpriority(os.PRIO_PROCESS, os.getpid()), p.nice())
- # XXX - going back to previous nice value raises
- # AccessDenied on MACOS
- if not MACOS:
- p.nice(0)
- self.assertEqual(p.nice(), 0)
- except psutil.AccessDenied:
- pass
- finally:
+ init = p.nice()
+ try:
+ if WINDOWS:
+ for prio in [psutil.NORMAL_PRIORITY_CLASS,
+ psutil.IDLE_PRIORITY_CLASS,
+ psutil.BELOW_NORMAL_PRIORITY_CLASS,
+ psutil.REALTIME_PRIORITY_CLASS,
+ psutil.HIGH_PRIORITY_CLASS,
+ psutil.ABOVE_NORMAL_PRIORITY_CLASS]:
+ with self.subTest(prio=prio):
+ try:
+ p.nice(prio)
+ except psutil.AccessDenied:
+ pass
+ else:
+ self.assertEqual(p.nice(), prio)
+ else:
try:
- p.nice(first_nice)
+ if hasattr(os, "getpriority"):
+ self.assertEqual(
+ os.getpriority(os.PRIO_PROCESS, os.getpid()),
+ p.nice())
+ p.nice(1)
+ self.assertEqual(p.nice(), 1)
+ if hasattr(os, "getpriority"):
+ self.assertEqual(
+ os.getpriority(os.PRIO_PROCESS, os.getpid()),
+ p.nice())
+ # XXX - going back to previous nice value raises
+ # AccessDenied on MACOS
+ if not MACOS:
+ p.nice(0)
+ self.assertEqual(p.nice(), 0)
except psutil.AccessDenied:
pass
+ finally:
+ try:
+ p.nice(init)
+ except psutil.AccessDenied:
+ pass
def test_status(self):
p = psutil.Process()
diff --git a/psutil/tests/test_testutils.py b/psutil/tests/test_testutils.py
index 901715a1..31ac4dee 100755
--- a/psutil/tests/test_testutils.py
+++ b/psutil/tests/test_testutils.py
@@ -27,6 +27,7 @@ from psutil.tests import bind_socket
from psutil.tests import bind_unix_socket
from psutil.tests import call_until
from psutil.tests import chdir
+from psutil.tests import CI_TESTING
from psutil.tests import create_sockets
from psutil.tests import get_free_port
from psutil.tests import HAS_CONNECTIONS_UNIX
@@ -364,6 +365,7 @@ class TestMemLeakClass(TestMemoryLeak):
self.assertRaises(ValueError, self.execute, lambda: 0, retries=-1)
@retry_on_failure()
+ @unittest.skipIf(CI_TESTING, "skipped on CI")
def test_leak_mem(self):
ls = []
@@ -373,7 +375,7 @@ class TestMemLeakClass(TestMemoryLeak):
try:
# will consume around 3M in total
self.assertRaisesRegex(AssertionError, "extra-mem",
- self.execute, fun, times=50, retries=2)
+ self.execute, fun, times=50)
finally:
del ls
diff --git a/psutil/tests/test_windows.py b/psutil/tests/test_windows.py
index 23ad0584..7647395a 100755
--- a/psutil/tests/test_windows.py
+++ b/psutil/tests/test_windows.py
@@ -166,6 +166,8 @@ class TestSystemAPIs(TestCase):
break
if 'cdrom' in ps_part.opts:
break
+ if ps_part.mountpoint.startswith('A:'):
+ break # floppy
try:
usage = psutil.disk_usage(ps_part.mountpoint)
except FileNotFoundError:
@@ -199,7 +201,8 @@ class TestSystemAPIs(TestCase):
sys_value = [
x + '\\' for x in win32api.GetLogicalDriveStrings().split("\\\x00")
if x and not x.startswith('A:')]
- psutil_value = [x.mountpoint for x in psutil.disk_partitions(all=True)]
+ psutil_value = [x.mountpoint for x in psutil.disk_partitions(all=True)
+ if not x.startswith('A:')]
self.assertEqual(sys_value, psutil_value)
def test_net_if_stats(self):