text: """
  Import as:
  
  import helpers.git as git
  """
  
  import collections
  import functools
  import logging
  import os
  import pprint
  import re
  from typing import Dict, List, Match, Optional, Tuple
  
  import helpers.dbg as dbg
  import helpers.io_ as hio
  import helpers.printing as hprint
  import helpers.system_interaction as hsinte
  
  _LOG = logging.getLogger(__name__)
  
  # We refer to "Git" when we talk about the control system (e.g., "in a Git
  # repository") and `git` when we refer to implementation of Git as a program
  # installed in a computer.
  
  # TODO(gp): Check
  #  https://git-scm.com/book/en/v2/Appendix-B%3A-Embedding-Git-in-your-Applications-Dulwich
  
  # TODO(gp): Avoid "stuttering": the module is already called "git", so no need
  #  to make reference to git again.
  
  # TODO(gp): Add mem caching to some functions below. We assume that one doesn't
  #  change dir (which is a horrible idea) and thus we can memoize.
  
  # TODO(gp): Spell super_module and sub_module always in the same way in both
  #  comments and code. For simplicity (e.g., instead of `super_module` in code and
  #  `super-module` in comment) we might want to spell `supermodule` everywhere.
  
  # #############################################################################
  # Git submodule functions
  # #############################################################################
  
  
  @functools.lru_cache()
  def get_client_root(super_module: bool) -> str:
      """
      Return the full path of the root of the Git client.
  
      E.g., `/Users/saggese/src/.../amp`.
  
      :param super_module: if True use the root of the Git super_module,
          if we are in a submodule. Otherwise use the Git sub_module root
      """
      if super_module and is_inside_submodule():
          # https://stackoverflow.com/questions/957928
          # > cd /Users/saggese/src/.../amp
          # > git rev-parse --show-superproject-working-tree
          # /Users/saggese/src/...
          cmd = "git rev-parse --show-superproject-working-tree"
      else:
          # > git rev-parse --show-toplevel
          # /Users/saggese/src/.../amp
          cmd = "git rev-parse --show-toplevel"
      # TODO(gp): Use system_to_one_line().
      _, out = hsinte.system_to_string(cmd)
      out = out.rstrip("\n")
      dbg.dassert_eq(len(out.split("\n")), 1, msg="Invalid out='%s'" % out)
      client_root: str = os.path.realpath(out)
      return client_root
  
  
  @functools.lru_cache()
  def get_branch_name(dir_name: str = ".") -> str:
      """
      Return the name of the Git branch including a certain dir.
  
      E.g., `master` or `AmpTask672_Add_script_to_check_and_merge_PR`
      """
      dbg.dassert_exists(dir_name)
      # > git rev-parse --abbrev-ref HEAD
      # master
      cmd = "cd %s && git rev-parse --abbrev-ref HEAD" % dir_name
      data: Tuple[int, str] = hsinte.system_to_one_line(cmd)
      _, output = data
      return output
  
  
  @functools.lru_cache()
  def is_inside_submodule(git_dir: str = ".") -> bool:
      """
      Return whether a dir is inside a Git submodule or a Git supermodule.
  
      We determine this checking if the current Git repo is included
      inside another Git repo.
      """
      cmd = []
      # - Find the git root of the current directory
      # - Check if the dir one level up is a valid Git repo
      # Go to the dir.
      cmd.append("cd %s" % git_dir)
      # > cd im/
      # > git rev-parse --show-toplevel
      # /Users/saggese/src/.../amp
      cmd.append('cd "$(git rev-parse --show-toplevel)/.."')
      # > git rev-parse --is-inside-work-tree
      # true
      cmd.append("(git rev-parse --is-inside-work-tree | grep -q true)")
      cmd_as_str = " && ".join(cmd)
      rc = hsinte.system(cmd_as_str, abort_on_error=False)
      ret: bool = rc == 0
      return ret
  
  
  def _is_repo(repo_short_name: str) -> bool:
      """
      Return whether we are inside `amp` and `amp` is a submodule.
      """
      repo_full_name = get_repo_full_name_from_dirname(".", include_host_name=False)
      return get_repo_name(repo_full_name, in_mode="full_name") == repo_short_name
  
  
  def is_amp() -> bool:
      """
      Return whether we are inside `amp` and `amp` is a sub-module.
      """
      return _is_repo("amp")
  
  
  def is_lem() -> bool:
      """
      Return whether we are inside `lm` and `lm` is a sub-module.
      """
      return _is_repo("lem")
  
  
  # TODO(gp): submodule -> sub_module
  def is_in_amp_as_submodule() -> bool:
      """
      Return whether we are in the `amp` repo and it's a sub-module, e.g., of
      `lm`.
      """
      return is_amp() and is_inside_submodule(".")
  
  
  # TODO(gp): supermodule -> super_module
  def is_in_amp_as_supermodule() -> bool:
      """
      Return whether we are in the `amp` repo and it's a super-module, i.e.,
      `amp` by itself.
      """
      return is_amp() and not is_inside_submodule(".")
  
  
  # #############################################################################
  
  
  def _get_submodule_hash(dir_name: str) -> str:
      """
      Report the Git hash that a submodule (e.g., amp) is at from the point of
      view of a supermodule.
  
      > git ls-tree master | grep <dir_name>
      """
      dbg.dassert_exists(dir_name)
      cmd = "git ls-tree master | grep %s" % dir_name
      data: Tuple[int, str] = hsinte.system_to_one_line(cmd)
      _, output = data
      # 160000 commit 0011776388b4c0582161eb2749b665fc45b87e7e  amp
      _LOG.debug("output=%s", output)
      data: List[str] = output.split()
      _LOG.debug("data=%s", data)
      git_hash = data[2]
      return git_hash
  
  
  @functools.lru_cache()
  def get_path_from_supermodule() -> Tuple[str, str]:
      """
      Return the path to the Git repo including the Git submodule for a
      submodule, and return empty for a supermodule. See AmpTask1017.
  
      E.g.,
      - for amp included in another repo returns 'amp'
      - for amp without supermodule returns ''
      """
      cmd = "git rev-parse --show-superproject-working-tree"
      # > cd /Users/saggese/src/.../lm/amp
      # > git rev-parse --show-superproject-working-tree
      # /Users/saggese/src/.../lm
      #
      # > cd /Users/saggese/src/.../lm
      # > git rev-parse --show-superproject-working-tree
      # (No result)
      superproject_path: str = hsinte.system_to_one_line(cmd)[1]
      _LOG.debug("superproject_path='%s'", superproject_path)
      #
      cmd = (
          f"git config --file {superproject_path}/.gitmodules --get-regexp path"
          '| grep $(basename "$(pwd)")'
          "| awk '{ print $2 }'"
      )
      # > git config --file /Users/saggese/src/.../.gitmodules --get-regexp path
      # submodule.amp.path amp
      submodule_path: str = hsinte.system_to_one_line(cmd)[1]
      _LOG.debug("submodule_path='%s'", submodule_path)
      return superproject_path, submodule_path
  
  
  @functools.lru_cache()
  def get_submodule_paths() -> List[str]:
      """
      Return the path of the submodules in this repo, e.g., `["amp"]` or `[]`.
      """
      # > git config --file .gitmodules --get-regexp path
      # submodule.amp.path amp
      cmd = "git config --file .gitmodules --get-regexp path | awk '{ print $2 }'"
      _, txt = hsinte.system_to_string(cmd)
      _LOG.debug("txt=%s", txt)
      files: List[str] = hsinte.text_to_list(txt)
      _LOG.debug("files=%s", files)
      return files
  
  
  def has_submodules() -> bool:
      return len(get_submodule_paths()) > 0
  
  
  # #############################################################################
  
  
  def _get_hash(git_hash: str, short_hash: bool, num_digits: int = 8) -> str:
      dbg.dassert_lte(1, num_digits)
      if short_hash:
          ret = git_hash[:num_digits]
      else:
          ret = git_hash
      return ret
  
  
  def _group_hashes(head_hash: str, remh_hash: str, subm_hash: str) -> str:
      """
      head_hash: a
      remh_hash: b
      subm_hash: c
      """
      map_ = collections.OrderedDict()
      map_["head_hash"] = head_hash
      map_["remh_hash"] = remh_hash
      if subm_hash:
          map_["subm_hash"] = subm_hash
      #
      inv_map = collections.OrderedDict()
      for k, v in map_.items():
          if v not in inv_map:
              inv_map[v] = [k]
          else:
              inv_map[v].append(k)
      #
      txt = []
      for k, v in inv_map.items():
          # Transform:
          #   ('a2bfc704', ['head_hash', 'remh_hash'])
          # into
          #   'head_hash = remh_hash = a2bfc704'
          txt.append("%s = %s" % (" = ".join(v), k))
      txt = "\n".join(txt)
      return txt
  
  
  def report_submodule_status(dir_names: List[str], short_hash: bool) -> str:
      """
      Return a string representing the status of the repos in `dir_names`.
      """
      txt = []
      for dir_name in dir_names:
          txt.append("dir_name='%s'" % dir_name)
          txt.append("  is_inside_submodule: %s" % is_inside_submodule(dir_name))
          #
          branch_name = get_branch_name(dir_name)
          if branch_name != "master":
              branch_name = "!!! %s !!!" % branch_name
          txt.append("  branch: %s" % branch_name)
          #
          head_hash = get_head_hash(dir_name)
          head_hash = _get_hash(head_hash, short_hash)
          txt.append("  head_hash: %s" % head_hash)
          #
          remh_hash = get_remote_head_hash(dir_name)
          remh_hash = _get_hash(remh_hash, short_hash)
          txt.append("  remh_hash: %s" % remh_hash)
          #
          if dir_name != ".":
              subm_hash = _get_submodule_hash(dir_name)
              subm_hash = _get_hash(subm_hash, short_hash)
              txt.append("  subm_hash: %s" % subm_hash)
      txt_as_str = "\n".join(txt)
      return txt_as_str
  
  
  # #############################################################################
  # GitHub repository name
  # #############################################################################
  
  
  # All functions should take as input `repo_short_name` and have a switch `mode`
  # to distinguish full vs short repo name.
  
  # TODO(gp): Maybe rename full -> long to keep it more symmetric "short vs long".
  
  
  def _parse_github_repo_name(repo_name: str) -> Tuple[str, str]:
      """
      Parse a repo name from `git remote`.
  
      The supported formats are both SSH and HTTPS, e.g.,
      - `git@github.com:alphamatic/amp`
      - `https://github.com/alphamatic/amp`
  
      For both of these strings the function returns ("github.com", "alphamatic/amp").
      """
      # Try to parse the SSH format, e.g., `git@github.com:alphamatic/amp`
      m = re.match(r"^git@(\S+.com):(\S+)$", repo_name)
      if not m:
          # Try tp parse the HTTPS format, e.g., `https://github.com/alphamatic/amp`
          m = re.match(r"^https://(\S+.com)/(\S+)$", repo_name)
      dbg.dassert(m, "Can't parse '%s'", repo_name)
      m: Match[str]
      host_name = m.group(1)
      repo_name = m.group(2)
      _LOG.debug("host_name=%s repo_name=%s", host_name, repo_name)
      # We expect something like "alphamatic/amp".
      m = re.match(r"^\S+/\S+$", repo_name)
      dbg.dassert(m, "repo_name='%s'", repo_name)
      # origin  git@github.com:.../ORG_....git (fetch)
      suffix_to_remove = ".git"
      if repo_name.endswith(suffix_to_remove):
          repo_name = repo_name[: -len(suffix_to_remove)]
      return host_name, repo_name
  
  
  def get_repo_full_name_from_dirname(
      dir_name: str, include_host_name: bool
  ) -> str:
      """
      Return the full name of the repo in `git_dir`, e.g., "alphamatic/amp".
  
      This function relies on `git remote` to gather the required information.
  
      :param include_hostname: prepend also the GitHub hostname, e.g., returning
          "github.com/alphamatic/amp"
      :return: the full name of the repo in `git_dir`, e.g., "alphamatic/amp".
      """
      dbg.dassert_exists(dir_name)
      #
      cmd = "cd %s; (git remote -v | grep origin | grep fetch)" % dir_name
      _, output = hsinte.system_to_string(cmd)
      # > git remote -v
      # origin  git@github.com:alphamatic/amp (fetch)
      # origin  git@github.com:alphamatic/amp (push)
      # TODO(gp): Make it more robust, by checking both fetch and push.
      #  "origin  git@github.com:alphamatic/amp (fetch)"
      data: List[str] = output.split()
      _LOG.debug("data=%s", data)
      dbg.dassert_eq(len(data), 3, "data='%s'", str(data))
      # Extract the middle string, e.g., "git@github.com:alphamatic/amp"
      repo_name = data[1]
      # Parse the string.
      host_name, repo_name = _parse_github_repo_name(repo_name)
      if include_host_name:
          res = f"{host_name}/{repo_name}"
      else:
          res = repo_name
      return res
  
  
  def get_repo_full_name_from_client(super_module: bool) -> str:
      """
      Return the full name of the repo (e.g., "alphamatic/amp") from a Git
      client.
  
      :param super_module: like in get_client_root()
      """
      # Get the Git remote in the dir containing the Git repo.
      git_dir = get_client_root(super_module)
      repo_name = get_repo_full_name_from_dirname(git_dir, include_host_name=False)
      return repo_name
  
  
  # /////////////////////////////////////////////////////////////////////////
  
  
  def _get_repo_config_code() -> str:
      """
      Return the text of the code stored in `repo_config.py`.
      """
      # TODO(gp): We should actually ask Git where the super-module is.
      file_name = "./repo_config.py"
      dbg.dassert_file_exists(file_name)
      code: str = hio.from_file(file_name)
      return code
  
  
  def _decorate_with_host_name(
      dict_: Dict[str, str], host_name: str
  ) -> Dict[str, str]:
      """
      Prepend the host name to all the values of the passed dictionary.
      """
      res = {k: f"{host_name}/{v}" for k, v in dict_.items()}
      return res
  
  
  @functools.lru_cache()
  def _get_repo_short_to_full_name(include_host_name: bool) -> Dict[str, str]:
      """
      Return the map from short name (e.g., "amp") to full name (e.g.,
      "alphamatic/amp") using the information in `repo_config.py`
      """
      # From short name to long name.
      repo_map = {
          "amp": "alphamatic/amp",
          "dev_tools": "alphamatic/dev_tools",
      }
      if include_host_name:
          host_name = "github.com"
          repo_map = _decorate_with_host_name(repo_map, host_name)
      _LOG.debug(
          "include_host_name=%s, repo_map=\n%s",
          include_host_name,
          pprint.pformat(repo_map),
      )
      # Read the info from the current repo.
      code = _get_repo_config_code()
      # TODO(gp): make the linter happy creating this symbol that comes from the
      # `exec()`.
      exec(code, globals())  # pylint: disable=exec-used
      current_repo_map = get_repo_map()  # noqa: F821  # type: ignore
      if include_host_name:
          host_name = get_host_name()  # noqa: F821  # type: ignore
          current_repo_map = _decorate_with_host_name(current_repo_map, host_name)
      _LOG.debug(
          "include_host_name=%s, current_repo_map=\n%s",
          include_host_name,
          pprint.pformat(current_repo_map),
      )
      # Update the map.
      dbg.dassert_not_intersection(repo_map.keys(), current_repo_map.keys())
      repo_map.update(get_repo_map())  # noqa: F821  # type: ignore
      dbg.dassert_no_duplicates(repo_map.values())
      _LOG.debug(
          "include_host_name=%s, repo_map=\n%s",
          include_host_name,
          pprint.pformat(repo_map),
      )
      return repo_map
  
  
  # /////////////////////////////////////////////////////////////////////////
  
  
  def get_complete_repo_map(
      in_mode: str, include_host_name: bool = False
  ) -> Dict[str, str]:
      """
      Return the full / short name of a Git repo based on the alternative name.
  
      :param in_mode: the values `full_name` or `short_name` determine how to interpret
          `name`
      """
      repo_map = _get_repo_short_to_full_name(include_host_name)
      if in_mode == "full_name":
          # Compute the reverse map.
          repo_map = {v: k for (k, v) in repo_map.items()}
      elif in_mode == "short_name":
          pass
      else:
          raise ValueError("Invalid in_mode='%s'" % in_mode)
      _LOG.debug(
          "For in_mode=%s, include_host_name=%s, repo_map=\n%s",
          in_mode,
          include_host_name,
          pprint.pformat(repo_map),
      )
      return repo_map
  
  
  def get_repo_name(
      name: str, in_mode: str, include_host_name: bool = False
  ) -> str:
      """
      Return the full/short name of a Git repo based on the other name.
  
      :param in_mode: the values `full_name` or `short_name` determine how to interpret
          `name`
      """
      repo_map = get_complete_repo_map(in_mode, include_host_name)
      dbg.dassert_in(
          name, repo_map, "Invalid name='%s' for in_mode='%s'", name, in_mode
      )
      ret = repo_map[name]
      return ret
  
  
  def get_all_repo_names(
      in_mode: str, include_host_name: bool = False
  ) -> List[str]:
      """
      Return the names (full or short depending on `mode`) of all the Git repos.
  
      :param in_mode: if "full_name" return the full names (e.g., "alphamatic/amp")
          if "short_name" return the short names (e.g., "amp")
      """
      repo_map = get_complete_repo_map(in_mode, include_host_name)
      return sorted(list(repo_map.keys()))
  
  
  def get_task_prefix_from_repo_short_name(short_name: str) -> str:
      """
      Return the task prefix for a repo (e.g., "amp" -> "AmpTask").
      """
      if short_name == "amp":
          prefix = "AmpTask"
      elif short_name == "dev_tools":
          prefix = "DevToolsTask"
      else:
          # We assume that we can build the prefix from the name (e.g., "lm" ->
          # "LmTask").
          # TODO(gp): A more general approach is to save this information inside
          #  `repo_config.py`.
          prefix = short_name.capitalize() + "Task"
      return prefix
  
  
  # #############################################################################
  # Git path
  # #############################################################################
  
  
  @functools.lru_cache()
  def find_file_in_git_tree(file_name: str, super_module: bool = True) -> str:
      """
      Find the path of a file in a Git tree.
  
      We get the Git root and then search for the file from there.
      """
      root_dir = get_client_root(super_module=super_module)
      # TODO(gp): Use -not -path '*/\.git/*'
      cmd = "find %s -name '%s' | grep -v .git" % (root_dir, file_name)
      _, file_name = hsinte.system_to_one_line(cmd)
      _LOG.debug("file_name=%s", file_name)
      dbg.dassert_ne(
          file_name, "", "Can't find file '%s' in dir '%s'", file_name, root_dir
      )
      file_name: str = os.path.abspath(file_name)
      dbg.dassert_exists(file_name)
      return file_name
  
  
  def get_path_from_git_root(file_name: str, super_module: bool) -> str:
      """
      Get the git path from the root of the tree.
  
      :param super_module: like get_client_root()
      """
      # Get Git root.
      git_root = get_client_root(super_module) + "/"
      # TODO(gp): Use os.path.relpath()
      # abs_path = os.path.abspath(file_name)
      # dbg.dassert(abs_path.startswith(git_root))
      # end_idx = len(git_root)
      # ret = abs_path[end_idx:]
      ret = os.path.relpath(file_name, git_root)
      return ret
  
  
  @functools.lru_cache()
  def get_amp_abs_path() -> str:
      """
      Return the absolute path of `amp` dir.
      """
      repo_sym_name = get_repo_full_name_from_client(super_module=False)
      if repo_sym_name == "alphamatic/amp":
          # If we are in the amp repo, then the git client root is the amp
          # directory.
          git_root = get_client_root(super_module=False)
          amp_dir = git_root
      else:
          # If we are not in the amp repo, then look for the amp dir.
          amp_dir = find_file_in_git_tree("amp", super_module=True)
          git_root = get_client_root(super_module=True)
          amp_dir = os.path.join(git_root, amp_dir)
      amp_dir = os.path.abspath(amp_dir)
      # Sanity check.
      dbg.dassert_dir_exists(amp_dir)
      return amp_dir
  
  
  # TODO(gp): Is this needed?
  def get_repo_dirs() -> List[str]:
      """
      Return the list of the repo repositories, e.g., `[".", "amp", "infra"]`.
      """
      dir_names = ["."]
      dirs = ["amp"]
      for dir_name in dirs:
          if os.path.exists(dir_name):
              dir_names.append(dir_name)
      return dir_names
  
  
  def purify_docker_file_from_git_client(
      file_name: str,
      super_module: Optional[bool],
      dir_depth: int = 1,
      mode: str = "return_all_results",
  ) -> Tuple[bool, str]:
      """
      Convert a file or dir that was generated inside Docker to a file in the
      current Git client.
  
      This operation is best effort since it might not be able to find the
      corresponding file in the current repo.
  
      E.g.,
      - A file like '/app/amp/core/dataflow_model/utils.py', in a Docker container with
        Git root in '/app' becomes 'amp/core/dataflow_model/utils.py'
      - For a file like '/app/amp/core/dataflow_model/utils.py' outside Docker, we look
        for the file 'dataflow_model/utils.py' in the current client and then normalize
        with respect to the
  
      :param super_module:
          - True/False: the file is with respect to a Git repo
          - `None`: the file is returned as relative to current dir
      :param dir_depth: same meaning as in `find_file_with_dir()`
      :param mode: same as `system_interaction.select_result_file_from_list()`
      :return: the best guess for the file name corresponding to `file_name`
      """
      _LOG.debug("# Processing file_name='%s'", file_name)
      dbg.dassert_isinstance(file_name, str)
      # Clean up file name.
      file_name = os.path.normpath(file_name)
      _LOG.debug("file_name=%s", file_name)
      file_names = hsinte.find_file_with_dir(
          file_name, ".", dir_depth=dir_depth, mode=mode
      )
      _LOG.debug("file_names=%s", file_names)
      if len(file_names) == 0:
          # We didn't find the file in the current client: leave the file as it was.
          _LOG.warning("Can't find file corresponding to '%s'", file_name)
          found = False
      elif len(file_names) == 1:
          # We have found the file.
          file_name = file_names[0]
          _LOG.debug("file_name=%s", file_name)
          #
          if super_module is not None:
              file_name = get_path_from_git_root(file_name, super_module)
          file_name = os.path.normpath(file_name)
          found = True
      else:
          _LOG.warning(
              "Found multiple potential files corresponding to '%s'", file_name
          )
          file_name = ",".join(file_names)
          found = False
      _LOG.debug("-> found=%s file_name='%s'", found, file_name)
      return (found, file_name)
  
  
  # #############################################################################
  # Git hash
  # #############################################################################
  
  
  def get_head_hash(dir_name: str = ".", short_hash: bool = False) -> str:
      """
      Report the hash that a Git repo is synced at.
  
      ```
      > git rev-parse HEAD
      4759b3685f903e6c669096e960b248ec31c63b69
      ```
      """
      dbg.dassert_exists(dir_name)
      opts = "--short " if short_hash else " "
      cmd = f"cd {dir_name} && git rev-parse {opts}HEAD"
      data: Tuple[int, str] = hsinte.system_to_one_line(cmd)
      _, output = data
      return output
  
  
  # TODO(gp): Use get_head_hash() and remove this.
  def get_current_commit_hash(dir_name: str = ".") -> str:
      dbg.dassert_exists(dir_name)
      cmd = f"cd {dir_name} && git rev-parse HEAD"
      data: Tuple[int, str] = hsinte.system_to_one_line(cmd)
      _, sha = data
      # 0011776388b4c0582161eb2749b665fc45b87e7e
      _LOG.debug("sha=%s", sha)
      return sha
  
  
  def get_remote_head_hash(dir_name: str) -> str:
      """
      Report the hash that the remote Git repo is at.
      """
      dbg.dassert_exists(dir_name)
      sym_name = get_repo_full_name_from_dirname(dir_name, include_host_name=False)
      cmd = f"git ls-remote git@github.com:{sym_name} HEAD 2>/dev/null"
      data: Tuple[int, str] = hsinte.system_to_one_line(cmd)
      _, output = data
      # > git ls-remote git@github.com:alphamatic/amp HEAD 2>/dev/null
      # 921676624f6a5f3f36ab507baed1b886227ac2e6        HEAD
      return output
  
  
  # #############################################################################
  # Modified files
  # #############################################################################
  
  
  def get_modified_files(
      dir_name: str = ".", remove_files_non_present: bool = True
  ) -> List[str]:
      """
      Return the files that are added and modified in the Git client.
  
      In other words the files that will be committed with a `git commit -am ...`.
      Equivalent to `dev_scripts/git_files.sh`
  
      :param dir_name: directory with Git client
      :param remove_files_non_present: remove the files that are not
          currently present in the client
      :return: list of files
      """
      # If the client status is:
      #   > git status -s
      #   AM dev_scripts/infra/ssh_tunnels.py
      #   M helpers/git.py
      #   ?? linter_warnings.txt
      #
      # The result is:
      #   > git diff --cached --name-only
      #   dev_scripts/infra/ssh_tunnels.py
      #
      #   > git ls-files -m
      #   dev_scripts/infra/ssh_tunnels.py
      #   helpers/git.py
      cmd = "(git diff --cached --name-only; git ls-files -m) | sort | uniq"
      files: List[str] = hsinte.system_to_files(
          cmd, dir_name, remove_files_non_present
      )
      return files
  
  
  # TODO(gp): -> ...previously...
  def get_previous_committed_files(
      dir_name: str = ".",
      num_commits: int = 1,
      remove_files_non_present: bool = True,
  ) -> List[str]:
      """
      Return files changed in the Git client in the last `num_commits` commits.
  
      Equivalent to `dev_scripts/git_previous_commit_files.sh`
  
      :param dir_name: directory with Git client
      :param num_commits: how many commits in the past to consider
      :param remove_files_non_present: remove the files that are not currently present
          in the client
      :return: list of files
      """
      cmd = []
      cmd.append('git show --pretty="" --name-only')
      cmd.append("$(git log --author $(git config user.name) -%d" % num_commits)
      cmd.append(r"""| \grep "^commit " | perl -pe 's/commit (.*)/$1/')""")
      cmd_as_str = " ".join(cmd)
      files: List[str] = hsinte.system_to_files(
          cmd_as_str, dir_name, remove_files_non_present
      )
      return files
  
  
  def get_modified_files_in_branch(
      dst_branch: str, dir_name: str = ".", remove_files_non_present: bool = True
  ) -> List[str]:
      """
      Return files modified in the current branch with respect to `dst_branch`.
  
      Equivalent to `git diff --name-only master...`
      Please remember that there is a difference between `master` and `origin/master`.
      See https://stackoverflow.com/questions/18137175
  
      :param dir_name: directory with Git client
      :param dst_branch: branch to compare to, e.g., master
      :param remove_files_non_present: remove the files that are not
          currently present in the client
      :return: list of files
      """
      cmd = "git diff --name-only %s..." % dst_branch
      files: List[str] = hsinte.system_to_files(
          cmd, dir_name, remove_files_non_present
      )
      return files
  
  
  def get_summary_files_in_branch(
      dst_branch: str,
      dir_name: str = ".",
  ) -> str:
      """
      Report summary of files in the current branch with respect to `dst_branch'.
  
      Same interface as `get_modified_files_in_branch`.
      """
      # File types (from https://git-scm.com/docs/git-diff).
      file_types = [
          ("added", "A"),
          ("copied", "C"),
          ("deleted", "D"),
          ("modified", "M"),
          ("renamed", "R"),
          ("type changed", "T"),
          ("unmerged", "U"),
          ("unknown", "X"),
          ("broken pairing", "B"),
      ]
      res = ""
      for tag, diff_type in file_types:
          cmd = f"git diff --diff-filter={diff_type} --name-only {dst_branch}..."
          files = hsinte.system_to_files(
              cmd, dir_name, remove_files_non_present=False
          )
          _LOG.debug("files=%s", "\n".join(files))
          if files:
              res += f"# {tag}: {len(files)}\n"
              res += hprint.indent("\n".join(files)) + "\n"
      res = res.rstrip("\n")
      return res
  
  
  # #############################################################################
  # Git commands.
  # #############################################################################
  
  
  # TODO(gp): -> get_user_name()
  @functools.lru_cache()
  def get_git_name() -> str:
      """
      Return the git user name.
      """
      cmd = "git config --get user.name"
      # For some reason data is annotated as Any by mypy, instead of
      # Tuple[int, str] so we need to cast it to the right value.
      data: Tuple[int, str] = hsinte.system_to_one_line(cmd)
      _, output = data
      return output
  
  
  def git_log(num_commits: int = 5, my_commits: bool = False) -> str:
      """
      Return the output of a pimped version of git log.
  
      :param num_commits: number of commits to report
      :param my_commits: True to report only the current user commits
      :return: string
      """
      cmd = []
      cmd.append("git log --date=local --oneline --graph --date-order --decorate")
      cmd.append(
          "--pretty=format:" "'%h %<(8)%aN%  %<(65)%s (%>(14)%ar) %ad %<(10)%d'"
      )
      cmd.append("-%d" % num_commits)
      if my_commits:
          # This doesn't work in a container if the user relies on `~/.gitconfig` to
          # set the user name.
          # TODO(gp): We should use `get_git_name()`.
          cmd.append("--author $(git config user.name)")
      cmd = " ".join(cmd)
      data: Tuple[int, str] = hsinte.system_to_string(cmd)
      _, txt = data
      return txt
  
  
  def git_stash_push(
      prefix: str, msg: Optional[str] = None, log_level: int = logging.DEBUG
  ) -> Tuple[str, bool]:
      import helpers.datetime_ as hdatet
  
      user_name = hsinte.get_user_name()
      server_name = hsinte.get_server_name()
      timestamp = hdatet.get_timestamp()
      tag = "%s-%s-%s" % (user_name, server_name, timestamp)
      tag = prefix + "." + tag
      _LOG.debug("tag='%s'", tag)
      cmd = "git stash push"
      _LOG.debug("msg='%s'", msg)
      push_msg = tag[:]
      if msg:
          push_msg += ": " + msg
      cmd += " -m '%s'" % push_msg
      hsinte.system(cmd, suppress_output=False, log_level=log_level)
      # Check if we actually stashed anything.
      cmd = r"git stash list | \grep '%s' | wc -l" % tag
      _, output = hsinte.system_to_string(cmd)
      was_stashed = int(output) > 0
      if not was_stashed:
          msg = "Nothing was stashed"
          _LOG.warning(msg)
          # raise RuntimeError(msg)
      return tag, was_stashed
  
  
  def git_stash_apply(mode: str, log_level: int = logging.DEBUG) -> None:
      _LOG.debug("# Checking stash head ...")
      cmd = "git stash list | head -3"
      hsinte.system(cmd, suppress_output=False, log_level=log_level)
      #
      _LOG.debug("# Restoring local changes...")
      if mode == "pop":
          cmd = "git stash pop --quiet"
      elif mode == "apply":
          cmd = "git stash apply --quiet"
      else:
          raise ValueError("mode='%s'" % mode)
      hsinte.system(cmd, suppress_output=False, log_level=log_level)
  
  
  def git_add_update(
      file_list: Optional[List[str]] = None, log_level: int = logging.DEBUG
  ) -> None:
      """
      Add list of files to git.
  
      If `file_list` is not specified, it adds all modified files.
  
      :param file_list: list of files to `git add`
      """
      _LOG.debug("# Adding all changed files to staging ...")
      cmd = "git add %s" % (" ".join(file_list) if file_list is not None else "-u")
      hsinte.system(cmd, suppress_output=False, log_level=log_level)
  
  
  def fetch_origin_master_if_needed() -> None:
      """
      If inside CI system, force fetching `master` branch from Git repo.
  
      When testing a branch, `master` is not always fetched, but it might
      be needed by tests.
      """
      if hsinte.is_inside_ci():
          _LOG.warning("Running inside CI so fetching master")
          cmd = "git branch -a"
          _, txt = hsinte.system_to_string(cmd)
          _LOG.debug("%s=%s", cmd, txt)
          cmd = r'git branch -a | egrep "\s+master\s*$" | wc -l'
          # * (HEAD detached at pull/1337/merge)
          # master
          # remotes/origin/master
          # remotes/pull/1337/merge
          _, num = hsinte.system_to_one_line(cmd)
          num = int(num)
          _LOG.debug("num=%s", num)
          if num == 0:
              # See AmpTask1321 and AmpTask1338 for details.
              cmd = "git fetch origin master:refs/remotes/origin/master"
              hsinte.system(cmd)
              cmd = "git branch --track master origin/master"
              hsinte.system(cmd)
  