refactor tests + make them more robust

Author: giampaolo

psutil/_pslinux.py

@@ -2031,11 +2031,7 @@ def memory_full_info(self):
             if HAS_PROC_SMAPS_ROLLUP:  # faster
                 try:
                     uss, pss, swap = self._parse_smaps_rollup()
-                except (ProcessLookupError, FileNotFoundError) as err:
-                    debug(
-                        "ignore %r for pid %s and retry using /proc/pid/smaps"
-                        % (err, self.pid)
-                    )
+                except (ProcessLookupError, FileNotFoundError):
                     uss, pss, swap = self._parse_smaps()
             else:
                 uss, pss, swap = self._parse_smaps()

psutil/tests/test_connections.py

@@ -55,15 +55,15 @@ def setUp(self):
             # Process opens a UNIX socket to /var/log/run.
             return
         cons = thisproc.connections(kind='all')
-        assert not cons, cons
+        self.assertEqual(cons, [])
 
     def tearDown(self):
         # Make sure we closed all resources.
         # Some BSDs open a UNIX socket to /var/log/run.
         if NETBSD or FREEBSD or (MACOS and not PY3):
             return
         cons = thisproc.connections(kind='all')
-        assert not cons, cons
+        self.assertEqual(cons, [])
 
     def compare_procsys_connections(self, pid, proc_cons, kind='all'):
         """Given a process PID and its list of connections compare
@@ -156,46 +156,46 @@ def test_tcp_v4(self):
         addr = ("127.0.0.1", 0)
         with closing(bind_socket(AF_INET, SOCK_STREAM, addr=addr)) as sock:
             conn = self.check_socket(sock)
-            assert not conn.raddr
+            self.assertEqual(conn.raddr, ())
             self.assertEqual(conn.status, psutil.CONN_LISTEN)
 
     @unittest.skipIf(not supports_ipv6(), "IPv6 not supported")
     def test_tcp_v6(self):
         addr = ("::1", 0)
         with closing(bind_socket(AF_INET6, SOCK_STREAM, addr=addr)) as sock:
             conn = self.check_socket(sock)
-            assert not conn.raddr
+            self.assertEqual(conn.raddr, ())
             self.assertEqual(conn.status, psutil.CONN_LISTEN)
 
     def test_udp_v4(self):
         addr = ("127.0.0.1", 0)
         with closing(bind_socket(AF_INET, SOCK_DGRAM, addr=addr)) as sock:
             conn = self.check_socket(sock)
-            assert not conn.raddr
+            self.assertEqual(conn.raddr, ())
             self.assertEqual(conn.status, psutil.CONN_NONE)
 
     @unittest.skipIf(not supports_ipv6(), "IPv6 not supported")
     def test_udp_v6(self):
         addr = ("::1", 0)
         with closing(bind_socket(AF_INET6, SOCK_DGRAM, addr=addr)) as sock:
             conn = self.check_socket(sock)
-            assert not conn.raddr
+            self.assertEqual(conn.raddr, ())
             self.assertEqual(conn.status, psutil.CONN_NONE)
 
     @unittest.skipIf(not POSIX, 'POSIX only')
     def test_unix_tcp(self):
         testfn = self.get_testfn()
         with closing(bind_unix_socket(testfn, type=SOCK_STREAM)) as sock:
             conn = self.check_socket(sock)
-            assert not conn.raddr
+            self.assertEqual(conn.raddr, "")
             self.assertEqual(conn.status, psutil.CONN_NONE)
 
     @unittest.skipIf(not POSIX, 'POSIX only')
     def test_unix_udp(self):
         testfn = self.get_testfn()
         with closing(bind_unix_socket(testfn, type=SOCK_STREAM)) as sock:
             conn = self.check_socket(sock)
-            assert not conn.raddr
+            self.assertEqual(conn.raddr, "")
             self.assertEqual(conn.status, psutil.CONN_NONE)
 
 
@@ -210,7 +210,7 @@ class TestConnectedSocket(ConnectionTestCase):
     @unittest.skipIf(SUNOS, "unreliable on SUONS")
     def test_tcp(self):
         addr = ("127.0.0.1", 0)
-        assert not thisproc.connections(kind='tcp4')
+        self.assertEqual(thisproc.connections(kind='tcp4'), [])
         server, client = tcp_socketpair(AF_INET, addr=addr)
         try:
             cons = thisproc.connections(kind='tcp4')
@@ -233,8 +233,8 @@ def test_unix(self):
         server, client = unix_socketpair(testfn)
         try:
             cons = thisproc.connections(kind='unix')
-            assert not (cons[0].laddr and cons[0].raddr)
-            assert not (cons[1].laddr and cons[1].raddr)
+            assert not (cons[0].laddr and cons[0].raddr), cons
+            assert not (cons[1].laddr and cons[1].raddr), cons
             if NETBSD or FREEBSD:
                 # On NetBSD creating a UNIX socket will cause
                 # a UNIX connection to  /var/run/log.
@@ -313,9 +313,9 @@ def check_conn(proc, conn, family, type, laddr, raddr, status, kinds):
             for kind in all_kinds:
                 cons = proc.connections(kind=kind)
                 if kind in kinds:
-                    assert cons
+                    self.assertNotEqual(cons, [])
                 else:
-                    assert not cons, cons
+                    self.assertEqual(cons, [])
             # compare against system-wide connections
             # XXX Solaris can't retrieve system-wide UNIX
             # sockets.

psutil/tests/test_contracts.py

@@ -371,7 +371,8 @@ def proc_info(pid):
 
     def check_exception(exc, proc, name, ppid):
         tcase.assertEqual(exc.pid, pid)
-        tcase.assertEqual(exc.name, name)
+        if exc.name is not None:
+            tcase.assertEqual(exc.name, name)
         if isinstance(exc, psutil.ZombieProcess):
             tcase.assertProcessZombie(proc)
             if exc.ppid is not None:
@@ -552,7 +553,7 @@ def username(self, ret, info):
 
     def status(self, ret, info):
         self.assertIsInstance(ret, str)
-        assert ret
+        assert ret, ret
         self.assertNotEqual(ret, '?')  # XXX
         self.assertIn(ret, VALID_PROC_STATUSES)
 
@@ -706,7 +707,7 @@ def is_running(self, ret, info):
 
     def cpu_affinity(self, ret, info):
         self.assertIsInstance(ret, list)
-        assert ret != [], ret
+        self.assertNotEqual(ret, [])
         cpus = list(range(psutil.cpu_count()))
         for n in ret:
             self.assertIsInstance(n, int)

psutil/tests/test_linux.py

@@ -845,7 +845,7 @@ def path_exists_mock(path):
             with mock.patch("os.path.exists", side_effect=path_exists_mock):
                 reload_module(psutil._pslinux)
                 ret = psutil.cpu_freq()
-                assert ret
+                assert ret, ret
                 self.assertEqual(ret.max, 0.0)
                 self.assertEqual(ret.min, 0.0)
                 for freq in psutil.cpu_freq(percpu=True):
@@ -1972,8 +1972,7 @@ def test_open_files_file_gone(self):
                 'psutil._pslinux.os.readlink',
                 side_effect=OSError(errno.ENOENT, ""),
             ) as m:
-                files = p.open_files()
-                assert not files
+                self.assertEqual(p.open_files(), [])
                 assert m.called
             # also simulate the case where os.readlink() returns EINVAL
             # in which case psutil is supposed to 'continue'
@@ -1997,8 +1996,7 @@ def test_open_files_fd_gone(self):
             with mock.patch(
                 patch_point, side_effect=IOError(errno.ENOENT, "")
             ) as m:
-                files = p.open_files()
-                assert not files
+                self.assertEqual(p.open_files(), [])
                 assert m.called
 
     def test_open_files_enametoolong(self):
@@ -2015,8 +2013,7 @@ def test_open_files_enametoolong(self):
                 patch_point, side_effect=OSError(errno.ENAMETOOLONG, "")
             ) as m:
                 with mock.patch("psutil._pslinux.debug"):
-                    files = p.open_files()
-                    assert not files
+                    self.assertEqual(p.open_files(), [])
                     assert m.called
 
     # --- mocked tests
@@ -2250,7 +2247,7 @@ def test_connections_enametoolong(self):
         ) as m:
             p = psutil.Process()
             with mock.patch("psutil._pslinux.debug"):
-                assert not p.connections()
+                self.assertEqual(p.connections(), [])
                 assert m.called
 
 

psutil/tests/test_process.py

@@ -78,7 +78,11 @@ class TestProcess(PsutilTestCase):
 
     def spawn_psproc(self, *args, **kwargs):
         sproc = self.spawn_testproc(*args, **kwargs)
-        return psutil.Process(sproc.pid)
+        try:
+            return psutil.Process(sproc.pid)
+        except psutil.NoSuchProcess:
+            self.assertPidGone(sproc.pid)
+            raise
 
     # ---
 

psutil/tests/test_testutils.py

@@ -318,7 +318,7 @@ def tcp_tcp_socketpair(self):
     def test_unix_socketpair(self):
         p = psutil.Process()
         num_fds = p.num_fds()
-        assert not p.connections(kind='unix')
+        self.assertEqual(p.connections(kind='unix'), [])
         name = self.get_testfn()
         server, client = unix_socketpair(name)
         try: