author Ted Mielczarek <>
Thu, 22 Jun 2017 12:54:25 -0400
changeset 411997 83004a183f631214b2c0375848fae11b5e690252
parent 394118 b6057c990bd73e39eb5e7c2c1774dbd9264f64d2
child 418505 b5701f3ce50613dc6c88f76e177631676ddbe2a9
permissions -rw-r--r--
Bug 1367329 - Fix error handling code in mozversioncontrol. r=gps, a=test-only MozReview-Commit-ID: Bu5FA9PhxJY

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this,
# file, You can obtain one at

from __future__ import absolute_import, print_function, unicode_literals

import os
import re
import subprocess
import which

from distutils.version import LooseVersion

def get_tool_path(tool):
    """Obtain the path of `tool`."""
    # We use subprocess in places, which expects a Win32 executable or
    # batch script. On some versions of MozillaBuild, we have "hg.exe",
    # "hg.bat," and "hg" (a Python script). "which" will happily return the
    # Python script, which will cause subprocess to choke. Explicitly favor
    # the Windows version over the plain script.
        return which.which(tool + '.exe')
    except which.WhichError:
            return which.which(tool)
        except which.WhichError as e:

    raise Exception('Unable to obtain %s path. Try running '
                    '|mach bootstrap| to ensure your environment is up to '
                    'date.' % tool)

class Repository(object):
    '''A class wrapping utility methods around version control repositories.'''
    def __init__(self, path, tool):
        self.path = os.path.abspath(path)
        self._tool = get_tool_path(tool)
        self._env = os.environ.copy()
        self._version = None

    def _run(self, *args):
        return subprocess.check_output((self._tool, ) + args,

    def tool_version(self):
        '''Return the version of the VCS tool in use as a `LooseVersion`.'''
        if self._version:
            return self._version
        info = self._run('--version').strip()
        match ='version ([^\+\)]+)', info)
        if not match:
            raise Exception('Unable to identify tool version.')

        self.version = LooseVersion(
        return self.version

    def get_modified_files(self):
        '''Return a list of files that are modified in this repository's
        working copy.'''
        raise NotImplementedError

    def get_added_files(self):
        '''Return a list of files that are added in this repository's
        working copy.'''
        raise NotImplementedError

    def add_remove_files(self, path):
        '''Add and remove files under `path` in this repository's working copy.
        raise NotImplementedError

    def forget_add_remove_files(self, path):
        '''Undo the effects of a previous add_remove_files call for `path`.
        raise NotImplementedError

    def get_files_in_working_directory(self):
        """Obtain a list of managed files in the working directory."""
        raise NotImplementedError

class HgRepository(Repository):
    '''An implementation of `Repository` for Mercurial repositories.'''
    def __init__(self, path):
        super(HgRepository, self).__init__(path, 'hg')
        self._env[b'HGPLAIN'] = b'1'

    def get_modified_files(self):
        # Use --no-status to print just the filename.
        return self._run('status', '--modified', '--no-status').splitlines()

    def get_added_files(self):
        # Use --no-status to print just the filename.
        return self._run('status', '--added', '--no-status').splitlines()

    def add_remove_files(self, path):
        args = ['addremove', path]
        if self.tool_version >= b'3.9':
            args = ['--config', 'extensions.automv='] + args

    def forget_add_remove_files(self, path):
        self._run('forget', path)

    def get_files_in_working_directory(self):
        # Can return backslashes on Windows. Normalize to forward slashes.
        return list(p.replace('\\', '/') for p in
                    self._run('files', '-0').split('\0'))

class GitRepository(Repository):
    '''An implementation of `Repository` for Git repositories.'''
    def __init__(self, path):
        super(GitRepository, self).__init__(path, 'git')

    def get_modified_files(self):
        return self._run('diff', '--diff-filter=M', '--name-only').splitlines()

    def get_added_files(self):
        return self._run('diff', '--diff-filter=A', '--name-only').splitlines()

    def add_remove_files(self, path):
        self._run('add', path)

    def forget_add_remove_files(self, path):
        self._run('reset', path)

    def get_files_in_working_directory(self):
        return self._run('ls-files', '-z').split('\0')

class InvalidRepoPath(Exception):
    """Represents a failure to find a VCS repo at a specified path."""

def get_repository_object(path):
    '''Get a repository object for the repository at `path`.
    If `path` is not a known VCS repository, raise an exception.
    if os.path.isdir(os.path.join(path, '.hg')):
        return HgRepository(path)
    elif os.path.exists(os.path.join(path, '.git')):
        return GitRepository(path)
        raise InvalidRepoPath('Unknown VCS, or not a source checkout: %s' %

def get_repository_from_env():
    """Obtain a repository object by looking at the environment."""
    def ancestors(path):
        while path:
            yield path
            path, child = os.path.split(path)
            if child == '':

    for path in ancestors(os.getcwd()):
            return get_repository_object(path)
        except InvalidRepoPath:

    raise Exception('Could not find Mercurial or Git checkout for %s' %