#!/usr/bin/python # encoding: UTF-8 """gnulib virtual filesystem API""" import codecs as _codecs import filecmp as _filecmp import os as _os import shutil as _shutil import sys as _sys import tempfile as _tempfile import subprocess as _sp from .error import UnknownModuleError as _UnknownModuleError from .module import DummyModule as _DummyModule from .module import GnulibModule as _GnulibModule from .misc import Executable as _Executable class BaseVFS: """gnulib generic virtual file system""" __slots__ = ("__origin", "__root", "__table") def __init__(self, origin, **table): self.__table = {} for (key, value) in table.items(): self.__table[key] = _os.path.normpath(value) self.__origin = _os.path.normpath(origin) self.__root = _os.path.abspath(origin) def __repr__(self): origin = self.__origin module = self.__class__.__module__ name = self.__class__.__name__ return f"{module}.{name}{{{origin}}}" def __enter__(self): return self def __exit__(self, exctype, excval, exctrace): pass def __contains__(self, name): if _os.path.isabs(name): raise ValueError("name must be a relative path") path = _os.path.join(self.__root, self[name]) return _os.path.exists(path) def __getitem__(self, name): parts = [] replaced = False name = _os.path.normpath(name) if _os.path.isabs(name): raise ValueError("name cannot be an absolute path") for part in name.split(_os.path.sep): if part == "..": parts += [part] continue if not replaced and part in self.__table: part = self.__table[part] replaced = True parts += [part] return _os.path.normpath(_os.path.sep.join(parts)) def __setitem__(self, src, dst): for name in (src, dst): if _os.path.isabs(name): raise ValueError("name cannot be an absoule path") src = _os.path.normpath(src) dst = _os.path.normpath(dst) self.__table[src] = dst @property def origin(self): """origin VFS path""" return self.__origin @property def root(self): """absolute VFS path""" return self.__root def lookup(name, origin, overrides, patch): """ Try to look up a regular file inside virtual file systems or combine it via patch utility. The name argument is a relative file name which is going to be looked up. The origin argument is the origin virtual file system to search for the file. The overrides argument is the list of override virtual file system to search for the file. The patch argument must be a path to the 'patch' utility binary. - file is present only inside the origin VFS: open the file inside the origin VFS. - file is present inside the override VFS: open the file inside the override VFS. - both file and patch are present: combine in memory. - file is not present: raise an FileNotFoundError exception. The function returns a pair of values, representing the file path and state. The first element, path, represents a regular file system path. The second element, vfs, is either the origin or the override VFS. If the file was obtained via dynamic patching, the vfs element is None. Each of the override VFS is checked for file existence, but only one is used. NOTE: It is up to the caller to unlink files obtained after dynamic patching. """ def _lookup(name, override): if not isinstance(name, str): raise TypeError("name: str expected") if not isinstance(origin, BaseVFS): raise TypeError("origin: VFS expected") if not isinstance(override, BaseVFS): raise TypeError("override: VFS expected") if not isinstance(patch, _Executable): raise TypeError("patch: executable expected") if name in override: return (override, name) diff = f"{name}.diff" if diff not in override: return (origin, name) name = _os.path.join(origin.root, origin[name]) diff = _os.path.join(override.root, override[diff]) with _codecs.open(name, "rb") as istream: with _tempfile.NamedTemporaryFile(mode="w+b", delete=False) as ostream: path = ostream.name _shutil.copyfileobj(istream, ostream) with _codecs.open(diff, "rb") as stdin: with patch("-s", path, stdin=stdin, stdout=_sp.PIPE, stderr=_sp.PIPE) as sp: (stdout, stderr) = sp.communicate() returncode = sp.returncode if returncode != 0: cmd = f"patch -s {path} < {diff}" raise _sp.CalledProcessError(returncode, cmd, stdout, stderr) return (None, path) for override in overrides: (vfs, path) = _lookup(name, override) if vfs is not origin: return (vfs, path) return (origin, name) def mkdir(root, name): """Create a leaf directory and all intermediate ones recursively.""" root = BaseVFS(".") if root is None else root path = name if _os.path.isabs(name) else _os.path.join(root.root, root[name]) _os.makedirs(root[name], exist_ok=True) def backup(root, name): """Backup the given file.""" root = BaseVFS(".") if root is None else root original_path = _os.path.join(root.root, root[name]) backup_path = "{}~".format(original_path) try: _os.unlink(backup_path) except FileNotFoundError: pass # ignore non-existent files _os.rename(original_path, backup_path) def compare(lhs_root, lhs_name, rhs_root, rhs_name): """Compare the given files; return True if files contain the same data.""" lhs_root = BaseVFS(".") if lhs_root is None else lhs_root rhs_root = BaseVFS(".") if rhs_root is None else rhs_root (lhs_path, rhs_path) = (lhs_name, rhs_name) if not _os.path.isabs(lhs_name): lhs_path = _os.path.join(lhs_root.root, lhs_root[lhs_name]) if not _os.path.isabs(rhs_name): rhs_path = _os.path.join(rhs_root.root, rhs_root[rhs_name]) return _filecmp.cmp(lhs_path, rhs_path, shallow=False) def copy(src_root, src_name, dst_root, dst_name): """Copy file data.""" src_abs = _os.path.isabs(src_name) dst_abs = _os.path.isabs(dst_name) if src_abs and dst_abs: raise ValueError("absolute src and dst") limit = (16 * 1024) src_root = BaseVFS(".") if src_root is None else src_root dst_root = BaseVFS(".") if dst_root is None else dst_root mkdir(dst_root, _os.path.dirname(dst_name)) (src_path, dst_path) = (src_name, dst_name) if not _os.path.isabs(src_name): src_path = _os.path.join(src_root.root, src_root[src_name]) if not _os.path.isabs(dst_name): dst_path = _os.path.join(dst_root.root, dst_root[dst_name]) with _codecs.open(src_path, "rb") as istream: with _codecs.open(dst_path, "wb") as ostream: data = True while data: data = istream.read(limit) ostream.write(data) def exists(root, name): """Check whether the given file exists.""" root = BaseVFS(".") if root is None else root path = name if _os.path.isabs(name) else _os.path.join(root.root, root[name]) return _os.path.exists(path) def hardlink(src_root, src_name, dst_root, dst_name): """Create a hard link to the file.""" src_abs = _os.path.isabs(src_name) dst_abs = _os.path.isabs(dst_name) if src_abs and dst_abs: raise ValueError("absolute src and dst") src_root = BaseVFS(".") if src_root is None else src_root dst_root = BaseVFS(".") if dst_root is None else dst_root mkdir(src_root, _os.path.dirname(src_name)) mkdir(dst_root, _os.path.dirname(dst_name)) (src_path, dst_path) = (src_name, dst_name) if not _os.path.isabs(src_name): src_path = _os.path.join(src_root.root, src_root[src_name]) if not _os.path.isabs(dst_name): dst_path = _os.path.join(dst_root.root, dst_root[dst_name]) _os.link(src_path, dst_path) def move(src_root, src_name, dst_root, dst_name): """Move file data.""" src_abs = _os.path.isabs(src_name) dst_abs = _os.path.isabs(dst_name) if src_abs and dst_abs: raise ValueError("absolute src and dst") src_root = BaseVFS(".") if src_root is None else src_root dst_root = BaseVFS(".") if dst_root is None else dst_root mkdir(dst_root, _os.path.dirname(dst_name)) (src_path, dst_path) = (src_name, dst_name) if not _os.path.isabs(src_name): src_path = _os.path.join(src_root.root, src_root[src_name]) if not _os.path.isabs(dst_name): dst_path = _os.path.join(dst_root.root, dst_root[dst_name]) _os.rename(src_path, dst_path) def iostream(root, name, mode="r", encoding=None): """Open file and return a stream. Raise IOError upon failure.""" root = BaseVFS(".") if root is None else root path = name if _os.path.isabs(name) else _os.path.join(root.root, root[name]) return _codecs.open(path, mode, encoding) def readlink(root, name): """Obtain the path to which the symbolic link points.""" root = BaseVFS(".") if root is None else root mkdir(root, _os.path.dirname(name)) path = name if _os.path.isabs(name) else _os.path.join(root.root, root[name]) return _os.readlink(path) def symlink(src_root, src_name, dst_root, dst_name, relative=True): """Create a symbolic link to the file.""" src_abs = _os.path.isabs(src_name) dst_abs = _os.path.isabs(dst_name) if src_abs and dst_abs: raise ValueError("absolute src and dst") src_root = BaseVFS(".") if src_root is None else src_root dst_root = BaseVFS(".") if dst_root is None else dst_root mkdir(dst_root, _os.path.dirname(dst_name)) if not relative: (src_path, dst_path) = (src_name, dst_name) if not _os.path.isabs(src_name): src_path = _os.path.join(src_root.root, src_root[src_name]) if not _os.path.isabs(dst_name): dst_path = _os.path.join(dst_root.root, dst_root[dst_name]) else: src_path = _os.path.join(src_root.origin, src_root[src_name]) dst_path = _os.path.join(dst_root.origin, dst_root[dst_name]) prefix = _os.path.relpath(_os.path.dirname(src_path), _os.path.dirname(dst_path)) suffix = _os.path.basename(src_root[src_name]) src_path = _os.path.join(prefix, suffix) dst_path = _os.path.join(dst_root.root, dst_root[dst_name]) _os.symlink(src_path, dst_path) def unlink(root, name): """Unlink a file, backing it up if necessary.""" root = BaseVFS(".") if root is None else root mkdir(root, _os.path.dirname(name)) path = name if _os.path.isabs(name) else _os.path.join(root.root, root[name]) _os.unlink(path) class GnulibGitVFS(BaseVFS): """gnulib git repository""" __slots__ = ("__cache", "__prefix") _EXCLUDE = { "." : str.startswith, "~" : str.endswith, "-tests" : str.endswith, "ChangeLog" : str.__eq__, "COPYING" : str.__eq__, "README" : str.__eq__, "TEMPLATE" : str.__eq__, "TEMPLATE-TESTS" : str.__eq__, "TEMPLATE-EXTENDED" : str.__eq__, } def __init__(self, origin): super().__init__(origin=origin) self.__cache = {"dummy": _DummyModule()} self.__prefix = _sys.intern(_os.path.join(self.root, "modules")) if not _os.path.exists(self.root): raise FileNotFoundError(self.root) if not _os.path.isdir(self.root): raise NotADirectoryError(self.root) if not _os.path.isdir(_os.path.join(self.root, ".git")): raise TypeError("{} is not a gnulib repository".format(prefix)) def module(self, name): """Try to find the module by name.""" if name in self.__cache: return self.__cache[name] path = _os.path.join(self.__prefix, name) try: result = self.__cache[name] = _GnulibModule(path=path, name=name) return result except _UnknownModuleError: return None def modules(self): """Iterate over all available modules.""" for root, _, files in _os.walk(self.__prefix): names = [] for name in files: exclude = False for (key, method) in GnulibGitVFS._EXCLUDE.items(): if method(name, key): exclude = True break if not exclude: names += [name] for name in names: path = _os.path.join(root, name) name = path[len(self.__prefix) + 1:] yield self.module(name)