Python sh 模块，git() 实例源码

def create_git_layout(self):
        """ Create module layout for GIT and initialize the repo """
        self.logger.info("Creating git repository")
        module_folder = self.pkg_dest_folder

        git_folder = os.path.join(module_folder, ".git")

        self.write_template( module_folder, { "git/gitignore" : ".gitignore"}, {"options" : self.options})
        git = sh.git

        self.logger.info("Git init")
        git.init(_cwd=module_folder)

        # get an repo object
        repo = git.bake(_cwd=module_folder)
        # add the module folder to git
        self.logger.info("Adding files")
        repo.add(".")
        # commit
        self.logger.info("Initial commit")
        try:
            repo.commit(m='Initial commit')
        except (Exception) as e:
            self.logger.error(e.message)

def test_run_command(capsys):
    cmd = sh.git.bake(version=True)
    util.run_command(cmd)

    result, _ = capsys.readouterr()
    assert '' == result

def test_run_command_with_debug(temp_dir, capsys):
    cmd = sh.git.bake(version=True)
    util.run_command(cmd, debug=True)

    result, _ = capsys.readouterr()
    x = 'COMMAND: {} --version'.format(sh.git)
    assert x in result
    x = 'PWD: {}'.format(temp_dir)
    assert x in result

def prebuild_arch(self, arch):
        super(VlcRecipe, self).prebuild_arch(arch)
        build_dir = self.get_build_dir(arch.arch)
        port_dir = join(build_dir, 'vlc-port-android')
        if self.ENV_LIBVLC_AAR in environ:
            self.aars[arch] = aar = environ.get(self.ENV_LIBVLC_AAR)
            if not exists(aar):
                warning("Error: libvlc-<ver>.aar bundle " \
                           "not found in {}".format(aar))
                info("check {} environment!".format(self.ENV_LIBVLC_AAR))
                exit(1)
        else:
            aar_path = join(port_dir, 'libvlc', 'build', 'outputs', 'aar')
            self.aars[arch] = aar = join(aar_path, 'libvlc-{}.aar'.format(self.version))
            warning("HINT: set path to precompiled libvlc-<ver>.aar bundle " \
                        "in {} environment!".format(self.ENV_LIBVLC_AAR))
            info("libvlc-<ver>.aar should build " \
                        "from sources at {}".format(port_dir))
            if not exists(join(port_dir, 'compile.sh')):
                info("clone vlc port for android sources from {}".format(
                            self.port_git))
                shprint(sh.git, 'clone', self.port_git, port_dir,
                            _tail=20, _critical=True)
            vlc_dir = join(port_dir, 'vlc')
            if not exists(join(vlc_dir, 'Makefile.am')):
                info("clone vlc sources from {}".format(self.vlc_git))
                shprint(sh.git, 'clone', self.vlc_git, vlc_dir,
                            _tail=20, _critical=True)

def __init__(self, path):
        # Paths #
        self.path = path
        self.git_dir = path + '.git'
        # Check exists #
        if not os.path.exists(self.git_dir):
            raise Exception("No git repository at '%s'" % (self.git_dir))
        # Default arguments #
        self.default = ["--git-dir=" + self.git_dir, "--work-tree=" + self.path]

def tag(self):
        """For instance: u'1.0.3-69-gf0c796d-dirty'"""
        tag = sh.git(self.default + ["describe", "--tags", "--dirty", "--always"])
        return tag.strip('\n')

def hash(self):
        """For instance: u'f0c796dae64a5a118d88e60523c011d535e8c476'"""
        sha1 = sh.git(self.default + ["rev-parse", "HEAD"])
        return sha1.strip('\n')

def short_hash(self):
        """For instance: u'f0c796dae64a5a118d88e60523c011d535e8c476'"""
        sha1 = sh.git(self.default + ["rev-parse", "--short", "HEAD"])
        return sha1.strip('\n')

