third_party/pigweed/repo/pw_package/py/pw_package/git_repo.py

   1 # Copyright 2020 The Pigweed Authors
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License"); you may not
   4 # use this file except in compliance with the License. You may obtain a copy of
   5 # the License at
   6 #
   7 #     https://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 # License for the specific language governing permissions and limitations under
  13 # the License.
  14 """Install and check status of Git repository-based packages."""
  15
  16 import os
  17 import pathlib
  18 import shutil
  19 import subprocess
  20 from typing import Union
  21 import urllib.parse
  22
  23 import pw_package.package_manager
  24
  25 PathOrStr = Union[pathlib.Path, str]
  26
  27
  28 def git_stdout(*args: PathOrStr,
  29                show_stderr=False,
  30                repo: PathOrStr = '.') -> str:
  31     return subprocess.run(['git', '-C', repo, *args],
  32                           stdout=subprocess.PIPE,
  33                           stderr=None if show_stderr else subprocess.DEVNULL,
  34                           check=True).stdout.decode().strip()
  35
  36
  37 def git(*args: PathOrStr,
  38         repo: PathOrStr = '.') -> subprocess.CompletedProcess:
  39     return subprocess.run(['git', '-C', repo, *args], check=True)
  40
  41
  42 class GitRepo(pw_package.package_manager.Package):
  43     """Install and check status of Git repository-based packages."""
  44     def __init__(self, url, commit, *args, **kwargs):
  45         super().__init__(*args, **kwargs)
  46         self._url = url
  47         self._commit = commit
  48
  49     def status(self, path: pathlib.Path) -> bool:
  50         if not os.path.isdir(path / '.git'):
  51             return False
  52
  53         remote = git_stdout('remote', 'get-url', 'origin', repo=path)
  54         url = urllib.parse.urlparse(remote)
  55         if url.scheme == 'sso' or '.git.corp.google.com' in url.netloc:
  56             host = url.netloc.replace(
  57                 '.git.corp.google.com',
  58                 '.googlesource.com',
  59             )
  60             if not host.endswith('.googlesource.com'):
  61                 host += '.googlesource.com'
  62             remote = 'https://{}{}'.format(host, url.path)
  63
  64         commit = git_stdout('rev-parse', 'HEAD', repo=path)
  65         status = git_stdout('status', '--porcelain=v1', repo=path)
  66         return remote == self._url and commit == self._commit and not status
  67
  68     def install(self, path: pathlib.Path) -> None:
  69         # If already installed and at correct version exit now.
  70         if self.status(path):
  71             return
  72
  73         # Otherwise delete current version and clone again.
  74         if os.path.isdir(path):
  75             shutil.rmtree(path)
  76
  77         # --filter=blob:none means we don't get history, just the current
  78         # revision. If we later run commands that need history it will be
  79         # retrieved on-demand. For small repositories the effect is negligible
  80         # but for large repositories this should be a significant improvement.
  81         git('clone', '--filter=blob:none', self._url, path)
  82         git('reset', '--hard', self._commit, repo=path)