# SPDX-License-Identifier: BSD-2-Clause """ This module provides support for directory states. """ # Copyright (C) 2020, 2023 embedded brains GmbH & Co. KG # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions # are met: # 1. Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # 2. Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in the # documentation and/or other materials provided with the distribution. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. import base64 import fnmatch import hashlib import json import logging import os from pathlib import Path import shutil import tarfile from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, \ Set, Tuple, Union from rtemsspec.items import Item, ItemGetValueContext, Link from rtemsspec.packagebuild import BuildItem, BuildItemFactory, \ PackageBuildDirector from rtemsspec.util import hash_file _Path = Union[Path, str] def _get_file_path(ctx: ItemGetValueContext) -> str: index = max(ctx.index, 0) return f"{ctx.item['directory']}/{ctx.item['files'][index]['file']}" def _get_file_path_without_extension(ctx: ItemGetValueContext) -> str: return os.path.splitext(_get_file_path(ctx))[0] def _file_nop(_source: _Path, _target: _Path) -> None: pass class DirectoryState(BuildItem): """ Maintains a directory state. """ # pylint: disable=too-many-public-methods @classmethod def prepare_factory(cls, factory: BuildItemFactory, type_name: str) -> None: BuildItem.prepare_factory(factory, type_name) factory.add_get_value(f"{type_name}:/file", _get_file_path) factory.add_get_value(f"{type_name}:/file-without-extension", _get_file_path_without_extension) def __init__(self, director: PackageBuildDirector, item: Item): super().__init__(director, item) self._discarded_files: Set[str] = set() self._files: Dict[str, Union[str, None]] = dict( (file_info["file"], file_info["hash"]) for file_info in item["files"]) def __iter__(self): yield from self.files() @property def directory(self) -> str: """ Returns the base directory of the directory state. """ return self["directory"] @property def digest(self) -> str: the_digest = self.item["hash"] if the_digest is None: raise ValueError(f"{self.uid}: directory state hash is not set") return the_digest def _get_hash(self, _base: str, relative_file_path: str) -> str: digest = self._files[relative_file_path] assert digest is not None return digest def _hash_file(self, base: str, relative_file_path: str) -> str: file_path = os.path.join(base, relative_file_path) digest = hash_file(file_path) logging.debug("%s: file '%s' hash is %s", self.uid, file_path, digest) self._files[relative_file_path] = digest return digest def _add_hashes(self, base: str, hash_file_handler: Callable[[str, str], str]) -> str: overall_hash = hashlib.sha512() overall_hash.update(base.encode("utf-8")) for relative_file_path in sorted(self._files): digest = hash_file_handler(base, relative_file_path) overall_hash.update(relative_file_path.encode("utf-8")) overall_hash.update(digest.encode("utf-8")) self._update_item_files() digest = base64.urlsafe_b64encode( overall_hash.digest()).decode("ascii") logging.info("%s: directory '%s' hash is %s", self.uid, base, digest) self.item["hash"] = digest return digest def _directory_state_exclude(self, base: str, files: Set[str]) -> None: for exclude_item in self.item.parents("directory-state-exclude"): exclude_state = self.director[exclude_item.uid] assert isinstance(exclude_state, DirectoryState) exclude_files = files.intersection( os.path.relpath(path, base) for path in exclude_state) logging.info( "%s: exclude files of directory state %s: %s", self.uid, exclude_item.uid, [os.path.join(base, path) for path in sorted(exclude_files)]) files.difference_update(exclude_files) def _load_from_patterns(self, base: str, patterns: List[Dict[str, Any]]) -> None: logging.info("%s: load pattern defined directory state: %s", self.uid, base) files: Set[str] = set() base_path = Path(base) for include_exclude in patterns: include = include_exclude["include"] logging.info("%s: add files matching '%s' in: %s", self.uid, include, base) more = set( os.path.relpath(path, base) for path in base_path.glob(include) if not path.is_dir()) for exclude in include_exclude["exclude"]: exclude_files = set( path for path in more if fnmatch.fnmatch(os.path.join("/", path), exclude)) logging.info("%s: exclude files for pattern '%s': %s", self.uid, exclude, [ os.path.join(base, path) for path in sorted(exclude_files) ]) more.difference_update(exclude_files) files.update(more) self._directory_state_exclude(base, files) self._files = dict.fromkeys(files, None) def load(self) -> str: """ Loads the directory state and returns the overall hash. """ base = self.directory patterns = self.item["patterns"] if patterns: self._load_from_patterns(base, patterns) else: logging.info("%s: load explicit directory state: %s", self.uid, base) return self._add_hashes(base, self._hash_file) def lazy_load(self) -> str: """ Loads the directory state if the overall hash is not present and returns the overall hash. """ digest = self.item["hash"] if digest is not None: return digest return self.load() @property def file(self) -> str: """ Is the path of the first file of the file state. """ return next(self.files()) def files(self, base: Optional[str] = None) -> Iterator[str]: """ Yields the file paths of the directory state. """ if base is None: base = self.directory for file_path in sorted(self._files): yield os.path.join(base, file_path) def files_and_hashes( self, base: Optional[str] = None) -> Iterator[Tuple[str, Optional[str]]]: """ Yields the file paths and hashes of the directory state. """ if base is None: base = self.directory for file_path, file_hash in sorted(self._files.items()): yield os.path.join(base, file_path), file_hash def compact(self) -> None: """ Removes the common prefix from the files and adds it to the base directory. """ prefix = os.path.commonprefix(list(self._files.keys())).rstrip("/") if prefix and not os.path.isabs(prefix): self.item["directory"] = os.path.join(self.item["directory"], prefix) self.item["hash"] = None self._files = dict( (os.path.relpath(path, prefix), None) for path in self._files) self._update_item_files() def _update_item_files(self): self.item["files"] = list({ "file": path, "hash": digest } for path, digest in sorted(self._files.items())) def clear(self) -> None: """ Clears the file set of the directory state. """ logging.info("%s: clear directory state", self.uid) self.item["hash"] = None self._files.clear() self._update_item_files() def invalidate(self) -> None: """ Invalidates the directory state. """ logging.info("%s: invalidate directory state", self.uid) self.item["hash"] = None if self.item["patterns"]: self._files.clear() else: self._files = dict.fromkeys(self._files.keys(), None) self._update_item_files() def remove_files(self) -> None: """ Removes the files of the directory state. """ for file in self.files(): try: logging.info("%s: remove: %s", self.uid, file) os.remove(file) except FileNotFoundError: if self.item["patterns"]: logging.warning("%s: file not found: %s", self.uid, file) else: logging.debug("%s: file not found: %s", self.uid, file) def add_files(self, files: Iterable[_Path]) -> None: """ Adds the files to the file set of the directory state. """ self.item["hash"] = None more = set(os.path.normpath(name) for name in files) self._directory_state_exclude(self.directory, more) self._files.update(dict.fromkeys(more, None)) self._update_item_files() def set_files(self, files: Iterable[_Path]) -> None: """ Sets the file set of the directory state to the files. """ self.clear() self.add_files(files) def _copy_file(self, source: _Path, target: _Path) -> None: logging.info("%s: copy '%s' to '%s'", self.uid, source, target) os.makedirs(os.path.dirname(target), exist_ok=True) shutil.copy2(source, target) def _move_file(self, source: _Path, target: _Path) -> None: logging.info("%s: move '%s' to '%s'", self.uid, source, target) os.makedirs(os.path.dirname(target), exist_ok=True) os.replace(source, target) def copy_file(self, source: _Path, target: _Path) -> None: """ Copies the file from the source path to the target path. Adds the target file to the file set of the directory state. The target path is relative to the base directory of the directory state. """ self._copy_file(source, os.path.join(self.directory, target)) self.add_files([target]) def copy_files(self, root_dir: _Path, files: Iterable[_Path], prefix: _Path = ".") -> None: """ Copies the files relative to the root directory to the base directory of the directory state using the prefix. The base directory of the directory state and the prefix is prepended to the file path for each file before it is added to the directory state. Adds the target files to the file set of the directory state. """ file_list: List[str] = [] base = self.directory for name in files: file_source = os.path.join(root_dir, name) file_list_path = os.path.join(prefix, name) file_list.append(file_list_path) file_target = os.path.join(base, file_list_path) self._copy_file(file_source, file_target) self.add_files(file_list) def _add_tree(self, root_dir: _Path, prefix: _Path, file_op: Callable[[_Path, _Path], None], excludes: Optional[List[str]] = None) -> None: file_list: List[str] = [] base = self.directory for path, _, files in os.walk(os.path.abspath(root_dir)): for name in files: file_source = os.path.join(path, name) file_list_path = os.path.join( prefix, os.path.relpath(file_source, root_dir)) file_target = os.path.join(base, file_list_path) if excludes is None: file_list.append(file_list_path) file_op(file_source, file_target) else: match_path = os.path.normpath( os.path.join("/", file_list_path)) for exclude in excludes: if fnmatch.fnmatch(match_path, exclude): logging.info( "%s: exclude file for pattern '%s': %s", self.uid, exclude, file_target) break else: file_list.append(file_list_path) file_op(file_source, file_target) self.add_files(file_list) def add_tree(self, root_dir: _Path, prefix: _Path = ".", excludes: Optional[List[str]] = None) -> None: """ Adds the files of the directory tree starting at the root directory to the file set of the directory state. The added file path is relative to the root directory. The prefix is prepended to the file path for each file before it is added to the directory state. The files are not copied or moved. """ self._add_tree(root_dir, prefix, _file_nop, excludes) def copy_tree(self, root_dir: _Path, prefix: _Path = ".", excludes: Optional[List[str]] = None) -> None: """ Adds the files of the directory tree starting at the root directory to the file set of the directory state. The added file path is relative to the root directory. The prefix is prepended to the file path for each file before it is added to the directory state. The files are copied. """ self._add_tree(root_dir, prefix, self._copy_file, excludes) def move_tree(self, root_dir: _Path, prefix: _Path = ".", excludes: Optional[List[str]] = None) -> None: """ Adds the files of the directory tree starting at the root directory to the file set of the directory state. The added file path is relative to the root directory. The prefix is prepended to the file path for each file before it is added to the directory state. The files are moved. """ self._add_tree(root_dir, prefix, self._move_file, excludes) def add_tarfile_members(self, archive: _Path, prefix: _Path, extract: bool) -> None: """ Appends the members of the archive to the file list of the directory state. For each member the prefix path and the member path are joined and then added to the file list of the directory state. If extract is true, then the members of the archive are extracted to the prefix path. """ extract_info = "and extract " if extract else "" logging.info("%s: add %smembers of '%s' using prefix '%s'", self.uid, extract_info, archive, prefix) with tarfile.open(archive, "r") as tar_file: base = self.directory file_list = [ os.path.relpath(os.path.join(prefix, info.name), base) for info in tar_file.getmembers() if not info.isdir() ] if extract: tar_file.extractall(prefix) self.add_files(file_list) def lazy_clone(self, other: "DirectoryState") -> str: """ Lazily clones the directory state. """ logging.info("%s: lazy clone from: %s", self.uid, other.uid) # pylint: disable=protected-access current = set(self._files.keys()) new = set(other._files.keys()) base = self.directory other_base = other.directory for file in sorted(current.difference(new)): target = os.path.join(base, file) try: logging.info("%s: remove: %s", self.uid, target) os.remove(target) except FileNotFoundError: logging.warning("%s: file not found: %s", self.uid, target) for file in sorted(new.difference(current)): target = os.path.join(base, file) self._copy_file(os.path.join(other_base, file), target) for file in sorted(current.intersection(new)): target = os.path.join(base, file) if self._files[file] == other._files[file]: logging.info("%s: keep as is: %s", self.uid, target) else: self._copy_file(os.path.join(other_base, file), target) self._files = other._files.copy() return self._add_hashes(base, self._get_hash) def json_dump(self, data: Any) -> None: """ Dumps the data into the file of the directory state. """ file_path = self.file os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, "w", encoding="utf-8") as file: json.dump(data, file, sort_keys=True, indent=2) def json_load(self) -> Any: """ Loads the data from the file of the directory state. """ with open(self.file, "r", encoding="utf-8") as file: return json.load(file) def save(self) -> None: """ Saves the directory state to the item file. """ self.item.save() def has_changed(self, link: Link) -> bool: digest = self.digest return link["hash"] is None or digest != link["hash"] def discard(self) -> None: """ Discards the directory state. """ logging.info("%s: discard", self.uid) self._discarded_files = set(self._files.keys()) self.remove_files() self.invalidate() self.save() def refresh(self) -> None: """ Refreshes the directory state. """ logging.info("%s: refresh", self.uid) self.load() self.commit("Update directory state")