refact assertProcessGone

psutil/tests/__init__.py

@@ -944,20 +944,45 @@ def pyrun(self, *args, **kwds):
         self.addCleanup(terminate, sproc)  # executed first
         return sproc
 
-    def assertProcessGone(self, proc):
-        self.assertRaises(psutil.NoSuchProcess, psutil.Process, proc.pid)
-        if isinstance(proc, (psutil.Process, psutil.Popen)):
-            assert not proc.is_running()
+    def _check_proc_exc(self, proc, exc):
+        self.assertIsInstance(exc, psutil.Error)
+        self.assertEqual(exc.pid, proc.pid)
+        self.assertEqual(exc.name, proc._name)
+        if exc.name:
+            self.assertNotEqual(exc.name, "")
+        if isinstance(exc, psutil.ZombieProcess):
+            self.assertEqual(exc.ppid, proc._ppid)
+            if exc.ppid is not None:
+                self.assertGreaterEqual(exc.ppid, 0)
+        str(exc)
+        repr(exc)
+
+    def assertPidGone(self, pid):
+        with self.assertRaises(psutil.NoSuchProcess) as cm:
             try:
-                status = proc.status()
-            except psutil.NoSuchProcess:
-                pass
-            else:
-                raise AssertionError("Process.status() didn't raise exception "
-                                     "(status=%s)" % status)
-            proc.wait(timeout=0)  # assert not raise TimeoutExpired
-        assert not psutil.pid_exists(proc.pid), proc.pid
-        self.assertNotIn(proc.pid, psutil.pids())
+                psutil.Process(pid)
+            except psutil.ZombieProcess:
+                raise AssertionError(
+                    "wasn't supposed to raise ZombieProcess")
+        self.assertEqual(cm.exception.pid, pid)
+        self.assertEqual(cm.exception.name, None)
+        assert not psutil.pid_exists(pid), pid
+        self.assertNotIn(pid, psutil.pids())
+        self.assertNotIn(pid, [x.pid for x in psutil.process_iter()])
+
+    def assertProcessGone(self, proc):
+        self.assertPidGone(proc.pid)
+        ns = process_namespace(proc)
+        for fun, name in ns.iter(ns.all, clear_cache=True):
+            with self.subTest(proc=proc, name=name):
+                with self.assertRaises(psutil.NoSuchProcess) as cm:
+                    try:
+                        fun()
+                    except psutil.ZombieProcess:
+                        raise AssertionError(
+                            "wasn't supposed to raise ZombieProcess")
+                self._check_proc_exc(proc, cm.exception)
+        proc.wait(timeout=0)  # assert not raise TimeoutExpired
 
     def assertProcessZombie(self, proc):
         # A zombie process should always be instantiable.
@@ -981,17 +1006,23 @@ def assertProcessZombie(self, proc):
         self.assertIn(proc.pid, [x.pid for x in psutil.process_iter()])
         # Call all methods.
         ns = process_namespace(proc)
-        for fun, name in ns.iter(ns.all):
-            with self.subTest(name):
+        for fun, name in ns.iter(ns.all, clear_cache=True):
+            with self.subTest(proc=proc, name=name):
                 try:
                     fun()
-                except (psutil.ZombieProcess, psutil.AccessDenied):
-                    pass
+                except (psutil.ZombieProcess, psutil.AccessDenied) as exc:
+                    self._check_proc_exc(proc, exc)
         if LINUX:
             # https://github.com/giampaolo/psutil/pull/2288
-            self.assertRaises(psutil.ZombieProcess, proc.cmdline)
-            self.assertRaises(psutil.ZombieProcess, proc.exe)
-            self.assertRaises(psutil.ZombieProcess, proc.memory_maps)
+            with self.assertRaises(psutil.ZombieProcess) as cm:
+                proc.cmdline()
+            self._check_proc_exc(proc, cm.exception)
+            with self.assertRaises(psutil.ZombieProcess) as cm:
+                proc.exe()
+            self._check_proc_exc(proc, cm.exception)
+            with self.assertRaises(psutil.ZombieProcess) as cm:
+                proc.memory_maps()
+            self._check_proc_exc(proc, cm.exception)
         # Zombie cannot be signaled or terminated.
         proc.suspend()
         proc.resume()

psutil/tests/test_contracts.py

@@ -366,6 +366,7 @@ def check_exception(exc, proc, name, ppid):
         elif isinstance(exc, psutil.NoSuchProcess):
             tcase.assertProcessGone(proc)
         str(exc)
