From b084dd84acb10edec2029783e3c359c2639f852c Mon Sep 17 00:00:00 2001 From: Jesse Leite Date: Sun, 6 Oct 2019 01:49:42 -0400 Subject: [PATCH] Add `create` directive (WIP). --- dotbot/cli.py | 2 +- dotbot/dispatcher.py | 2 +- dotbot/plugins/__init__.py | 1 + dotbot/plugins/create.py | 147 +++++++++++++++++++++++++++++++++++++ 4 files changed, 150 insertions(+), 2 deletions(-) create mode 100644 dotbot/plugins/create.py diff --git a/dotbot/cli.py b/dotbot/cli.py index fdc2a13..2680acf 100644 --- a/dotbot/cli.py +++ b/dotbot/cli.py @@ -56,7 +56,7 @@ def main(): log.use_color(False) plugin_directories = list(options.plugin_dirs) if not options.disable_built_in_plugins: - from .plugins import Clean, Link, Shell + from .plugins import Clean, Create, Link, Shell plugin_paths = [] for directory in plugin_directories: for plugin_path in glob.glob(os.path.join(directory, '*.py')): diff --git a/dotbot/dispatcher.py b/dotbot/dispatcher.py index d1a4f95..1fac1ad 100644 --- a/dotbot/dispatcher.py +++ b/dotbot/dispatcher.py @@ -27,8 +27,8 @@ class Dispatcher(object): # keep going, let other plugins handle this if they want for plugin in self._plugins: if plugin.can_handle(action): + success &= plugin.handle(action, task[action]) try: - success &= plugin.handle(action, task[action]) handled = True except Exception as err: self._log.error( diff --git a/dotbot/plugins/__init__.py b/dotbot/plugins/__init__.py index 93bd981..f75bef5 100644 --- a/dotbot/plugins/__init__.py +++ b/dotbot/plugins/__init__.py @@ -1,3 +1,4 @@ from .clean import Clean +from .create import Create from .link import Link from .shell import Shell diff --git a/dotbot/plugins/create.py b/dotbot/plugins/create.py new file mode 100644 index 0000000..8a86ba5 --- /dev/null +++ b/dotbot/plugins/create.py @@ -0,0 +1,147 @@ +import os +import glob +import shutil +import dotbot +import subprocess + + +class Create(dotbot.Plugin): + ''' + Create empty paths. + ''' + + _directive = 'create' + + def can_handle(self, directive): + return directive == self._directive + + def handle(self, directive, data): + if directive != self._directive: + raise ValueError('Create cannot handle directive %s' % directive) + return self._process_paths(data) + + def _process_paths(self, paths): + success = True + defaults = self._context.defaults().get('create', {}) + for path in paths: + relative = defaults.get('relative', False) + force = defaults.get('force', False) + recreate = defaults.get('recreate', False) + use_glob = defaults.get('glob', False) + test = defaults.get('if', None) + # if isinstance(config, dict): + # # extended config + # test = config.get('if', test) + # relative = config.get('relative', relative) + # force = config.get('force', force) + # recreate = config.get('recreate', force) + # use_glob = config.get('glob', use_glob) + if test is not None and not self._test_success(test): + self._log.lowinfo('Skipping %s' % path) + continue + path = os.path.expandvars(os.path.expanduser(path)) + if use_glob: + self._log.debug("Globbing with path: " + str(path)) + glob_results = glob.glob(path) + if len(glob_results) is 0: + self._log.warning("Globbing couldn't find anything matching " + str(path)) + success = False + continue + glob_star_loc = path.find('*') + if glob_star_loc is -1 and path[-1] is '/': + self._log.error("Ambiguous action requested.") + self._log.error("No wildcard in glob, directory use undefined: " + + path + " -> " + str(glob_results)) + self._log.warning("Did you want to link the directory or into it?") + success = False + continue + elif glob_star_loc is -1 and len(glob_results) is 1: + success &= self._create(path) + else: + self._log.lowinfo("Globs from '" + path + "': " + str(glob_results)) + glob_base = path[:glob_star_loc] + for glob_full_item in glob_results: + glob_item = glob_full_item[len(glob_base):] + glob_link_destination = os.path.join(path, glob_item) + if force or recreate: + success &= self._delete(glob_full_item, glob_link_destination, relative, force) + success &= self._create(glob_link_destination) + else: + if force or recreate: + success &= self._delete(path, destination, relative, force) + success &= self._create(path) + if success: + self._log.info('All paths have been set up') + else: + self._log.error('Some paths were not successfully set up') + return success + + def _test_success(self, command): + with open(os.devnull, 'w') as devnull: + ret = subprocess.call( + command, + shell=True, + stdout=devnull, + stderr=devnull, + executable=os.environ.get('SHELL'), + ) + if ret != 0: + self._log.debug('Test \'%s\' returned false' % command) + return ret == 0 + + def _exists(self, path): + ''' + Returns true if the path exists. + ''' + path = os.path.expanduser(path) + return os.path.exists(path) + + def _create(self, path): + success = True + if not self._exists(path): + self._log.debug("Try to create path: " + str(path)) + try: + self._log.lowinfo('Creating path %s' % path) + os.makedirs(path) + except OSError: + self._log.warning('Failed to create path %s' % path) + success = False + else: + self._log.lowinfo('Path exists %s' % path) + return success + + def _delete(self, source, path, relative, force): + success = True + source = os.path.join(self._context.base_directory(), source) + fullpath = os.path.expanduser(path) + if relative: + source = self._relative_path(source, fullpath) + if ((self._is_link(path) and self._link_destination(path) != source) or + (self._exists(path) and not self._is_link(path))): + removed = False + try: + if os.path.islink(fullpath): + os.unlink(fullpath) + removed = True + elif force: + if os.path.isdir(fullpath): + shutil.rmtree(fullpath) + removed = True + else: + os.remove(fullpath) + removed = True + except OSError: + self._log.warning('Failed to remove %s' % path) + success = False + else: + if removed: + self._log.lowinfo('Removing %s' % path) + return success + + def _relative_path(self, source, destination): + ''' + Returns the relative path to get to the source file from the + destination file. + ''' + destination_dir = os.path.dirname(destination) + return os.path.relpath(source, destination_dir)