#777 / linux / open_files(): return also file mode

psutil/_pslinux.py

@@ -179,6 +179,16 @@ def readlink(path):
     return path
 
 
+def file_flags_to_mode(flags):
+    md = {os.O_RDONLY: 'r', os.O_WRONLY: 'w', os.O_RDWR: 'w+'}
+    m = md[flags & (os.O_RDONLY | os.O_WRONLY | os.O_RDWR)]
+    if flags & os.O_APPEND:
+        m = m.replace('w', 'a', 1)
+    m = m.replace('w+', 'r+')
+    # possible values: r, w, a, r+, a+
+    return m
+
+
 def get_sector_size():
     try:
         with open(b"/sys/block/sda/queue/hw_sector_size") as f:
@@ -237,7 +247,7 @@ def set_scputimes_ntuple(procfs_path):
                                  'read_merged_count', 'write_merged_count',
                                  'busy_time'])
 
-popenfile = namedtuple('popenfile', ['path', 'fd', 'position'])
+popenfile = namedtuple('popenfile', ['path', 'fd', 'position', 'mode'])
 pmem = namedtuple('pmem', 'rss vms shared text lib data dirty')
 pfullmem = namedtuple('pfullmem', pmem._fields + ('uss', 'pss', 'swap'))
 
@@ -1318,8 +1328,12 @@ def open_files(self):
                     file = "%s/%s/fdinfo/%s" % (
                         self._procfs_path, self.pid, fd)
                     with open_binary(file) as f:
-                        pos = int(f.readline().split()[1])
-                    ntuple = popenfile(path, int(fd), pos)
+                        pos = f.readline().split()[1]
+                        flags = f.readline().split()[1]
+                    # flags is an octal number
+                    flags_oct = int(flags, 8)
+                    mode = file_flags_to_mode(flags_oct)
+                    ntuple = popenfile(path, int(fd), int(pos), mode)
                     retlist.append(ntuple)
         if hit_enoent:
             # raise NSP if the process disappeared on us

psutil/tests/test_linux.py

@@ -37,6 +37,7 @@
 from psutil.tests import reap_children
 from psutil.tests import retry_before_failing
 from psutil.tests import run_test_module_by_name
+from psutil.tests import safe_remove
 from psutil.tests import sh
 from psutil.tests import skip_on_not_implemented
 from psutil.tests import TESTFN
@@ -719,6 +720,11 @@ def open_mock(name, *args, **kwargs):
 @unittest.skipUnless(LINUX, "not a Linux system")
 class TestProcess(unittest.TestCase):
 
+    def setUp(self):
+        safe_remove(TESTFN)
+
+    tearDown = setUp
+
     def test_memory_maps(self):
         src = textwrap.dedent("""
             import time
@@ -765,6 +771,41 @@ def test_memory_full_info(self):
         self.assertEqual(
             mem.swap, sum([x.swap for x in maps]))
 
+    def test_open_files_mode(self):
+        def get_test_file():
+            p = psutil.Process()
+            giveup_at = time.time() + 2
+            while True:
+                for file in p.open_files():
+                    if file.path == os.path.abspath(TESTFN):
+                        return file
+                    elif time.time() > giveup_at:
+                        break
+            raise RuntimeError("timeout looking for test file")
+
+        #
+        with open(TESTFN, "w"):
+            self.assertEqual(get_test_file().mode, "w")
+        with open(TESTFN, "r"):
+            self.assertEqual(get_test_file().mode, "r")
+        with open(TESTFN, "a"):
+            self.assertEqual(get_test_file().mode, "a")
+        #
+        with open(TESTFN, "r+"):
+            self.assertEqual(get_test_file().mode, "r+")
+        with open(TESTFN, "w+"):
+            self.assertEqual(get_test_file().mode, "r+")
+        with open(TESTFN, "a+"):
+            self.assertEqual(get_test_file().mode, "a+")
+        # note: "x" bit is not supported
+        if PY3:
+            safe_remove(TESTFN)
+            with open(TESTFN, "x"):
+                self.assertEqual(get_test_file().mode, "w")
+            safe_remove(TESTFN)
+            with open(TESTFN, "x+"):
+                self.assertEqual(get_test_file().mode, "r+")
+
     def test_open_files_file_gone(self):
         # simulates a file which gets deleted during open_files()
         # execution

psutil/tests/test_process.py

@@ -1704,6 +1704,7 @@ def open_files(self, ret, proc):
             if LINUX:
                 self.assertIsInstance(f.position, int)
                 self.assertGreaterEqual(f.position, 0)
+                self.assertIn(f.mode, ('r', 'w', 'a', 'r+', 'a+'))
             if BSD and not f.path:
                 # XXX see: https://github.com/giampaolo/psutil/issues/595
                 continue