project = BaseVFS(config.root)
overrides = {BaseVFS(override) for override in config.overrides}
for (alias, key) in SUBSTITUTION.items():
- path = config[key] if key else ""
+ path = config[key] if key else "."
if path is not None:
project[alias] = path
for override in overrides:
remove_file(project, file)
for (present, files) in ((False, added_files), (True, kept_files)):
for dst in sorted(files):
- match = tuple(override for override in overrides if dst in override)
- override = match[0] if match else gnulib
- (vfs, src) = vfs_lookup(dst, gnulib, override, patch=PATCH)
+ (vfs, src) = vfs_lookup(dst, gnulib, overrides, patch=PATCH)
action = update_file if vfs_exists(project, dst) else add_file
with vfs_iostream(vfs, src, "rb", "UTF-8") as istream:
dst_data = src_data = istream.read()
for transformation in transformations:
dst_data = transformation(dst, dst_data)
temporary = False
+ local = (vfs and vfs != gnulib)
if src_data != dst_data:
with tempfile.NamedTemporaryFile("w", encoding="UTF-8", delete=False) as ostream:
ostream.write(dst_data)
src = ostream.name
temporary = True
- match = False
- action(bool(match), vfs, src, project, dst, present)
+ local = False
+ action(local, vfs, src, project, dst, present)
if temporary:
os.unlink(src)
def __repr__(self):
+ origin = self.__origin
module = self.__class__.__module__
name = self.__class__.__name__
- return "{}.{}{{{}}}".format(module, name, repr(self.__origin))
+ return f"{module}.{name}{{{origin}}}"
def __enter__(self):
-def lookup(name, primary, secondary, patch):
+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 primary argument is the primary virtual file system to search for the file.
- The secondary argument is the secondary virtual file system to search for the file.
+ 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 primary VFS: open the file inside the primary VFS.
- - file is present inside the secondary VFS: open the file inside the secondary VFS.
+ - 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 primary or the secondary VFS.
+ 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.
"""
- if not isinstance(name, str):
- raise TypeError("name: str expected")
- if not isinstance(primary, BaseVFS):
- raise TypeError("primary: VFS expected")
- if not isinstance(secondary, BaseVFS):
- raise TypeError("secondary: VFS expected")
- if not isinstance(patch, _Executable):
- raise TypeError("patch: executable expected")
-
- if name in secondary:
- return (secondary, name)
- diff = f"{name}.diff"
- if diff not in secondary:
- return (primary, name)
- with _codecs.open(primary[name], "rb") as istream:
- with _tempfile.NamedTemporaryFile(mode="w+b", delete=False) as ostream:
- path = ostream.name
- _shutil.copyfileobj(istream, ostream)
- stdin = _codecs.open(secondary[diff], "rb")
- cmd = (patch, "-s", path)
- pipes = _sp.Popen(cmd, stdin=stdin, stdout=_sp.PIPE, stderr=_sp.PIPE)
- (stdout, stderr) = pipes.communicate()
- stdout = stdout.decode("UTF-8")
- stderr = stderr.decode("UTF-8")
- returncode = pipes.returncode
- if returncode != 0:
- cmd = "patch -s {} < {}".format(path, secondary[diff])
- raise _sp.CalledProcessError(returncode, cmd, stdout, stderr)
- return (None, path)
+ 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):
_os.makedirs(root[name], exist_ok=True)
+
def backup(root, name):
"""Backup the given file."""
root = BaseVFS(".") if root is None else root
_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
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)
ostream.write(data)
+
def exists(root, name):
"""Check whether the given file exists."""
root = BaseVFS(".") if root is None else root
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)
_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)
_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
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
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)
_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