+        repr(exc)
 
     def do_wait():
         if pid != 0:
@@ -376,23 +377,27 @@ def do_wait():
 
     try:
         proc = psutil.Process(pid)
-        d = proc.as_dict(['ppid', 'name'])
     except psutil.NoSuchProcess:
+        tcase.assertPidGone(pid)
         return {}
-
-    name, ppid = d['name'], d['ppid']
-    info = {'pid': proc.pid}
-    ns = process_namespace(proc)
-    # We don't use oneshot() because in order not to fool
-    # check_exception() in case of NSP.
-    for fun, fun_name in ns.iter(ns.getters, clear_cache=False):
-        try:
-            info[fun_name] = fun()
-        except psutil.Error as exc:
-            check_exception(exc, proc, name, ppid)
-            continue
-    do_wait()
-    return info
+    try:
+        d = proc.as_dict(['ppid', 'name'])
+    except psutil.NoSuchProcess:
+        tcase.assertProcessGone(proc)
+    else:
+        name, ppid = d['name'], d['ppid']
+        info = {'pid': proc.pid}
+        ns = process_namespace(proc)
+        # We don't use oneshot() because in order not to fool
+        # check_exception() in case of NSP.
+        for fun, fun_name in ns.iter(ns.getters, clear_cache=False):
+            try:
+                info[fun_name] = fun()
+            except psutil.Error as exc:
+                check_exception(exc, proc, name, ppid)
+                continue
+        do_wait()
+        return info
 
 
 @serialrun
@@ -402,7 +407,7 @@ class TestFetchAllProcesses(PsutilTestCase):
     Uses a process pool to get info about all processes.
     """
 
-    use_proc_pool = not CI_TESTING
+    use_proc_pool = 0
 
     def setUp(self):
         # Using a pool in a CI env may result in deadlock, see:
@@ -435,9 +440,9 @@ def test_all(self):
                 meth = getattr(self, name)
                 try:
                     meth(value, info)
-                except AssertionError:
+                except Exception:
                     s = '\n' + '=' * 70 + '\n'
-                    s += "FAIL: test_%s pid=%s, ret=%s\n" % (
+                    s += "FAIL: name=test_%s, pid=%s, ret=%s\n" % (
                         name, info['pid'], repr(value))
                     s += '-' * 70
                     s += "\n%s" % traceback.format_exc()
@@ -480,6 +485,7 @@ def pid(self, ret, info):
     def ppid(self, ret, info):
         self.assertIsInstance(ret, (int, long))
         self.assertGreaterEqual(ret, 0)
+        proc_info(ret)
 
     def name(self, ret, info):
         self.assertIsInstance(ret, (str, unicode))

psutil/tests/test_process.py

@@ -1072,11 +1072,6 @@ def test_ppid(self):
             self.assertEqual(p.ppid(), os.getppid())
         p = self.spawn_psproc()
         self.assertEqual(p.ppid(), os.getpid())
-        if APPVEYOR:
-            # Occasional failures, see:
-            # https://ci.appveyor.com/project/giampaolo/psutil/build/
-            #     job/0hs623nenj7w4m33
-            return
 
     def test_parent(self):
         p = self.spawn_psproc()

psutil/tests/test_testutils.py

@@ -252,32 +252,32 @@ def test_terminate(self):
         # by subprocess.Popen
         p = self.spawn_testproc()
         terminate(p)
-        self.assertProcessGone(p)
+        self.assertPidGone(p.pid)
         terminate(p)
         # by psutil.Process
         p = psutil.Process(self.spawn_testproc().pid)
         terminate(p)
-        self.assertProcessGone(p)
+        self.assertPidGone(p.pid)
         terminate(p)
         # by psutil.Popen
         cmd = [PYTHON_EXE, "-c", "import time; time.sleep(60);"]
         p = psutil.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          env=PYTHON_EXE_ENV)
         terminate(p)
-        self.assertProcessGone(p)
+        self.assertPidGone(p.pid)
         terminate(p)
         # by PID
         pid = self.spawn_testproc().pid
         terminate(pid)
-        self.assertProcessGone(p)
+        self.assertPidGone(p.pid)
         terminate(pid)
         # zombie
         if POSIX:
             parent, zombie = self.spawn_zombie()
             terminate(parent)
             terminate(zombie)
-            self.assertProcessGone(parent)
-            self.assertProcessGone(zombie)
+            self.assertPidGone(parent.pid)
+            self.assertPidGone(zombie.pid)
 
 
 class TestNetUtils(PsutilTestCase):