def branch(self):
        """For instance: u'master'"""
        result = sh.git(self.default + ['symbolic-ref', '--short', 'HEAD'])
        return result.strip('\n')

def git(env, repo, branch, dest):
    '''Add a git repository to the source area'''

    echo('Adding git repository ' + style(repo, fg='blue'))

    # Ensure that the destination direcotry doesn't exist
    # Maybe not necessary

    lUrl = urlparse(repo)
    # Strip '.git' at the end
    lRepoName = splitext(basename(lUrl.path))[0] if dest is None else dest
    lRepoLocalPath = join(env.workPath, kSourceDir, lRepoName)

    if exists(lRepoLocalPath):
        raise click.ClickException(
            'Repository already exists \'%s\'' % lRepoLocalPath)

    lArgs = ['clone', repo]

    if dest is not None:
        lArgs += [dest]

    sh.git(*lArgs, _out=sys.stdout, _cwd=env.src)

    if branch is not None:
        echo('Checking out branch/tag ' + style(branch, fg='blue'))
        sh.git('checkout', branch, '-q', _out=sys.stdout, _cwd=lRepoLocalPath)
# ------------------------------------------------------------------------------


# ------------------------------------------------------------------------------

def cli(is_json):
    """
    See the aeriscloud version information
    """
    versions = {
        'aeriscloud': {'version': ac_version},
        'ansible': {'version': ansible_version},
        'vagrant': {'version': vagrant_version()},
        'virtualbox': {'version': virtualbox_version()},
        'ruby': _ruby_version(),
        'python': {'version': _python_version()},
        'git': {'version': str(sh.git('--version'))[12:].strip()}
    }

    # aeriscloud get information
    if os.path.exists(os.path.join(aeriscloud_path, '.git')):
        repo = Repo(aeriscloud_path)
        rev = str(repo.head.commit)[:8]
        branch = str(repo.active_branch)

        versions['aeriscloud']['revision'] = rev
        versions['aeriscloud']['branch'] = branch

    # operating system
    linux_version = _linux_version()
    if linux_version:
        versions['linux'] = linux_version
    else:
        try:
            # this is for osx
            sw_vers = dict([map(unicode.strip, line.split(':'))
                            for line in sh.sw_vers()])
            versions['osx'] = {
                'name': sw_vers['ProductName'],
                'version': sw_vers['ProductVersion'],
                'build': sw_vers['BuildVersion']
            }
        except sh.CommandNotFound:
            pass

    try:
        uname = str(sh.uname('-sr')).strip()
        versions['kernel'] = {'version': uname}
    except sh.CommandNotFound:
        pass

    if is_json:
        click.echo(json.dumps(versions))
    else:
        click.echo(render_cli('version', **versions))

def cli(is_json):
    """
    See the aeriscloud version information
    """
    versions = {
        'aeriscloud': {'version': ac_version},
        'ansible': {'version': ansible_version},
        'vagrant': {'version': vagrant_version()},
        'virtualbox': {'version': virtualbox_version()},
        'ruby': _ruby_version(),
        'python': {'version': _python_version()},
        'git': {'version': str(sh.git('--version'))[12:].strip()}
    }

    # aeriscloud get information
    if os.path.exists(os.path.join(aeriscloud_path, '.git')):
        repo = Repo(aeriscloud_path)
        rev = str(repo.head.commit)[:8]
        branch = str(repo.active_branch)

        versions['aeriscloud']['revision'] = rev
        versions['aeriscloud']['branch'] = branch

    # operating system
    linux_version = _linux_version()
    if linux_version:
        versions['linux'] = linux_version
    else:
        try:
            # this is for osx
            sw_vers = dict([map(unicode.strip, line.split(':'))
                            for line in sh.sw_vers()])
            versions['osx'] = {
                'name': sw_vers['ProductName'],
                'version': sw_vers['ProductVersion'],
                'build': sw_vers['BuildVersion']
            }
        except sh.CommandNotFound:
            pass

    try:
        uname = str(sh.uname('-sr')).strip()
        versions['kernel'] = {'version': uname}
    except sh.CommandNotFound:
        pass

    if is_json:
        click.echo(json.dumps(versions))
    else:
        click.echo(render_cli('version', **versions))

def download_file(self, url, target, cwd=None):
        """
        (internal) Download an ``url`` to a ``target``.
        """
        if not url:
            return
        info('Downloading {} from {}'.format(self.name, url))

        if cwd:
            target = join(cwd, target)

        parsed_url = urlparse(url)
        if parsed_url.scheme in ('http', 'https'):
            def report_hook(index, blksize, size):
                if size <= 0:
                    progression = '{0} bytes'.format(index * blksize)
                else:
                    progression = '{0:.2f}%'.format(
                        index * blksize * 100. / float(size))
                stdout.write('- Download {}\r'.format(progression))
                stdout.flush()

            if exists(target):
                unlink(target)

            urlretrieve(url, target, report_hook)
            return target
        elif parsed_url.scheme in ('git', 'git+ssh', 'git+http', 'git+https'):
            if isdir(target):
                with current_directory(target):
                    shprint(sh.git, 'fetch', '--tags')
                    if self.version:
                        shprint(sh.git, 'checkout', self.version)
                    shprint(sh.git, 'pull')
                    shprint(sh.git, 'pull', '--recurse-submodules')
                    shprint(sh.git, 'submodule', 'update', '--recursive')
            else:
                if url.startswith('git+'):
                    url = url[4:]
                shprint(sh.git, 'clone', '--recursive', url, target)
                if self.version:
                    with current_directory(target):
                        shprint(sh.git, 'checkout', self.version)
                        shprint(sh.git, 'submodule', 'update', '--recursive')
            return target

def srcstat(env):

    if not env.workPath:
        secho  ( 'ERROR: No ipbb work area detected', fg='red' )
        return

    secho ( "Packages", fg='blue' )
    lSrcs = env.getSources()
    if not lSrcs:
        return

    lSrcTable = Texttable(max_width=0)
    lSrcTable.set_deco(Texttable.HEADER | Texttable.BORDER)
    lSrcTable.set_chars(['-', '|', '+', '-'])
    lSrcTable.header(['name', 'kind', 'version'])
    for lSrc in lSrcs:
        lSrcDir = join(env.src, lSrc)

        lKind, lBranch = "unknown", None

        # Check if a git repository
        if exists(join( lSrcDir, '.git')):
            with DirSentry(lSrcDir) as _:
                lKind = 'git'

                try:
                    # lBranch = sh.git('symbolic-ref','--short', 'HEAD').strip()
                    lBranch = sh.git('symbolic-ref', 'HEAD').split('/')[-1].strip()
                except sh.ErrorReturnCode_128:
                    lBranch = sh.git('rev-parse', '--short', 'HEAD').strip()+'...'

                try:
                    sh.git('diff', '--no-ext-diff', '--quiet').strip()
                except sh.ErrorReturnCode_1:
                    lBranch += '*'

                try:
                    sh.git('diff', '--no-ext-diff', '--cached', '--quiet').strip()
                except sh.ErrorReturnCode_1:
                    lBranch += '+'
        elif exists(join( lSrcDir, '.svn')):
            with DirSentry(lSrcDir) as _:
                lKind = 'svn'

                lSVNInfoRaw = sh.svn('info')

                lSVNInfo = { lEntry[0]:lEntry[1].strip() for lEntry in ( lLine.split(':',1) for lLine in lSVNInfoRaw.split('\n') if lLine )}

                lBranch = lSVNInfo['URL'].replace( lSVNInfo['Repository Root']+'/', '' )

                lSVNStatus = sh.svn('status','-q')
                if len(lSVNStatus):
                    lBranch += '*'

        lSrcTable.add_row([lSrc, lKind, lBranch])
    echo  ( lSrcTable.draw() )

# ------------------------------------------------------------------------------