catch errors of system commands

src/mpsd_hpc_tools/quota.py

@@ -45,7 +45,7 @@ def get_ceph_attribute(path: str, attribute: str) -> int:
     return int(output)
 
 
-def df_ceph(path_obj: pathlib.Path) -> Tuple[int, int, int]:
+def df_ceph(path: pathlib.Path) -> Tuple[int, int, int]:
     """
     Given a path (such as '/scratch/fangohr') return the number of bytes
     (size, used, available).
@@ -57,14 +57,17 @@ def df_ceph(path_obj: pathlib.Path) -> Tuple[int, int, int]:
     >>> df_ceph("/scratch/fangohr")
     (25000000000000, 1780686359, 24998219313641)
     """
-    path = str(path_obj)
-    used = get_ceph_attribute(path, "ceph.dir.rbytes")
-    size = get_ceph_attribute(path, "ceph.quota.max_bytes")
-    avail = size - used
+    try:
+        used = get_ceph_attribute(path, "ceph.dir.rbytes")
+        size = get_ceph_attribute(path, "ceph.quota.max_bytes")
+    except:
+        used, size, avail = None, None, None
+    else:
+        avail = size - used
     return size, used, avail
 
 
-def df_mountpoint(path_obj: pathlib.Path) -> Tuple[int, int, int]:
+def df_mountpoint(path: pathlib.Path) -> Tuple[int, int, int]:
     """
     Given a path (such as '/home/fangohr') return the number of bytes
     (size, used, available).
@@ -76,9 +79,12 @@ def df_mountpoint(path_obj: pathlib.Path) -> Tuple[int, int, int]:
     >>> df("/home/fangohr")
     (25000000000000, 1780686359, 24998219313641)
     """
-    path = str(path_obj)
+    size, used, avail = None, None, None
     cmd = ["df", "--block-size=1", path]
-    df_run = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, text=True)
+    try:
+        df_run = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, text=True)
+    except:
+        return None, None, None
     for line in df_run.stdout.splitlines():
         l = line.split()
         if (l[0] == "Filesystem"):