diff options
author | Giampaolo Rodola <g.rodola@gmail.com> | 2020-05-18 21:41:23 +0200 |
---|---|---|
committer | Giampaolo Rodola <g.rodola@gmail.com> | 2020-05-18 21:41:23 +0200 |
commit | 978296429c3eac20f25e6dff7c2e2ab59327221e (patch) | |
tree | 6dec1af2bcdb3c58e488ec4504a5e81b931c52a8 | |
parent | 7130b3043800077b48c617325fb36358ee1d33ee (diff) | |
download | psutil-978296429c3eac20f25e6dff7c2e2ab59327221e.tar.gz |
some fixes to help #1762
-rwxr-xr-x | psutil/tests/runner.py | 4 | ||||
-rwxr-xr-x | psutil/tests/test_contracts.py | 5 | ||||
-rwxr-xr-x | psutil/tests/test_process.py | 93 | ||||
-rwxr-xr-x | psutil/tests/test_testutils.py | 4 | ||||
-rwxr-xr-x | psutil/tests/test_windows.py | 5 |
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): |