[Linux] convert /dev/root device to real path (fixes #1999) (#2000)

Signed-off-by: Giampaolo Rodola <g.rodola@gmail.com>

3 files changed, 142 insertions, 0 deletions

diff --git a/HISTORY.rst b/HISTORY.rst
index 3399e186..fb43a196 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -14,6 +14,8 @@ XXXX-XX-XX
 - 1992_: error classes (NoSuchProcess, AccessDenied, etc.) now have a better
   formatted and separated `__repr__` and `__str__` implementations.
 - 1996_: add support for MidnightBSD.  (patch by Saeed Rasooli)
+- 1999_: [Linux] disk_partitions(): convert "/dev/root" device (an alias used
+  on some Linux distros) to real root device path.
 
 **Bug fixes**
 
diff --git a/psutil/_pslinux.py b/psutil/_pslinux.py
index 1917e20d..f6a83301 100644
--- a/psutil/_pslinux.py
+++ b/psutil/_pslinux.py
@@ -1189,6 +1189,80 @@ def disk_io_counters(perdisk=False):
     return retdict
 
 
+class RootFsDeviceFinder:
+    """disk_partitions() may return partitions with device == "/dev/root"
+    or "rootfs". This container class uses different strategies to try to
+    obtain the real device path. Resources:
+    https://bootlin.com/blog/find-root-device/
+    https://www.systutorials.com/how-to-find-the-disk-where-root-is-on-in-bash-on-linux/
+    """
+    __slots__ = ['major', 'minor']
+
+    def __init__(self):
+        dev = os.stat("/").st_dev
+        self.major = os.major(dev)
+        self.minor = os.minor(dev)
+
+    def ask_proc_partitions(self):
+        with open_text("%s/partitions" % get_procfs_path()) as f:
+            for line in f.readlines()[2:]:
+                fields = line.split()
+                if len(fields) < 4:  # just for extra safety
+                    continue
+                major = int(fields[0]) if fields[0].isdigit() else None
+                minor = int(fields[1]) if fields[1].isdigit() else None
+                name = fields[3]
+                if major == self.major and minor == self.minor:
+                    if name:  # just for extra safety
+                        return "/dev/%s" % name
+
+    def ask_sys_dev_block(self):
+        path = "/sys/dev/block/%s:%s/uevent" % (self.major, self.minor)
+        with open_text(path) as f:
+            for line in f:
+                if line.startswith("DEVNAME="):
+                    name = line.strip().rpartition("DEVNAME=")[2]
+                    if name:  # just for extra safety
+                        return "/dev/%s" % name
+
+    def ask_sys_class_block(self):
+        needle = "%s:%s" % (self.major, self.minor)
+        files = glob.iglob("/sys/class/block/*/dev")
+        for file in files:
+            try:
+                f = open_text(file)
+            except FileNotFoundError:  # race condition
+                continue
+            else:
+                with f:
+                    data = f.read().strip()
+                    if data == needle:
+                        name = os.path.basename(os.path.dirname(file))
+                        return "/dev/%s" % name
+
+    def find(self):
+        path = None
+        if path is None:
+            try:
+                path = self.ask_proc_partitions()
+            except (IOError, OSError) as err:
+                debug(err)
+        if path is None:
+            try:
+                path = self.ask_sys_dev_block()
+            except (IOError, OSError) as err:
+                debug(err)
+        if path is None:
+            try:
+                path = self.ask_sys_class_block()
+            except (IOError, OSError) as err:
+                debug(err)
+        # We use exists() because the "/dev/*" part of the path is hard
+        # coded, so we want to be sure.
+        if path is not None and os.path.exists(path):
+            return path
+
+
 def disk_partitions(all=False):
     """Return mounted disk partitions as a list of namedtuples."""
     fstypes = set()
@@ -1216,6 +1290,8 @@ def disk_partitions(all=False):
         device, mountpoint, fstype, opts = partition
         if device == 'none':
             device = ''
+        if device in ("/dev/root", "rootfs"):
+            device = RootFsDeviceFinder().find() or device
         if not all:
             if device == '' or fstype not in fstypes:
                 continue
diff --git a/psutil/tests/test_linux.py b/psutil/tests/test_linux.py
index d2d8e7d4..38322db6 100755
--- a/psutil/tests/test_linux.py
+++ b/psutil/tests/test_linux.py
@@ -28,6 +28,7 @@ from psutil._compat import FileNotFoundError
 from psutil._compat import PY3
 from psutil._compat import u
 from psutil.tests import call_until
+from psutil.tests import GITHUB_ACTIONS
 from psutil.tests import GLOBAL_TIMEOUT
 from psutil.tests import HAS_BATTERY
 from psutil.tests import HAS_CPU_FREQ
@@ -51,6 +52,7 @@ if LINUX:
     from psutil._pslinux import calculate_avail_vmem
     from psutil._pslinux import CLOCK_TICKS
     from psutil._pslinux import open_binary
+    from psutil._pslinux import RootFsDeviceFinder
 
 
 HERE = os.path.abspath(os.path.dirname(__file__))
@@ -1263,6 +1265,68 @@ class TestSystemDiskIoCounters(PsutilTestCase):
             self.assertRaises(NotImplementedError, psutil.disk_io_counters)
 
 
+@unittest.skipIf(not LINUX, "LINUX only")
+class TestRootFsDeviceFinder(PsutilTestCase):
+
+    def setUp(self):
+        dev = os.stat("/").st_dev
+        self.major = os.major(dev)
+        self.minor = os.minor(dev)
+
+    def test_call_methods(self):
+        finder = RootFsDeviceFinder()
+        if os.path.exists("/proc/partitions"):
+            finder.ask_proc_partitions()
+        else:
+            self.assertRaises(FileNotFoundError, finder.ask_proc_partitions)
+        if os.path.exists("/sys/dev/block/%s:%s/uevent" % (
+                self.major, self.minor)):
+            finder.ask_sys_dev_block()
+        else:
+            self.assertRaises(FileNotFoundError, finder.ask_sys_dev_block)
+        finder.ask_sys_class_block()
+
+    @unittest.skipIf(GITHUB_ACTIONS, "unsupported on GITHUB_ACTIONS")
+    def test_comparisons(self):
+        finder = RootFsDeviceFinder()
+        self.assertIsNotNone(finder.find())
+
+        a = b = c = None
+        if os.path.exists("/proc/partitions"):
+            a = finder.ask_proc_partitions()
+        if os.path.exists("/sys/dev/block/%s:%s/uevent" % (
+                self.major, self.minor)):
+            b = finder.ask_sys_class_block()
+        c = finder.ask_sys_dev_block()
+
+        base = a or b or c
+        if base and a:
+            self.assertEqual(base, a)
+        if base and b:
+            self.assertEqual(base, b)
+        if base and c:
+            self.assertEqual(base, c)
+
+    @unittest.skipIf(not which("findmnt"), "findmnt utility not available")
+    @unittest.skipIf(GITHUB_ACTIONS, "unsupported on GITHUB_ACTIONS")
+    def test_against_findmnt(self):
+        psutil_value = RootFsDeviceFinder().find()
+        findmnt_value = sh("findmnt -o SOURCE -rn /")
+        self.assertEqual(psutil_value, findmnt_value)
+
+    def test_disk_partitions_mocked(self):
+        with mock.patch(
+                'psutil._pslinux.cext.disk_partitions',
+                return_value=[('/dev/root', '/', 'ext4', 'rw')]) as m:
+            part = psutil.disk_partitions()[0]
+            assert m.called
+            if not GITHUB_ACTIONS:
+                self.assertNotEqual(part.device, "/dev/root")
+                self.assertEqual(part.device, RootFsDeviceFinder().find())
+            else:
+                self.assertEqual(part.device, "/dev/root")
+
+
 # =====================================================================
 # --- misc
 # =====================================================================