| # |
| # Copyright (C) 2018 Bloomberg LP |
| # |
| # This program is free software; you can redistribute it and/or |
| # modify it under the terms of the GNU Lesser General Public |
| # License as published by the Free Software Foundation; either |
| # version 2 of the License, or (at your option) any later version. |
| # |
| # This library is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| # Lesser General Public License for more details. |
| # |
| # You should have received a copy of the GNU Lesser General Public |
| # License along with this library. If not, see <http://www.gnu.org/licenses/>. |
| # |
| # Authors: |
| # Jim MacArthur <jim.macarthur@codethink.co.uk> |
| |
| """ |
| CasBasedDirectory |
| ========= |
| |
| Implementation of the Directory class which backs onto a Merkle-tree based content |
| addressable storage system. |
| |
| See also: :ref:`sandboxing`. |
| """ |
| |
| import os |
| import stat |
| import tarfile as tarfilelib |
| from contextlib import contextmanager |
| from io import StringIO |
| from google.protobuf import timestamp_pb2 |
| |
| from .. import utils |
| from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2 |
| from .directory import Directory, VirtualDirectoryError, _FileType |
| from ._filebaseddirectory import FileBasedDirectory |
| from ..utils import FileListResult, BST_ARBITRARY_TIMESTAMP |
| |
| |
| class IndexEntry: |
| """ Directory entry used in CasBasedDirectory.index """ |
| |
| def __init__( |
| self, |
| name, |
| entrytype, |
| *, |
| digest=None, |
| target=None, |
| is_executable=False, |
| buildstream_object=None, |
| modified=False, |
| mtime=None |
| ): |
| self.name = name |
| self.type = entrytype |
| self.digest = digest |
| self.target = target |
| self.is_executable = is_executable |
| self.buildstream_object = buildstream_object |
| self.modified = modified |
| self.mtime = mtime |
| |
| def get_directory(self, parent): |
| if not self.buildstream_object: |
| assert self.type == _FileType.DIRECTORY |
| self.buildstream_object = CasBasedDirectory( |
| parent.cas_cache, digest=self.digest, parent=parent, filename=self.name |
| ) |
| self.digest = None |
| |
| return self.buildstream_object |
| |
| def get_digest(self): |
| if self.buildstream_object: |
| # directory with buildstream object |
| return self.buildstream_object._get_digest() |
| else: |
| # regular file, symlink or directory without buildstream object |
| return self.digest |
| |
| # clone(): |
| # |
| # Create a deep copy of this object. If this is a directory, a |
| # CasBasedDirectory can also be passed to assign an appropriate |
| # parent directory. |
| # |
| def clone(self) -> "IndexEntry": |
| return IndexEntry( |
| self.name, |
| self.type, |
| # If this is a directory, the digest will be converted |
| # later if necessary. For other non-file types, digests |
| # are always None. |
| digest=self.get_digest(), |
| target=self.target, |
| is_executable=self.is_executable, |
| mtime=self.mtime, |
| ) |
| |
| def __eq__(self, other: object) -> bool: |
| if not isinstance(other, IndexEntry): |
| return NotImplemented |
| |
| def get_equivalency_properties(e: IndexEntry): |
| return (e.name, e.type, e.target, e.is_executable, e.mtime, e.get_digest()) |
| |
| return get_equivalency_properties(self) == get_equivalency_properties(other) |
| |
| |
| # CasBasedDirectory intentionally doesn't call its superclass constuctor, |
| # which is meant to be unimplemented. |
| # pylint: disable=super-init-not-called |
| |
| |
| class CasBasedDirectory(Directory): |
| """ |
| CAS-based directories can have two names; one is a 'common name' which has no effect |
| on functionality, and the 'filename'. If a CasBasedDirectory has a parent, then 'filename' |
| must be the name of an entry in the parent directory's index which points to this object. |
| This is used to inform a parent directory that it must update the given hash for this |
| object when this object changes. |
| |
| Typically a top-level CasBasedDirectory will have a common_name and no filename, and |
| subdirectories wil have a filename and no common_name. common_name can used to identify |
| CasBasedDirectory objects in a log file, since they have no unique position in a file |
| system. |
| """ |
| |
| # Two constants which define the separators used by the remote execution API. |
| _pb2_path_sep = "/" |
| _pb2_absolute_path_prefix = "/" |
| |
| def __init__(self, cas_cache, *, digest=None, parent=None, common_name="untitled", filename=None): |
| self.filename = filename |
| self.common_name = common_name |
| self.cas_cache = cas_cache |
| self.__digest = digest |
| self.index = {} |
| self.parent = parent |
| self.__subtree_read_only = None |
| |
| if digest: |
| self._populate_index(digest) |
| |
| # _clear(): |
| # |
| # Remove all entries from this directory. |
| # |
| def _clear(self): |
| self.__invalidate_digest() |
| self.index = {} |
| |
| # _reset(): |
| # |
| # Replace the contents of this directory with the entries from the specified |
| # directory digest. |
| # |
| # Args: |
| # digest (Digest): The digest of the replacement directory |
| # |
| def _reset(self, *, digest=None): |
| self._clear() |
| |
| if digest: |
| self.__digest = digest |
| self._populate_index(digest) |
| |
| def _populate_index(self, digest): |
| try: |
| pb2_directory = remote_execution_pb2.Directory() |
| with open(self.cas_cache.objpath(digest), "rb") as f: |
| pb2_directory.ParseFromString(f.read()) |
| except FileNotFoundError as e: |
| raise VirtualDirectoryError("Directory not found in local cache: {}".format(e)) from e |
| |
| for prop in pb2_directory.node_properties.properties: |
| if prop.name == "SubtreeReadOnly": |
| self.__subtree_read_only = prop.value == "true" |
| |
| for entry in pb2_directory.directories: |
| self.index[entry.name] = IndexEntry(entry.name, _FileType.DIRECTORY, digest=entry.digest) |
| for entry in pb2_directory.files: |
| if entry.node_properties.HasField("mtime"): |
| mtime = entry.node_properties.mtime |
| else: |
| mtime = None |
| |
| self.index[entry.name] = IndexEntry( |
| entry.name, |
| _FileType.REGULAR_FILE, |
| digest=entry.digest, |
| is_executable=entry.is_executable, |
| mtime=mtime, |
| ) |
| for entry in pb2_directory.symlinks: |
| self.index[entry.name] = IndexEntry(entry.name, _FileType.SYMLINK, target=entry.target) |
| |
| def _find_self_in_parent(self): |
| assert self.parent is not None |
| parent = self.parent |
| for (k, v) in parent.index.items(): |
| if v.buildstream_object == self: |
| return k |
| return None |
| |
| def _add_directory(self, name): |
| assert name not in self.index |
| |
| newdir = CasBasedDirectory(self.cas_cache, parent=self, filename=name) |
| |
| self.index[name] = IndexEntry(name, _FileType.DIRECTORY, buildstream_object=newdir) |
| |
| self.__invalidate_digest() |
| |
| return newdir |
| |
| def _add_file(self, name, path, modified=False, can_link=False, properties=None): |
| digest = self.cas_cache.add_object(path=path) |
| is_executable = os.access(path, os.X_OK) |
| mtime = None |
| if properties and "mtime" in properties: |
| mtime = timestamp_pb2.Timestamp() |
| utils._get_file_protobuf_mtimestamp(mtime, path) |
| |
| entry = IndexEntry( |
| name, |
| _FileType.REGULAR_FILE, |
| digest=digest, |
| is_executable=is_executable, |
| modified=modified or name in self.index, |
| mtime=mtime, |
| ) |
| self.index[name] = entry |
| |
| self.__invalidate_digest() |
| |
| def _add_entry(self, entry: IndexEntry): |
| self.index[entry.name] = entry.clone() |
| self.__invalidate_digest() |
| |
| def _contains_entry(self, entry: IndexEntry) -> bool: |
| return entry == self.index.get(entry.name) |
| |
| # _apply_changes(): |
| # |
| # Apply changes from dir_a to dir_b to this directory. The use |
| # case for this is to merge changes between different workspace |
| # versions into a buildtree. |
| # |
| # If a change was made both to this directory, as well as between |
| # the given directories, it is applied, overwriting any changes to |
| # this directory. This is desirable because we want to keep user |
| # changes, however it may need to be re-considered for other use |
| # cases. |
| # |
| # We perform this computation this way, instead of with a _diff |
| # method and a subsequent _apply_diff, because it prevents leaking |
| # IndexEntry objects, which contain mutable references and may |
| # therefore cause problems if used outside of this class. |
| # |
| # Args: |
| # dir_a: The directory from which to start computing differences. |
| # dir_b: The directory whose changes to apply |
| # |
| def _apply_changes(self, dir_a: "CasBasedDirectory", dir_b: "CasBasedDirectory"): |
| # If the digests are the same, the directories are the same |
| # (child properties affect the digest). We can skip any work |
| # in such a case. |
| if dir_a._get_digest() == dir_b._get_digest(): |
| return |
| |
| def get_subdir(entry: IndexEntry, directory: CasBasedDirectory) -> CasBasedDirectory: |
| return directory.index[entry.name].get_directory(directory) |
| |
| def is_dir_in(entry: IndexEntry, directory: CasBasedDirectory) -> bool: |
| return directory.index[entry.name].type == _FileType.DIRECTORY |
| |
| # We first check which files were added, and add them to our |
| # directory. |
| for entry in dir_b.index.values(): |
| if self._contains_entry(entry): |
| # We can short-circuit checking entries from b that |
| # already exist in our index. |
| continue |
| |
| if not dir_a._contains_entry(entry): |
| if entry.name in self.index and is_dir_in(entry, self) and is_dir_in(entry, dir_b): |
| # If the entry changed, and is a directory in both |
| # the current and to-merge-into tree, we need to |
| # merge recursively. |
| |
| # If the entry is not a directory in dir_a, we |
| # want to overwrite the file, but we need an empty |
| # directory for recursion. |
| if entry.name in dir_a.index and is_dir_in(entry, dir_a): |
| sub_a = get_subdir(entry, dir_a) |
| else: |
| sub_a = CasBasedDirectory(dir_a.cas_cache) |
| |
| subdir = get_subdir(entry, self) |
| subdir._apply_changes(sub_a, get_subdir(entry, dir_b)) |
| else: |
| # In any other case, we just add/overwrite the file/directory |
| self._add_entry(entry) |
| |
| # We can't iterate and remove entries at the same time |
| to_remove = [entry for entry in dir_a.index.values() if entry.name not in dir_b.index] |
| for entry in to_remove: |
| self.remove(entry.name, recursive=True) |
| |
| self.__invalidate_digest() |
| |
| def _add_new_link_direct(self, name, target): |
| self.index[name] = IndexEntry(name, _FileType.SYMLINK, target=target, modified=name in self.index) |
| |
| self.__invalidate_digest() |
| |
| def remove(self, *path, recursive=False): |
| if len(path) > 1: |
| # Delegate remove to subdirectory |
| subdir = self.descend(*path[:-1]) |
| subdir.remove(path[-1], recursive=recursive) |
| return |
| |
| name = path[0] |
| self.__validate_path_component(name) |
| entry = self.index.get(name) |
| if not entry: |
| raise FileNotFoundError("{} not found in {}".format(name, str(self))) |
| |
| if entry.type == _FileType.DIRECTORY and not recursive: |
| subdir = entry.get_directory(self) |
| if not subdir.is_empty(): |
| raise VirtualDirectoryError("{} is not empty".format(str(subdir))) |
| |
| del self.index[name] |
| self.__invalidate_digest() |
| |
| def rename(self, src, dest): |
| srcdir = self.descend(*src[:-1]) |
| entry = srcdir._entry_from_path(src[-1]) |
| |
| destdir = self.descend(*dest[:-1]) |
| self.__validate_path_component(dest[-1]) |
| |
| srcdir.remove(src[-1], recursive=True) |
| entry.name = dest[-1] |
| destdir._add_entry(entry) |
| |
| def descend(self, *paths, create=False, follow_symlinks=False): |
| """Descend one or more levels of directory hierarchy and return a new |
| Directory object for that directory. |
| |
| Arguments: |
| * *paths (str): A list of strings which are all directory names. |
| * create (boolean): If this is true, the directories will be created if |
| they don't already exist. |
| |
| Note: At the moment, creating a directory by descending does |
| not update this object in the CAS cache. However, performing |
| an import_files() into a subdirectory of any depth obtained by |
| descending from this object *will* cause this directory to be |
| updated and stored. |
| |
| """ |
| |
| current_dir = self |
| paths = list(paths) |
| |
| for path in paths: |
| # Skip empty path segments |
| if not path: |
| continue |
| |
| self.__validate_path_component(path) |
| |
| entry = current_dir.index.get(path) |
| |
| if entry: |
| if entry.type == _FileType.DIRECTORY: |
| current_dir = entry.get_directory(current_dir) |
| elif follow_symlinks and entry.type == _FileType.SYMLINK: |
| linklocation = entry.target |
| newpaths = linklocation.split(os.path.sep) |
| if os.path.isabs(linklocation): |
| current_dir = current_dir._find_root().descend(*newpaths, follow_symlinks=True) |
| else: |
| current_dir = current_dir.descend(*newpaths, follow_symlinks=True) |
| else: |
| error = "Cannot descend into {}, which is a '{}' in the directory {}" |
| raise VirtualDirectoryError( |
| error.format(path, current_dir.index[path].type, current_dir), reason="not-a-directory" |
| ) |
| else: |
| if path == ".": |
| continue |
| if path == "..": |
| if current_dir.parent is not None: |
| current_dir = current_dir.parent |
| # In POSIX /.. == / so just stay at the root dir |
| continue |
| if create: |
| current_dir = current_dir._add_directory(path) |
| else: |
| error = "'{}' not found in {}" |
| raise VirtualDirectoryError(error.format(path, str(current_dir)), reason="directory-not-found") |
| |
| return current_dir |
| |
| def _check_replacement(self, name, relative_pathname, fileListResult): |
| """ Checks whether 'name' exists, and if so, whether we can overwrite it. |
| If we can, add the name to 'overwritten_files' and delete the existing entry. |
| Returns 'True' if the import should go ahead. |
| fileListResult.overwritten and fileListResult.ignore are updated depending |
| on the result. """ |
| existing_entry = self.index.get(name) |
| if existing_entry is None: |
| return True |
| elif existing_entry.type == _FileType.DIRECTORY: |
| # If 'name' maps to a DirectoryNode, then there must be an entry in index |
| # pointing to another Directory. |
| subdir = existing_entry.get_directory(self) |
| if subdir.is_empty(): |
| self.remove(name) |
| fileListResult.overwritten.append(relative_pathname) |
| return True |
| else: |
| # We can't overwrite a non-empty directory, so we just ignore it. |
| fileListResult.ignored.append(relative_pathname) |
| return False |
| else: |
| self.remove(name) |
| fileListResult.overwritten.append(relative_pathname) |
| return True |
| |
| def _partial_import_cas_into_cas(self, source_directory, filter_callback, *, path_prefix="", origin=None, result): |
| """ Import files from a CAS-based directory. """ |
| if origin is None: |
| origin = self |
| |
| for name, entry in source_directory.index.items(): |
| # The destination filename, relative to the root where the import started |
| relative_pathname = os.path.join(path_prefix, name) |
| |
| is_dir = entry.type == _FileType.DIRECTORY |
| |
| if is_dir: |
| create_subdir = name not in self.index |
| |
| if create_subdir and not filter_callback: |
| # If subdirectory does not exist yet and there is no filter, |
| # we can import the whole source directory by digest instead |
| # of importing each directory entry individually. |
| subdir_digest = entry.get_digest() |
| dest_entry = IndexEntry(name, _FileType.DIRECTORY, digest=subdir_digest) |
| self.index[name] = dest_entry |
| self.__invalidate_digest() |
| |
| # However, we still need to iterate over the directory entries |
| # to fill in `result.files_written`. |
| |
| # Use source subdirectory object if it already exists, |
| # otherwise create object for destination subdirectory. |
| # This is based on the assumption that the destination |
| # subdirectory is more likely to be modified later on |
| # (e.g., by further import_files() calls). |
| if entry.buildstream_object: |
| subdir = entry.buildstream_object |
| else: |
| subdir = dest_entry.get_directory(self) |
| |
| subdir.__add_files_to_result(path_prefix=relative_pathname, result=result) |
| else: |
| src_subdir = source_directory.descend(name) |
| if src_subdir == origin: |
| continue |
| |
| try: |
| dest_subdir = self.descend(name, create=create_subdir) |
| except VirtualDirectoryError: |
| filetype = self.index[name].type |
| raise VirtualDirectoryError( |
| "Destination is a {}, not a directory: /{}".format(filetype, relative_pathname) |
| ) |
| |
| dest_subdir._partial_import_cas_into_cas( |
| src_subdir, filter_callback, path_prefix=relative_pathname, origin=origin, result=result |
| ) |
| |
| if filter_callback and not filter_callback(relative_pathname): |
| if is_dir and create_subdir and dest_subdir.is_empty(): |
| # Complete subdirectory has been filtered out, remove it |
| self.remove(name) |
| |
| # Entry filtered out, move to next |
| continue |
| |
| if not is_dir: |
| if self._check_replacement(name, relative_pathname, result): |
| if entry.type == _FileType.REGULAR_FILE: |
| self._add_entry(entry) |
| self.index[entry.name].modified = True |
| else: |
| assert entry.type == _FileType.SYMLINK |
| self._add_new_link_direct(name=name, target=entry.target) |
| result.files_written.append(relative_pathname) |
| |
| def import_files( |
| self, |
| external_pathspec, |
| *, |
| filter_callback=None, |
| report_written=True, |
| update_mtime=None, |
| can_link=False, |
| properties=None |
| ): |
| """ See superclass Directory for arguments """ |
| |
| result = FileListResult() |
| |
| if isinstance(external_pathspec, FileBasedDirectory): |
| external_pathspec = external_pathspec._get_underlying_directory() |
| |
| if isinstance(external_pathspec, str): |
| # Import files from local filesystem by first importing complete |
| # directory into CAS (using buildbox-casd) and then importing its |
| # content into this CasBasedDirectory using CAS-to-CAS import |
| # to write the report, handle possible conflicts (if the target |
| # directory is not empty) and apply the optional filter. |
| digest = self.cas_cache.import_directory(external_pathspec, properties=properties) |
| external_pathspec = CasBasedDirectory(self.cas_cache, digest=digest) |
| |
| assert isinstance(external_pathspec, CasBasedDirectory) |
| self._partial_import_cas_into_cas(external_pathspec, filter_callback, result=result) |
| |
| # TODO: No notice is taken of report_written or update_mtime. |
| # Current behaviour is to fully populate the report, which is inefficient, |
| # but still correct. |
| |
| return result |
| |
| def import_single_file(self, external_pathspec, properties=None): |
| result = FileListResult() |
| if self._check_replacement(os.path.basename(external_pathspec), os.path.dirname(external_pathspec), result): |
| self._add_file( |
| os.path.basename(external_pathspec), |
| external_pathspec, |
| modified=os.path.basename(external_pathspec) in result.overwritten, |
| properties=properties, |
| ) |
| result.files_written.append(external_pathspec) |
| return result |
| |
| def set_deterministic_user(self): |
| """ Sets all files in this directory to the current user's euid/egid. |
| We also don't store user data, so this can be ignored. |
| """ |
| |
| def export_files(self, to_directory, *, can_link=False, can_destroy=False): |
| """Copies everything from this into to_directory, which must be the name |
| of a traditional filesystem directory. |
| |
| Arguments: |
| |
| to_directory (string): a path outside this directory object |
| where the contents will be copied to. |
| |
| can_link (bool): Whether we can create hard links in to_directory |
| instead of copying. |
| |
| can_destroy (bool): Whether we can destroy elements in this |
| directory to export them (e.g. by renaming them as the |
| target). |
| |
| """ |
| |
| self.cas_cache.checkout(to_directory, self._get_digest(), can_link=can_link) |
| |
| def export_to_tar(self, tarfile, destination_dir, mtime=BST_ARBITRARY_TIMESTAMP): |
| for filename, entry in sorted(self.index.items()): |
| arcname = os.path.join(destination_dir, filename) |
| if entry.type == _FileType.DIRECTORY: |
| tarinfo = tarfilelib.TarInfo(arcname) |
| tarinfo.mtime = mtime |
| tarinfo.type = tarfilelib.DIRTYPE |
| tarinfo.mode = 0o755 |
| tarfile.addfile(tarinfo) |
| self.descend(filename).export_to_tar(tarfile, arcname, mtime) |
| elif entry.type == _FileType.REGULAR_FILE: |
| source_name = self.cas_cache.objpath(entry.digest) |
| tarinfo = tarfilelib.TarInfo(arcname) |
| tarinfo.mtime = mtime |
| if entry.is_executable: |
| tarinfo.mode |= stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH |
| tarinfo.size = os.path.getsize(source_name) |
| with open(source_name, "rb") as f: |
| tarfile.addfile(tarinfo, f) |
| elif entry.type == _FileType.SYMLINK: |
| tarinfo = tarfilelib.TarInfo(arcname) |
| tarinfo.mtime = mtime |
| tarinfo.mode = 0o777 |
| tarinfo.linkname = entry.target |
| tarinfo.type = tarfilelib.SYMTYPE |
| f = StringIO(entry.target) |
| tarfile.addfile(tarinfo, f) |
| else: |
| raise VirtualDirectoryError("can not export file type {} to tar".format(entry.type)) |
| |
| def _mark_changed(self): |
| """ It should not be possible to externally modify a CAS-based |
| directory at the moment.""" |
| raise NotImplementedError() |
| |
| def is_empty(self): |
| """ Return true if this directory has no files, subdirectories or links in it. |
| """ |
| return len(self.index) == 0 |
| |
| def _mark_directory_unmodified(self): |
| # Marks all entries in this directory and all child directories as unmodified. |
| for i in self.index.values(): |
| i.modified = False |
| if i.type == _FileType.DIRECTORY and i.buildstream_object: |
| i.buildstream_object._mark_directory_unmodified() |
| |
| def _mark_entry_unmodified(self, name): |
| # Marks an entry as unmodified. If the entry is a directory, it will |
| # recursively mark all its tree as unmodified. |
| self.index[name].modified = False |
| if self.index[name].buildstream_object: |
| self.index[name].buildstream_object._mark_directory_unmodified() |
| |
| def mark_unmodified(self): |
| """ Marks all files in this directory (recursively) as unmodified. |
| If we have a parent, we mark our own entry as unmodified in that parent's |
| index. |
| """ |
| if self.parent: |
| self.parent._mark_entry_unmodified(self._find_self_in_parent()) |
| else: |
| self._mark_directory_unmodified() |
| |
| def _lightweight_resolve_to_index(self, path): |
| """A lightweight function for transforming paths into IndexEntry |
| objects. This does not follow symlinks. |
| |
| path: The string to resolve. This should be a series of path |
| components separated by the protocol buffer path separator |
| _pb2_path_sep. |
| |
| Returns: the IndexEntry found, or None if any of the path components were not present. |
| |
| """ |
| directory = self |
| path_components = path.split(CasBasedDirectory._pb2_path_sep) |
| for component in path_components[:-1]: |
| if component not in directory.index: |
| return None |
| if directory.index[component].type == _FileType.DIRECTORY: |
| directory = directory.index[component].get_directory(self) |
| else: |
| return None |
| return directory.index.get(path_components[-1], None) |
| |
| def list_modified_paths(self): |
| """Provide a list of relative paths which have been modified since the |
| last call to mark_unmodified. |
| |
| Return value: List(str) - list of modified paths |
| """ |
| |
| for p in self.list_relative_paths(): |
| i = self._lightweight_resolve_to_index(p) |
| if i and i.modified: |
| yield p |
| |
| def list_relative_paths(self): |
| """Provide a list of all relative paths. |
| |
| Yields: |
| (List(str)) - list of all files with relative paths. |
| |
| """ |
| yield from self._list_prefixed_relative_paths() |
| |
| def _list_prefixed_relative_paths(self, prefix=""): |
| """Provide a list of all relative paths. |
| |
| Arguments: |
| prefix (str): an optional prefix to the relative paths, this is |
| also emitted by itself. |
| |
| Yields: |
| (List(str)) - list of all files with relative paths. |
| |
| """ |
| |
| file_list = list(filter(lambda i: i[1].type != _FileType.DIRECTORY, self.index.items())) |
| directory_list = filter(lambda i: i[1].type == _FileType.DIRECTORY, self.index.items()) |
| |
| if prefix != "": |
| yield prefix |
| |
| for (k, v) in sorted(file_list): |
| yield os.path.join(prefix, k) |
| |
| for (k, v) in sorted(directory_list): |
| subdir = v.get_directory(self) |
| yield from subdir._list_prefixed_relative_paths(prefix=os.path.join(prefix, k)) |
| |
| def walk(self): |
| """Provide a list of dictionaries containing information about the files. |
| |
| Yields: |
| info (dict) - a dictionary containing name, type and size of the files. |
| |
| """ |
| yield from self._walk() |
| |
| def _walk(self, prefix=""): |
| """ Walk through the files, collecting the required data |
| |
| Arguments: |
| prefix (str): an optional prefix to the relative paths, this is |
| also emitted by itself. |
| |
| Yields: |
| info (dict) - a dictionary containing name, type and size of the files. |
| |
| """ |
| for leaf in sorted(self.index.keys()): |
| entry = self.index[leaf] |
| info = {"name": os.path.join(prefix, leaf), "type": entry.type} |
| if entry.type == _FileType.REGULAR_FILE: |
| info["executable"] = entry.is_executable |
| info["size"] = self.get_size() |
| elif entry.type == _FileType.SYMLINK: |
| info["target"] = entry.target |
| info["size"] = len(entry.target) |
| if entry.type == _FileType.DIRECTORY: |
| directory = entry.get_directory(self) |
| info["size"] = len(directory.index) |
| yield info |
| yield from directory._walk(os.path.join(prefix, leaf)) |
| else: |
| yield info |
| |
| def get_size(self): |
| digest = self._get_digest() |
| total = digest.size_bytes |
| for i in self.index.values(): |
| if i.type == _FileType.DIRECTORY: |
| subdir = i.get_directory(self) |
| total += subdir.get_size() |
| elif i.type == _FileType.REGULAR_FILE: |
| total += i.digest.size_bytes |
| # Symlink nodes are encoded as part of the directory serialization. |
| return total |
| |
| def _get_identifier(self): |
| path = "" |
| if self.parent: |
| path = self.parent._get_identifier() |
| if self.filename: |
| path += "/" + self.filename |
| else: |
| path += "/" + self.common_name |
| return path |
| |
| @contextmanager |
| def open_file(self, *path: str, mode: str = "r"): |
| subdir = self.descend(*path[:-1]) |
| self.__validate_path_component(path[-1]) |
| entry = subdir.index.get(path[-1]) |
| |
| if entry and entry.type != _FileType.REGULAR_FILE: |
| raise VirtualDirectoryError("{} in {} is not a file".format(path[-1], str(subdir))) |
| |
| if mode not in ["r", "rb", "w", "wb", "w+", "w+b", "x", "xb", "x+", "x+b"]: |
| raise ValueError("Unsupported mode: `{}`".format(mode)) |
| |
| if "b" in mode: |
| encoding = None |
| else: |
| encoding = "utf-8" |
| |
| if "r" in mode: |
| if not entry: |
| raise FileNotFoundError("{} not found in {}".format(path[-1], str(subdir))) |
| |
| # Read-only access, allow direct access to CAS object |
| with open(self.cas_cache.objpath(entry.digest), mode, encoding=encoding) as f: |
| yield f |
| else: |
| if "x" in mode and entry: |
| raise FileExistsError("{} already exists in {}".format(path[-1], str(subdir))) |
| |
| with utils._tempnamedfile(mode, encoding=encoding, dir=self.cas_cache.tmpdir) as f: |
| # Make sure the temporary file is readable by buildbox-casd |
| os.chmod(f.name, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH) |
| yield f |
| # Import written temporary file into CAS |
| f.flush() |
| subdir._add_file(path[-1], f.name, modified=True) |
| |
| def __str__(self): |
| return "[CAS:{}]".format(self._get_identifier()) |
| |
| def _get_underlying_directory(self): |
| """ There is no underlying directory for a CAS-backed directory, so |
| throw an exception. """ |
| raise VirtualDirectoryError( |
| "_get_underlying_directory was called on a CAS-backed directory," + " which has no underlying directory." |
| ) |
| |
| def _find_root(self): |
| """ Finds the root of this directory tree by following 'parent' until there is |
| no parent. """ |
| if self.parent: |
| return self.parent._find_root() |
| else: |
| return self |
| |
| # _get_digest(): |
| # |
| # Return the Digest for this directory. |
| # |
| # Returns: |
| # (Digest): The Digest protobuf object for the Directory protobuf |
| # |
| def _get_digest(self): |
| if not self.__digest: |
| # Create updated Directory proto |
| pb2_directory = remote_execution_pb2.Directory() |
| |
| if self.__subtree_read_only is not None: |
| node_property = pb2_directory.node_properties.properties.add() |
| node_property.name = "SubtreeReadOnly" |
| node_property.value = "true" if self.__subtree_read_only else "false" |
| |
| for name, entry in sorted(self.index.items()): |
| if entry.type == _FileType.DIRECTORY: |
| dirnode = pb2_directory.directories.add() |
| dirnode.name = name |
| |
| # Update digests for subdirectories in DirectoryNodes. |
| # No need to call entry.get_directory(). |
| # If it hasn't been instantiated, digest must be up-to-date. |
| subdir = entry.buildstream_object |
| if subdir: |
| dirnode.digest.CopyFrom(subdir._get_digest()) |
| else: |
| dirnode.digest.CopyFrom(entry.digest) |
| elif entry.type == _FileType.REGULAR_FILE: |
| filenode = pb2_directory.files.add() |
| filenode.name = name |
| filenode.digest.CopyFrom(entry.digest) |
| filenode.is_executable = entry.is_executable |
| if entry.mtime is not None: |
| filenode.node_properties.mtime.CopyFrom(entry.mtime) |
| elif entry.type == _FileType.SYMLINK: |
| symlinknode = pb2_directory.symlinks.add() |
| symlinknode.name = name |
| symlinknode.target = entry.target |
| |
| self.__digest = self.cas_cache.add_object(buffer=pb2_directory.SerializeToString()) |
| |
| 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: |
| 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.mtime is not None: |
| st_mtime = utils._parse_protobuf_timestamp(entry.mtime) |
| |
| return os.stat_result((st_mode, 0, 0, st_nlink, 0, 0, st_size, st_mtime, st_mtime, st_mtime)) |
| |
| def file_digest(self, *path): |
| entry = self._entry_from_path(*path) |
| if entry.type != _FileType.REGULAR_FILE: |
| raise VirtualDirectoryError("Unsupported file type for digest: {}".format(entry.type)) |
| |
| return entry.digest.hash |
| |
| def readlink(self, *path): |
| entry = self._entry_from_path(*path) |
| if entry.type != _FileType.SYMLINK: |
| raise VirtualDirectoryError("Unsupported file type for readlink: {}".format(entry.type)) |
| |
| return entry.target |
| |
| def __iter__(self): |
| yield from self.index.keys() |
| |
| def _set_subtree_read_only(self, read_only): |
| self.__subtree_read_only = read_only |
| |
| self.__invalidate_digest() |
| |
| def __invalidate_digest(self): |
| if self.__digest: |
| self.__digest = None |
| if self.parent: |
| self.parent.__invalidate_digest() |
| |
| def __add_files_to_result(self, *, path_prefix="", result): |
| for name, entry in self.index.items(): |
| # The destination filename, relative to the root where the import started |
| relative_pathname = os.path.join(path_prefix, name) |
| |
| if entry.type == _FileType.DIRECTORY: |
| subdir = self.descend(name) |
| subdir.__add_files_to_result(path_prefix=relative_pathname, result=result) |
| else: |
| result.files_written.append(relative_pathname) |
| |
| def __validate_path_component(self, path): |
| if "/" in path: |
| raise VirtualDirectoryError("Invalid path component: '{}'".format(path)) |
