Add psutil build step to fix pyauto media issues. Upgrade psutil to 0.4.0.

third_party/psutil/test/test_psutil.py

 #!/usr/bin/env python
 #
-# $Id: test_psutil.py 1142 2011-10-05 18:45:49Z g.rodola $
+# $Id: test_psutil.py 1204 2011-10-24 19:19:01Z g.rodola $
 #
 # Copyright (c) 2009, Jay Loden, 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.
 
 """
 psutil test suite.
 
 Note: this is targeted for both python 2.x and 3.x so there's no need
 to use 2to3 tool first.
@@ skipping 418 matching lines @@
             path = os.path.abspath(path)
             while not os.path.ismount(path):
                 path = os.path.dirname(path)
             return path
 
         mount = find_mount_point(__file__)
         mounts = [x.mountpoint for x in psutil.disk_partitions(all=True)]
         self.assertTrue(mount in mounts)
         psutil.disk_usage(mount)
 
-    # XXX
-    @skipUnless(hasattr(psutil, "network_io_counters"))
-    def test_anetwork_io_counters(self):
+    def test_network_io_counters(self):
         def check_ntuple(nt):
             self.assertEqual(nt[0], nt.bytes_sent)
             self.assertEqual(nt[1], nt.bytes_recv)
             self.assertEqual(nt[2], nt.packets_sent)
             self.assertEqual(nt[3], nt.packets_recv)
             self.assertTrue(nt.bytes_sent >= 0)
             self.assertTrue(nt.bytes_recv >= 0)
             self.assertTrue(nt.packets_sent >= 0)
             self.assertTrue(nt.packets_recv >= 0)
 
         ret = psutil.network_io_counters(pernic=False)
         check_ntuple(ret)
         ret = psutil.network_io_counters(pernic=True)
-        for name, ntuple in ret.iteritems():
-            self.assertTrue(name)
-            check_ntuple(ntuple)
-    # XXX
-    @skipUnless(hasattr(psutil, "disk_io_counters"))
+        self.assertTrue(ret != [])
+        for key in ret:
+            self.assertTrue(key)
+            check_ntuple(ret[key])
+
     def test_disk_io_counters(self):
         def check_ntuple(nt):
             self.assertEqual(nt[0], nt.read_count)
             self.assertEqual(nt[1], nt.write_count)
             self.assertEqual(nt[2], nt.read_bytes)
             self.assertEqual(nt[3], nt.write_bytes)
             self.assertEqual(nt[4], nt.read_time)
             self.assertEqual(nt[5], nt.write_time)
             self.assertTrue(nt.read_count >= 0)
             self.assertTrue(nt.write_count >= 0)
             self.assertTrue(nt.read_bytes >= 0)
             self.assertTrue(nt.write_bytes >= 0)
             self.assertTrue(nt.read_time >= 0)
             self.assertTrue(nt.write_time >= 0)
 
         ret = psutil.disk_io_counters(perdisk=False)
         check_ntuple(ret)
         ret = psutil.disk_io_counters(perdisk=True)
-        for name, ntuple in ret.iteritems():
-            self.assertTrue(name)
-            check_ntuple(ntuple)
+        for key in ret:
+            self.assertTrue(key)
+            check_ntuple(ret[key])
 
     # ====================
     # Process object tests
     # ====================
 
     def test_kill(self):
         sproc = get_test_subprocess()
         test_pid = sproc.pid
         wait_for_pid(test_pid)
         p = psutil.Process(test_pid)
@@ skipping 115 matching lines @@
         self.assertFalse(p.is_running())
 
     def test_cpu_percent(self):
         p = psutil.Process(os.getpid())
         p.get_cpu_percent(interval=0.001)
         p.get_cpu_percent(interval=0.001)
         for x in range(100):
             percent = p.get_cpu_percent(interval=None)
             self.assertTrue(isinstance(percent, float))
             self.assertTrue(percent >= 0.0)
-            self.assertTrue(percent <= 100.0)
+            if os.name != 'posix':
+                self.assertTrue(percent <= 100.0)
+            else:
+                self.assertTrue(percent >= 0.0)
 
     def test_cpu_times(self):
         times = psutil.Process(os.getpid()).get_cpu_times()
         self.assertTrue((times.user > 0.0) or (times.system > 0.0))
         # make sure returned values can be pretty printed with strftime
         time.strftime("%H:%M:%S", time.localtime(times.user))
         time.strftime("%H:%M:%S", time.localtime(times.system))
 
     # Test Process.cpu_times() against os.times()
     # os.times() is broken on Python 2.6
@@ skipping 253 matching lines @@
                     first_nice = p.nice
                     p.nice = 1
                     self.assertEqual(p.nice, 1)
                     # going back to previous nice value raises AccessDenied on OSX
                     if not OSX:
                         p.nice = 0
                         self.assertEqual(p.nice, 0)
                 except psutil.AccessDenied:
                     pass
             finally:
-                # going back to previous nice value raises AccessDenied on OSX
-                if not OSX:
+                try:
                     p.nice = first_nice
+                except psutil.AccessDenied:
+                    pass
 
     if os.name == 'nt':
 
         def test_nice(self):
             p = psutil.Process(os.getpid())
             self.assertRaises(TypeError, setattr, p, "nice", "str")
             try:
                 self.assertEqual(p.nice, psutil.NORMAL_PRIORITY_CLASS)
                 p.nice = psutil.HIGH_PRIORITY_CLASS
                 self.assertEqual(p.nice, psutil.HIGH_PRIORITY_CLASS)
@@ skipping 18 matching lines @@
             self.assertEqual(p.username, pwd.getpwuid(os.getuid()).pw_name)
         elif WINDOWS:
             expected_username = os.environ['USERNAME']
             expected_domain = os.environ['USERDOMAIN']
             domain, username = p.username.split('\\')
             self.assertEqual(domain, expected_domain)
             self.assertEqual(username, expected_username)
         else:
             p.username
 
-    @skipUnless(WINDOWS or LINUX)
+    @skipIf(not hasattr(psutil.Process, "getcwd"))
     def test_getcwd(self):
         sproc = get_test_subprocess()
         wait_for_pid(sproc.pid)
         p = psutil.Process(sproc.pid)
         self.assertEqual(p.getcwd(), os.getcwd())
 
-    @skipUnless(WINDOWS or LINUX)
+    @skipIf(not hasattr(psutil.Process, "getcwd"))
     def test_getcwd_2(self):
         cmd = [PYTHON, "-c", "import os, time; os.chdir('..'); time.sleep(10)"]
         sproc = get_test_subprocess(cmd)
         wait_for_pid(sproc.pid)
         p = psutil.Process(sproc.pid)
         time.sleep(0.1)
         expected_dir = os.path.dirname(os.getcwd())
         self.assertEqual(p.getcwd(), expected_dir)
 
     def test_get_open_files(self):
@@ skipping 83 matching lines @@
             self.assertEqual(con.fd, -1)
         else:
             self.assertTrue(con.fd > 0)
         # test positions
         self.assertEqual(con[0], con.fd)
         self.assertEqual(con[1], con.family)
         self.assertEqual(con[2], con.type)
         self.assertEqual(con[3], con.local_address)
         self.assertEqual(con[4], con.remote_address)
         self.assertEqual(con[5], con.status)
+        # test kind arg
+        self.assertRaises(ValueError, p.get_connections, 'foo')
 
     @skipUnless(supports_ipv6())
     def test_get_connections_ipv6(self):
         s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
         s.bind(('::1', 0))
         s.listen(1)
         cons = psutil.Process(os.getpid()).get_connections()
         s.close()
         self.assertEqual(len(cons), 1)
         self.assertEqual(cons[0].local_address[0], '::1')
@@ skipping 54 matching lines @@
         tcp4_template = Template(tcp_template).substitute(family=socket.AF_INET,
                                                           addr="127.0.0.1")
         udp4_template = Template(udp_template).substitute(family=socket.AF_INET,
                                                           addr="127.0.0.1")
         tcp6_template = Template(tcp_template).substitute(family=socket.AF_INET6,
                                                           addr="::1")
         udp6_template = Template(udp_template).substitute(family=socket.AF_INET6,
                                                           addr="::1")
 
         # launch various subprocess instantiating a socket of various
-        # families  and tupes to enrich psutil results
+        # families and types to enrich psutil results
         tcp4_proc = get_test_subprocess([PYTHON, "-c", tcp4_template])
         udp4_proc = get_test_subprocess([PYTHON, "-c", udp4_template])
         if supports_ipv6():
             tcp6_proc = get_test_subprocess([PYTHON, "-c", tcp6_template])
             udp6_proc = get_test_subprocess([PYTHON, "-c", udp6_template])
         else:
             tcp6_proc = None
             udp6_proc = None
 
         # --- check connections of all processes
@@ skipping 39 matching lines @@
                             if hasattr(dupsock, "family"):
                                 self.assertEqual(dupsock.family, conn.family)
                                 self.assertEqual(dupsock.type, conn.type)
                         finally:
                             if dupsock is not None:
                                 dupsock.close()
 
 
         # --- check matches against subprocesses
 
+        all_kinds = ("all", "inet", "inet4", "inet6", "tcp", "tcp4", "tcp6",
+                     "udp", "udp4", "udp6")
+
         for p in psutil.Process(os.getpid()).get_children():
             for conn in p.get_connections():
                 # TCP v4
                 if p.pid == tcp4_proc.pid:
                     self.assertEqual(conn.family, socket.AF_INET)
                     self.assertEqual(conn.type, socket.SOCK_STREAM)
                     self.assertEqual(conn.local_address[0], "127.0.0.1")
                     self.assertEqual(conn.remote_address, ())
                     self.assertEqual(conn.status, "LISTEN")
+                    for kind in all_kinds:
+                        cons = p.get_connections(kind=kind)
+                        if kind in ("all", "inet", "inet4", "tcp", "tcp4"):
+                            self.assertTrue(cons != [])
+                        else:
+                            self.assertEqual(cons, [])
                 # UDP v4
                 elif p.pid == udp4_proc.pid:
                     self.assertEqual(conn.family, socket.AF_INET)
                     self.assertEqual(conn.type, socket.SOCK_DGRAM)
                     self.assertEqual(conn.local_address[0], "127.0.0.1")
                     self.assertEqual(conn.remote_address, ())
                     self.assertEqual(conn.status, "")
+                    for kind in all_kinds:
+                        cons = p.get_connections(kind=kind)
+                        if kind in ("all", "inet", "inet4", "udp", "udp4"):
+                            self.assertTrue(cons != [])
+                        else:
+                            self.assertEqual(cons, [])
                 # TCP v6
                 elif p.pid == getattr(tcp6_proc, "pid", None):
                     self.assertEqual(conn.family, socket.AF_INET6)
                     self.assertEqual(conn.type, socket.SOCK_STREAM)
                     self.assertTrue(conn.local_address[0] in ("::", "::1"))
                     self.assertEqual(conn.remote_address, ())
                     self.assertEqual(conn.status, "LISTEN")
+                    for kind in all_kinds:
+                        cons = p.get_connections(kind=kind)
+                        if kind in ("all", "inet", "inet6", "tcp", "tcp6"):
+                            self.assertTrue(cons != [])
+                        else:
+                            self.assertEqual(cons, [])
                 # UDP v6
                 elif p.pid == getattr(udp6_proc, "pid", None):
                     self.assertEqual(conn.family, socket.AF_INET6)
                     self.assertEqual(conn.type, socket.SOCK_DGRAM)
                     self.assertTrue(conn.local_address[0] in ("::", "::1"))
                     self.assertEqual(conn.remote_address, ())
                     self.assertEqual(conn.status, "")
+                    for kind in all_kinds:
+                        cons = p.get_connections(kind=kind)
+                        if kind in ("all", "inet", "inet6", "udp", "udp6"):
+                            self.assertTrue(cons != [])
+                        else:
+                            self.assertEqual(cons, [])
 
     def test_parent_ppid(self):
         this_parent = os.getpid()
         sproc = get_test_subprocess()
         p = psutil.Process(sproc.pid)
         self.assertEqual(p.ppid, this_parent)
         self.assertEqual(p.parent.pid, this_parent)
         # no other process is supposed to have us as parent
         for p in psutil.process_iter():
             if p.pid == sproc.pid:
@@ skipping 81 matching lines @@
         self.assertTrue(str(sproc.pid) in str(p))
         self.assertTrue("terminated" in str(p))
 
     def test_fetch_all(self):
         valid_procs = 0
         excluded_names = ['send_signal', 'suspend', 'resume', 'terminate',
                           'kill', 'wait']
         excluded_names += ['get_cpu_percent', 'get_children']
         # XXX - skip slow lsof implementation;
         if BSD:
-           excluded_names += ['get_open_files', 'get_connections']
-        if OSX:
            excluded_names += ['get_connections']
         attrs = []
         for name in dir(psutil.Process):
             if name.startswith("_"):
                 continue
             if name.startswith("set_"):
                 continue
             if name in excluded_names:
                 continue
             attrs.append(name)
@@ skipping 11 matching lines @@
                     except (psutil.NoSuchProcess, psutil.AccessDenied):
                         err = sys.exc_info()[1]
                         self.assertEqual(err.pid, p.pid)
                         if err.name:
                             self.assertEqual(err.name, p.name)
                         self.assertTrue(str(err))
                         self.assertTrue(err.msg)
                     else:
                         if name == 'parent' or ret in (0, 0.0, [], None):
                             continue
+                        # getcwd() on FreeBSD may be an empty string
+                        # in case of a system process
+                        if name == 'getcwd' and BSD and ret == '':
+                            continue
                         self.assertTrue(ret)
                         if name == "exe":
                             self.assertTrue(os.path.isfile(ret))
                         elif name == "getcwd":
                             # XXX - temporary fix; on my Linux box
                             # chrome process cws is errnously reported
                             # as /proc/4144/fdinfo whichd doesn't exist
                             if 'chrome' in p.name:
                                 continue
                             self.assertTrue(os.path.isdir(ret))
@@ skipping 37 matching lines @@
         elif WINDOWS:
             self.assertEqual(p.username, 'NT AUTHORITY\\SYSTEM')
         else:
             p.username
 
         self.assertTrue(0 in psutil.get_pid_list())
         self.assertTrue(psutil.pid_exists(0))
 
     def test_Popen(self):
         # Popen class test
+        # XXX this test causes a ResourceWarning on Python 3 because
+        # psutil.__subproc instance doesn't get propertly freed.
+        # Not sure what to do though.
         cmd = [PYTHON, "-c", "import time; time.sleep(3600);"]
         proc = psutil.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         try:
             proc.name
             proc.stdin
             self.assertTrue(hasattr(proc, 'name'))
             self.assertTrue(hasattr(proc, 'stdin'))
             self.assertRaises(AttributeError, getattr, proc, 'foo')
         finally:
             proc.kill()
@@ skipping 74 matching lines @@
 
     f = open(TESTFN, 'w')
     f.close()
     atexit.register(lambda: os.remove(TESTFN))
 
     unittest.TextTestRunner(verbosity=2).run(test_suite)
     DEVNULL.close()
 
 if __name__ == '__main__':
     test_main()