Merge d91541f into ca395f6

reana_commons/utils.py

@@ -187,8 +187,8 @@ def build_caching_info_message(
     return caching_info_message
 
 
-def get_workspace_disk_usage(workspace, summarize=False, block_size=None):
-    """Retrieve disk usage information of a workspace."""
+def get_disk_usage(directory, summarize=False, block_size=None):
+    """Retrieve directory disk usage information."""
     command = ["du"]
     if block_size in ["b", "k"]:
         command.append("-{}".format(block_size))
@@ -198,8 +198,13 @@ def get_workspace_disk_usage(workspace, summarize=False, block_size=None):
         command.append("-s")
     else:
         command.append("-a")
-    command.append(workspace)
-    disk_usage_info = subprocess.check_output(command).decode().split()
+    command.append(directory)
+    return subprocess.check_output(command).decode().split()
+
+
+def get_workspace_disk_usage(workspace, summarize=False, block_size=None):
+    """Retrieve disk usage information of a workspace."""
+    disk_usage_info = get_disk_usage(workspace, summarize, block_size)
     # create pairs of (size, filename)
     filesize_pairs = list(zip(disk_usage_info[::2], disk_usage_info[1::2]))
     filesizes = []
@@ -210,6 +215,15 @@ def get_workspace_disk_usage(workspace, summarize=False, block_size=None):
     return filesizes
 
 
+def get_user_disk_usage(user_path, summarize=True, block_size=None):
+    """Retrieve user disk usage information."""
+    try:
+        size, name = get_disk_usage(user_path, summarize, block_size)
+        return size
+    except Exception:
+        return -1
+
+
 def render_cvmfs_pvc(cvmfs_volume):
     """Render REANA_CVMFS_PVC_TEMPLATE."""
     name = CVMFS_REPOSITORIES[cvmfs_volume]

tests/test_utils.py

@@ -21,6 +21,7 @@
     calculate_job_input_hash,
     click_table_printer,
     format_cmd,
+    get_user_disk_usage,
 )
 
 
@@ -99,6 +100,17 @@ def test_calculate_file_access_time(sample_workflow_workspace):
         assert str(file_path) in access_times
 
 
+def test_get_user_disk_usage(sample_workflow_workspace):
+    """Test get_user_disk_usage."""
+    disk_usage = get_user_disk_usage("a/b/c")
+    assert disk_usage == -1
+    sample_workflow_workspace_path = next(sample_workflow_workspace("sample"))
+    disk_usage = get_user_disk_usage(sample_workflow_workspace_path, block_size="b")
+    assert disk_usage == "32083"
+    disk_usage = get_user_disk_usage(sample_workflow_workspace_path, block_size="k")
+    assert disk_usage == "36"
+
+
 def test_format_cmd():
     """Test format_cmd."""
     test_cmd = "ls -l"