112 lines
2.8 KiB
Python
112 lines
2.8 KiB
Python
"""
|
|
The implementation.
|
|
"""
|
|
|
|
import logging
|
|
from shutil import rmtree
|
|
from subprocess import run
|
|
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
|
|
class Vfs(object):
|
|
"""An abstract filesystem device which can accumulate changes, and apply them in a batch."""
|
|
|
|
def __init__(self, log=None):
|
|
self._log = log or []
|
|
|
|
def _execute_exec(self, e):
|
|
_, dir, cmd = e
|
|
run(cmd, cwd=str(dir))
|
|
|
|
def _execute_link(self, e):
|
|
_, src, dest = e
|
|
if dest.is_file() or dest.is_symlink():
|
|
if dest.is_symlink() and dest.readlink() == src:
|
|
return
|
|
else:
|
|
_log.warn(f"Replacing {dest}")
|
|
dest.unlink()
|
|
elif dest.is_dir():
|
|
_log.warn(f"Replacing {dest}")
|
|
rmtree(dest)
|
|
|
|
assert not dest.exists(), f"{dest} should not exist"
|
|
dest.symlink_to(src)
|
|
|
|
def _execute_chmod(self, e):
|
|
_, dest, mode = e
|
|
dest.chmod(mode)
|
|
|
|
def _execute_mkdir(self, e):
|
|
_, dest = e
|
|
if dest.is_dir():
|
|
return
|
|
elif dest.exists() or dest.is_symlink():
|
|
dest.unlink()
|
|
|
|
dest.mkdir(exist_ok=True)
|
|
|
|
def _execute_unlink(self, e):
|
|
_, dest = e
|
|
# Note that a path which is a dangling symlink will NOT exist but WILL be a symlink
|
|
if not dest.exists() and not dest.is_symlink():
|
|
return
|
|
# Files and dirs just unlink
|
|
if dest.is_symlink() or dest.is_file():
|
|
dest.unlink()
|
|
# Dirs require recursion
|
|
elif dest.is_dir():
|
|
rmtree(dest)
|
|
# Don't succeed silently
|
|
else:
|
|
raise Exception(f"Couldn't unlink {dest}")
|
|
|
|
def _execute_unimplemented(self, e):
|
|
raise NotImplementedError()
|
|
|
|
def _entry_to_command(self, e):
|
|
return e
|
|
|
|
def execute(self, /, callback=None):
|
|
for e in self._log:
|
|
cmd = self._entry_to_command(e)
|
|
_log.debug(f"Executing %r as %r", e, cmd)
|
|
|
|
if callback:
|
|
callback(cmd)
|
|
|
|
# Using self as a dispatch table lol
|
|
getattr(self, f"_execute_{cmd[0]}", self._execute_unimplemented)(cmd)
|
|
|
|
def _command_to_entry(self, cmd):
|
|
return cmd
|
|
|
|
def _append(self, cmd):
|
|
self._log.append(self._command_to_entry(cmd))
|
|
|
|
def link(self, src, dest):
|
|
self._append(("link", src, dest))
|
|
|
|
def copy(self, src, dest):
|
|
self._append(("copy", src, dest))
|
|
|
|
def chmod(self, dest, mode):
|
|
self._append(("chmod", dest, mode))
|
|
|
|
def mkdir(self, dest):
|
|
self._append(("mkdir", dest))
|
|
|
|
def exec(self, dest, cmd):
|
|
self._append(("exec", dest, cmd))
|
|
|
|
def unlink(self, dest):
|
|
self._append(("unlink", dest))
|
|
|
|
def copy(self):
|
|
return Vfs(list(self._log))
|
|
|
|
def merge(self, other: "Vfs"):
|
|
self._log.extend(other._log)
|