...

Text file src/github.com/datawire/ambassador/v2/releng/lib/gitutil.py

Documentation: github.com/datawire/ambassador/v2/releng/lib

     1from typing import Optional, Union
     2
     3import http.client
     4import json
     5
     6from distutils.util import strtobool
     7
     8from .uiutil import run, check_command
     9from .uiutil import run_txtcapture as run_capture
    10
    11
    12# parse_bool is lifted from python/ambassador/utils.py -- it's just too useful.
    13def parse_bool(s: Optional[Union[str, bool]]) -> bool:
    14    """
    15    Parse a boolean value from a string. T, True, Y, y, 1 return True;
    16    other things return False.
    17    """
    18
    19    # If `s` is already a bool, return its value.
    20    #
    21    # This allows a caller to not know or care whether their value is already
    22    # a boolean, or if it is a string that needs to be parsed below.
    23    if isinstance(s, bool):
    24        return s
    25
    26    # If we didn't get anything at all, return False.
    27    if not s:
    28        return False
    29
    30    # OK, we got _something_, so try strtobool.
    31    try:
    32        return strtobool(s)
    33    except ValueError:
    34        return False
    35
    36
    37def branch_exists(branch_name: str) -> bool:
    38    return check_command(["git", "rev-parse", "--verify", branch_name])
    39
    40
    41def has_open_pr(gh_repo: str, base: str, branchname: str) -> bool:
    42    conn = http.client.HTTPSConnection("api.github.com")
    43    conn.request("GET", f"/repos/{gh_repo}/pulls?base={base}", headers={"User-Agent":"python"})
    44    r1 = conn.getresponse()
    45    body = r1.read()
    46    json_body = json.loads(body)
    47    for pr_info in json_body:
    48        if pr_info.get('head',{}).get('ref') == branchname:
    49            # check that it is open
    50            if pr_info.get('state') == 'open':
    51                return True
    52    return False
    53
    54
    55def git_add(filename: str) -> None:
    56    """
    57    Use `git add` to stage a single file.
    58    """
    59
    60    run(['git', 'add', '--', filename])
    61
    62
    63def git_check_clean(allow_staged: bool = False, allow_untracked: bool = False) -> None:
    64    """
    65    Use `git status --porcelain` to check if the working tree is dirty.
    66    If allow_staged is True, allow staged files, but no unstaged changes.
    67    If allow_untracked is True, allow untracked files.
    68    """
    69
    70    cmdvec = [ 'git', 'status', '--porcelain' ]
    71
    72    if allow_untracked:
    73        cmdvec += [ "--untracked-files=no" ]
    74
    75    out = run_capture(cmdvec)
    76
    77    if out:
    78        # Can we allow staged changes?
    79        if not allow_staged:
    80            # Nope. ANY changes are unacceptable, so we can short-circuit
    81            # here.
    82            raise Exception(out)
    83
    84        # If here, staged changes are OK, and unstaged changes are not.
    85        # In the porcelain output, staged changes start with a change
    86        # character followed by a space, and unstaged changes start with a
    87        # space followed by a change character. So any lines with a non-space
    88        # in the second column are a problem here.
    89
    90        lines = out.split('\n')
    91        problems = [line for line in lines if line[1] != ' ']
    92
    93        if problems:
    94            raise Exception("\n".join(problems))

View as plain text