storage: Add Directory.stat() method

3 files changed, 76 insertions, 25 deletions

diff --git a/src/buildstream/storage/_casbaseddirectory.py b/src/buildstream/storage/_casbaseddirectory.py
index 99cf1113f..2762a6684 100644
--- a/src/buildstream/storage/_casbaseddirectory.py
+++ b/src/buildstream/storage/_casbaseddirectory.py
@@ -825,24 +825,58 @@ class CasBasedDirectory(Directory):
 
         return self.__digest
 
+    def _entry_from_path(self, *path, follow_symlinks=False):
+        subdir = self.descend(*path[:-1], follow_symlinks=follow_symlinks)
+        self.__validate_path_component(path[-1])
+        target = subdir.index.get(path[-1])
+        if target is None:
+            raise FileNotFoundError("{} not found in {}".format(path[-1], str(subdir)))
+
+        if follow_symlinks and target.type == _FileType.SYMLINK:
+            linklocation = target.target
+            newpath = linklocation.split(os.path.sep)
+            if os.path.isabs(linklocation):
+                return subdir._find_root()._entry_from_path(*newpath, follow_symlinks=True)
+            return subdir._entry_from_path(*newpath, follow_symlinks=True)
+        else:
+            return target
+
     def exists(self, *path, follow_symlinks=False):
         try:
-            subdir = self.descend(*path[:-1], follow_symlinks=follow_symlinks)
-            self.__validate_path_component(path[-1])
-            target = subdir.index.get(path[-1])
-            if target is not None:
-                if follow_symlinks and target.type == _FileType.SYMLINK:
-                    linklocation = target.target
-                    newpath = linklocation.split(os.path.sep)
-                    if os.path.isabs(linklocation):
-                        return subdir._find_root().exists(*newpath, follow_symlinks=True)
-                    return subdir.exists(*newpath, follow_symlinks=True)
-                else:
-                    return True
-            return False
-        except VirtualDirectoryError:
+            self._entry_from_path(*path, follow_symlinks=follow_symlinks)
+            return True
+        except (VirtualDirectoryError, FileNotFoundError):
             return False
 
+    def stat(self, *path, follow_symlinks=False):
+        entry = self._entry_from_path(*path, follow_symlinks=follow_symlinks)
+
+        st_mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
+        st_nlink = 1
+        st_mtime = BST_ARBITRARY_TIMESTAMP
+
+        if entry.type == _FileType.REGULAR_FILE:
+            st_mode |= stat.S_IFREG
+            st_size = entry.get_digest().size_bytes
+        elif entry.type == _FileType.DIRECTORY:
+            st_mode |= stat.S_IFDIR
+            st_size = 0
+        elif entry.type == _FileType.SYMLINK:
+            st_mode |= stat.S_IFLNK
+            st_size = len(entry.target)
+        else:
+            raise VirtualDirectoryError("Unsupported file type {}".format(entry.type))
+
+        if entry.type == _FileType.DIRECTORY or entry.is_executable:
+            st_mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
+
+        if entry.node_properties:
+            for prop in entry.node_properties:
+                if prop.name == "MTime" and prop.value:
+                    st_mtime = utils._parse_timestamp(prop.value)
+
+        return os.stat_result((st_mode, 0, 0, st_nlink, 0, 0, st_size, st_mtime, st_mtime, st_mtime))
+
     def __iter__(self):
         yield from self.index.keys()
 
diff --git a/src/buildstream/storage/_filebaseddirectory.py b/src/buildstream/storage/_filebaseddirectory.py
index 1799a84b8..99769ce76 100644
--- a/src/buildstream/storage/_filebaseddirectory.py
+++ b/src/buildstream/storage/_filebaseddirectory.py
@@ -236,19 +236,23 @@ class FileBasedDirectory(Directory):
     def get_size(self):
         return utils._get_dir_size(self.external_directory)
 
+    def stat(self, *path, follow_symlinks=False):
+        subdir = self.descend(*path[:-1], follow_symlinks=follow_symlinks)
+        newpath = os.path.join(subdir.external_directory, path[-1])
+        st = os.lstat(newpath)
+        if follow_symlinks and stat.S_ISLNK(st.st_mode):
+            linklocation = os.readlink(newpath)
+            newpath = linklocation.split(os.path.sep)
+            if os.path.isabs(linklocation):
+                return subdir._find_root().stat(*newpath, follow_symlinks=True)
+            return subdir.stat(*newpath, follow_symlinks=True)
+        else:
+            return st
+
     def exists(self, *path, follow_symlinks=False):
         try:
-            subdir = self.descend(*path[:-1], follow_symlinks=follow_symlinks)
-            newpath = os.path.join(subdir.external_directory, path[-1])
-            st = os.lstat(newpath)
-            if follow_symlinks and stat.S_ISLNK(st.st_mode):
-                linklocation = os.readlink(newpath)
-                newpath = linklocation.split(os.path.sep)
-                if os.path.isabs(linklocation):
-                    return subdir._find_root().exists(*newpath, follow_symlinks=True)
-                return subdir.exists(*newpath, follow_symlinks=True)
-            else:
-                return True
+            self.stat(*path, follow_symlinks=follow_symlinks)
+            return True
         except (VirtualDirectoryError, FileNotFoundError):
             return False
 
diff --git a/src/buildstream/storage/directory.py b/src/buildstream/storage/directory.py
index bb9d78f7e..5d85cbe46 100644
--- a/src/buildstream/storage/directory.py
+++ b/src/buildstream/storage/directory.py
@@ -32,6 +32,7 @@ See also: :ref:`sandboxing`.
 """
 
 
+import os
 from typing import Callable, Optional, Union, List
 
 from .._exceptions import BstError
@@ -208,6 +209,18 @@ class Directory:
         """
         raise NotImplementedError()
 
+    def stat(self, *paths: str, follow_symlinks: bool = False) -> os.stat_result:
+        """ Get the status of a file.
+
+        Args:
+          *paths: A list of strings which are all path components.
+          follow_symlinks: True to follow symlinks.
+
+        Returns:
+          A `os.stat_result` object.
+        """
+        raise NotImplementedError()
+
     def open_file(self, *paths: str, mode: str = "r"):
         """ Open file and return a corresponding file object. In text mode,
         UTF-8 is used as encoding.