npm dotbot plugin (wip)

alternate.py

import os, subprocess, dotbot
from enum import Enum

class PkgStatus(Enum):
    UPDATED    = 'Updated (npm)'
    OUTDATED   = 'Not up to date (npm)'
    UP_TO_DATE = 'Already up to date (npm)'
    INSTALLED  = 'Newly installed (npm)'

    NOT_FOUND  = 'Not found (npm)'
    NOT_SURE   = 'Could not determine (npm)'

class NpmHandler(dotbot.Plugin):
    _directive = 'npm'

    def __init__(self, context):
        super(NpmHandler, self).__init__(self)
        self._context = context

    def can_handle(self, directive):
        return directive == self._directive

    def handle(self, directive, data):
        if directive != self._directive:
            raise ValueError('npm cannot handle directive %s' % directive)
        return self._process(data)

    def _process(self, packages):
        results = {}
        result = PkgStatus.NOT_SURE
        success = True
        successful = [PkgStatus.UP_TO_DATE, PkgStatus.INSTALLED, PkgStatus.UPDATED, PkgStatus.OUTDATED, PkgStatus.UP_TO_DATE]
        defaults = self._context.defaults().get('npm', {})
        npm_auto_update = self._context.defaults().get('npm_auto_update', True)

        for pkg in packages:
            if isinstance(pkg, dict):
                self._log.error('Incorrect format')
            else:
                pass

            if self._package_is_installed(pkg):
                if self._package_is_updated(pkg):
                    self._log.info('Package already up to date: {}.'.format(pkg))
                    result = PkgStatus.UP_TO_DATE
                else:
                    self._log.lowinfo('Package out of date: {}.'.format(pkg))
                    # Only update the outdated the package if this plugin's auto-update flag is set
                    if npm_auto_update:
                        self._log.lowinfo('Updating package: {}...'.format(pkg))
                        result = self._update_package(pkg)
                        if result == PkgStatus.UPDATED:
                            self._log.info('Updated package: {}.'.format(pkg))

                    else:
                        result = PkgStatus.OUTDATED
            else:
                self._log.lowinfo('Installing {}...'.format(pkg))
                result = self._install(pkg)
                if result == PkgStatus.INSTALLED:
                    self._log.info('Installed package: {}.'.format(pkg))

            # Check if the result was one of the listed successful results
            results[result] = results.get(result, 0) + 1
            if result not in successful:
                self._log.error("Could not install npm package '{}'".format(pkg))


        if all([result in successful for result in results.keys()]):
            self._log.info('All npm packages were handled successfully')
            success = True
        else:
            success = False

        # Output status at the end, and list actions performed
        for status, amount in results.items():
            log = self._log.info if status in successful else self._log.error
            log('{} {}'.format(amount ,status.value))

        # Delete the logs npm automatically generates when there is an error
        self._delete_logs()
        return success

    # Installs a package
    def _install(self, pkg):
        cmd = 'npm install -g {}'.format(pkg)
        returnCode = self._execute_subprocess(cmd)['returnCode']
        if rc == 0:
            return PkgStatus.INSTALLED
        else:
            self._log.warn("Could not determine what happened with npm package {}".format(pkg))
            return PkgStatus.NOT_SURE

    # Checks if a package is installed
    def _package_is_installed(self, pkg):
        cmd = 'npm list --depth 1 --global {} > /dev/null 2>&1'.format(pkg)
        returnCode = self._execute_subprocess(cmd)['returnCode']
        return True if returnCode == 0 else False

    # Checks if a package is up to date
    def _package_is_updated(self, pkg):
        cmd = 'npm outdated -g {} | grep -w "{}\s"'.format(pkg, pkg)
        # Any results here should indicate an outdated package
        output = self._execute_subprocess(cmd)['stdout']
        return False if output else True

    # Updates a package, if it isn't already up to date
    def _update_package(self, pkg):
        if _package_is_updated(pkg):
            return PkgStatus.UP_TO_DATE
        cmd = 'npm update -g {}'.format(pkg)
        returnCode = _execute_subprocess(cmd)['returnCode']
        return PkgStatus.UPDATED if returnCode == 0 else PkgStatus.NOT_SURE

    # NPM will generate in this directory when there is an error.
    # This will remove them so they don't
    # accumulate and clutter the directory.
    def _delete_logs(self):
        self._execute_subprocess('rm *.log.*')

    # Executes a synchronous shell subprocess
    # and returns the exit code and stdout output as a dictionary
    def _execute_subprocess(self, cmd):
        process = subprocess.run(cmd, shell=True,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        return {'returncode': process.returncode, 'stdout': process.output }

npmdotbot.py

import os, subprocess, dotbot
from enum import Enum

class PkgStatus(Enum):
    UPDATED    = 'Updated (npm)'
    OUTDATED   = 'Not up to date (npm)'
    UP_TO_DATE = 'Already up to date (npm)'
    INSTALLED  = 'Newly installed (npm)'

    NOT_FOUND  = 'Not found (npm)'
    NOT_SURE   = 'Could not determine (npm)'

class NpmHandler(dotbot.Plugin):
    _directive = 'npm'

    def __init__(self, context):
        super(NpmHandler, self).__init__(self)
        self._context = context

    def can_handle(self, directive):
        return directive == self._directive

    def handle(self, directive, data):
        if directive != self._directive:
            raise ValueError('npm cannot handle directive %s' % directive)
        return self._process(data)

    def _process(self, packages):
        results = {}
        result = PkgStatus.NOT_SURE
        success = True
        successful = [PkgStatus.UP_TO_DATE, PkgStatus.INSTALLED, PkgStatus.UPDATED, PkgStatus.OUTDATED, PkgStatus.UP_TO_DATE]
        defaults = self._context.defaults().get('npm', {})
        npm_auto_update = self._context.defaults().get('npm_auto_update', True)

        for pkg in packages:
            if isinstance(pkg, dict):
                self._log.error('Incorrect format')
            else:
                pass

            if self._package_is_installed(pkg):
                if self._package_is_updated(pkg):
                    self._log.info('Package already up to date: {}.'.format(pkg))
                    result = PkgStatus.UP_TO_DATE
                else:
                    self._log.lowinfo('Package out of date: {}.'.format(pkg))
                    if npm_auto_update:
                        self._log.lowinfo('Updating package: {}...'.format(pkg))
                        result = self._update_package(pkg)
                        if result == PkgStatus.UPDATED:
                            self._log.info('Updated package: {}.'.format(pkg))

                    else:
                        result = PkgStatus.OUTDATED
            else:
                self._log.lowinfo('Installing {}...'.format(pkg))
                result = self._install(pkg)
                if result == PkgStatus.INSTALLED:
                    self._log.info('Installed package: {}.'.format(pkg))

            results[result] = results.get(result, 0) + 1
            if result not in successful:
                self._log.error("Could not install npm package '{}'".format(pkg))


        if all([result in successful for result in results.keys()]):
            self._log.info('All npm packages were handled successfully')
            success = True
        else:
            success = False

        for status, amount in results.items():
            log = self._log.info if status in successful else self._log.error
            log('{} {}'.format(amount ,status.value))

        self._delete_logs()
        return success

    def _install(self, pkg):
        cmd = 'npm install -g {}'.format(pkg)
        process = subprocess.Popen(cmd, shell=True,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        process.wait()
        process.communicate()
        rc = process.returncode
        if rc == 0:
            return PkgStatus.INSTALLED
        else:
            self._log.warn("Could not determine what happened with npm package {}".format(pkg))
            return PkgStatus.NOT_SURE

    def _package_is_installed(self, pkg):
        cmd = 'npm list --depth 1 --global {} > /dev/null 2>&1'.format(pkg)
        process = subprocess.Popen(cmd, shell=True,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        process.wait()
        process.communicate()
        rc = process.returncode

        return True if rc == 0 else False

    def _package_is_updated(self, pkg):
        cmd = 'npm outdated -g {} | grep -w "{}\s"'.format(pkg, pkg)
        process = subprocess.Popen(cmd, shell=True,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

        # Any results here should indicate an outdated package
        out = process.stdout.read()
        process.stdout.close()

        return False if out else True

    def _update_package(self, pkg):
        if _package_is_updated(pkg):
            return PkgStatus.UP_TO_DATE

        cmd = 'npm update -g {}'.format(pkg)
        process = subprocess.Popen(cmd, shell=True,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        process.wait()
        process.communicate()
        rc = process.returncode

        return PkgStatus.UPDATED if rc == 0 else PkgStatus.NOT_SURE

    # NPM will generate in this directory when there is an error.
    # This will remove them so they don't
    # accumulate and clutter the directory.
    def _delete_logs(self):
        cmd = 'rm *.log.*'
        process = subprocess.Popen(cmd, shell=True,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT)