ansible/test/lib/ansible_test/_internal/git.py
Sam Doran dc5f369245 [stable-2.9] ansible-test - Continue if the git command returns an error (#61605)
* ansible-test - Contiune if the git command returns an error

* Just return stdout

* Use to_text() when displaying exception

* Add a message property to SubprocessError
(cherry picked from commit e218c98)

Co-authored-by: Sam Doran <sdoran@redhat.com>
2019-08-30 13:23:26 -07:00

133 lines
3.6 KiB
Python

"""Wrapper around git command-line tools."""
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import re
from . import types as t
from .util import (
SubprocessError,
display,
raw_command,
to_text,
)
class Git:
"""Wrapper around git command-line tools."""
def __init__(self, root=None): # type: (t.Optional[str]) -> None
self.git = 'git'
self.root = root
def get_diff(self, args, git_options=None):
"""
:type args: list[str]
:type git_options: list[str] | None
:rtype: list[str]
"""
cmd = ['diff'] + args
if git_options is None:
git_options = ['-c', 'core.quotePath=']
return self.run_git_split(git_options + cmd, '\n', str_errors='replace')
def get_diff_names(self, args):
"""
:type args: list[str]
:rtype: list[str]
"""
cmd = ['diff', '--name-only', '--no-renames', '-z'] + args
return self.run_git_split(cmd, '\0')
def get_submodule_paths(self): # type: () -> t.List[str]
"""Return a list of submodule paths recursively."""
cmd = ['submodule', 'status', '--recursive', '.']
output = self.run_git_split(cmd, '\n')
submodule_paths = [re.search(r'^.[0-9a-f]+ (?P<path>[^ ]+)', line).group('path') for line in output]
return submodule_paths
def get_file_names(self, args):
"""
:type args: list[str]
:rtype: list[str]
"""
cmd = ['ls-files', '-z'] + args
return self.run_git_split(cmd, '\0')
def get_branches(self):
"""
:rtype: list[str]
"""
cmd = ['for-each-ref', 'refs/heads/', '--format', '%(refname:strip=2)']
return self.run_git_split(cmd)
def get_branch(self):
"""
:rtype: str
"""
cmd = ['symbolic-ref', '--short', 'HEAD']
return self.run_git(cmd).strip()
def get_rev_list(self, commits=None, max_count=None):
"""
:type commits: list[str] | None
:type max_count: int | None
:rtype: list[str]
"""
cmd = ['rev-list']
if commits:
cmd += commits
else:
cmd += ['HEAD']
if max_count:
cmd += ['--max-count', '%s' % max_count]
return self.run_git_split(cmd)
def get_branch_fork_point(self, branch):
"""
:type branch: str
:rtype: str
"""
cmd = ['merge-base', '--fork-point', branch]
return self.run_git(cmd).strip()
def is_valid_ref(self, ref):
"""
:type ref: str
:rtype: bool
"""
cmd = ['show', ref]
try:
self.run_git(cmd, str_errors='replace')
return True
except SubprocessError:
return False
def run_git_split(self, cmd, separator=None, str_errors='strict'):
"""
:type cmd: list[str]
:type separator: str | None
:type str_errors: str
:rtype: list[str]
"""
output = self.run_git(cmd, str_errors=str_errors).strip(separator)
if not output:
return []
return output.split(separator)
def run_git(self, cmd, str_errors='strict'):
"""
:type cmd: list[str]
:type str_errors: str
:rtype: str
"""
try:
return raw_command([self.git] + cmd, cwd=self.root, capture=True, str_errors=str_errors)[0]
except SubprocessError as spe:
display.warning(to_text(spe.message))
return spe.stdout