files: # Copyright 2012-2015 The Meson development team
  
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  
  #     http://www.apache.org/licenses/LICENSE-2.0
  
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  """A library of random helper functionality."""
  from pathlib import Path
  import sys
  import stat
  import time
  import platform, subprocess, operator, os, shlex, shutil, re
  import collections
  from enum import Enum
  from functools import lru_cache, update_wrapper
  from itertools import tee, filterfalse
  import typing as T
  import uuid
  
  from mesonbuild import mlog
  
  _T = T.TypeVar('_T')
  _U = T.TypeVar('_U')
  
  have_fcntl = False
  have_msvcrt = False
  # {subproject: project_meson_version}
  project_meson_versions = {}
  
  try:
      import fcntl
      have_fcntl = True
  except Exception:
      pass
  
  try:
      import msvcrt
      have_msvcrt = True
  except Exception:
      pass
  
  from glob import glob
  
  if os.path.basename(sys.executable) == 'meson.exe':
      # In Windows and using the MSI installed executable.
      python_command = [sys.executable, 'runpython']
  else:
      python_command = [sys.executable]
  meson_command = None
  
  GIT = shutil.which('git')
  def git(cmd: T.List[str], workingdir: str, **kwargs) -> subprocess.CompletedProcess:
      pc = subprocess.run([GIT, '-C', workingdir] + cmd,
                          # Redirect stdin to DEVNULL otherwise git messes up the
                          # console and ANSI colors stop working on Windows.
                          stdin=subprocess.DEVNULL, **kwargs)
      # Sometimes git calls git recursively, such as `git submodule update
      # --recursive` which will be without the above workaround, so set the
      # console mode again just in case.
      if platform.system().lower() == 'windows':
          mlog._windows_ansi()
      return pc
  
  
  def set_meson_command(mainfile):
      global python_command
      global meson_command
      # On UNIX-like systems `meson` is a Python script
      # On Windows `meson` and `meson.exe` are wrapper exes
      if not mainfile.endswith('.py'):
          meson_command = [mainfile]
      elif os.path.isabs(mainfile) and mainfile.endswith('mesonmain.py'):
          # Can't actually run meson with an absolute path to mesonmain.py, it must be run as -m mesonbuild.mesonmain
          meson_command = python_command + ['-m', 'mesonbuild.mesonmain']
      else:
          # Either run uninstalled, or full path to meson-script.py
          meson_command = python_command + [mainfile]
      # We print this value for unit tests.
      if 'MESON_COMMAND_TESTS' in os.environ:
          mlog.log('meson_command is {!r}'.format(meson_command))
  
  def is_ascii_string(astring) -> bool:
      try:
          if isinstance(astring, str):
              astring.encode('ascii')
          elif isinstance(astring, bytes):
              astring.decode('ascii')
      except UnicodeDecodeError:
          return False
      return True
  
  def check_direntry_issues(direntry_array):
      import locale
      # Warn if the locale is not UTF-8. This can cause various unfixable issues
      # such as os.stat not being able to decode filenames with unicode in them.
      # There is no way to reset both the preferred encoding and the filesystem
      # encoding, so we can just warn about it.
      e = locale.getpreferredencoding()
      if e.upper() != 'UTF-8' and not is_windows():
          if not isinstance(direntry_array, list):
              direntry_array = [direntry_array]
          for de in direntry_array:
              if is_ascii_string(de):
                  continue
              mlog.warning('''You are using {!r} which is not a Unicode-compatible '
  locale but you are trying to access a file system entry called {!r} which is
  not pure ASCII. This may cause problems.
  '''.format(e, de), file=sys.stderr)
  
  # Put this in objects that should not get dumped to pickle files
  # by accident.
  import threading
  an_unpicklable_object = threading.Lock()
  
  class MesonException(Exception):
      '''Exceptions thrown by Meson'''
  
      def get_msg_with_context(self):
          s = ''
          if hasattr(self, 'lineno') and hasattr(self, 'file'):
              s = get_error_location_string(self.file, self.lineno) + ' '
          s += str(self)
          return s
  
  class EnvironmentException(MesonException):
      '''Exceptions thrown while processing and creating the build environment'''
  
  class FileMode:
      # The first triad is for owner permissions, the second for group permissions,
      # and the third for others (everyone else).
      # For the 1st character:
      #  'r' means can read
      #  '-' means not allowed
      # For the 2nd character:
      #  'w' means can write
      #  '-' means not allowed
      # For the 3rd character:
      #  'x' means can execute
      #  's' means can execute and setuid/setgid is set (owner/group triads only)
      #  'S' means cannot execute and setuid/setgid is set (owner/group triads only)
      #  't' means can execute and sticky bit is set ("others" triads only)
      #  'T' means cannot execute and sticky bit is set ("others" triads only)
      #  '-' means none of these are allowed
      #
      # The meanings of 'rwx' perms is not obvious for directories; see:
      # https://www.hackinglinuxexposed.com/articles/20030424.html
      #
      # For information on this notation such as setuid/setgid/sticky bits, see:
      # https://en.wikipedia.org/wiki/File_system_permissions#Symbolic_notation
      symbolic_perms_regex = re.compile('[r-][w-][xsS-]' # Owner perms
                                        '[r-][w-][xsS-]' # Group perms
                                        '[r-][w-][xtT-]') # Others perms
  
      def __init__(self, perms=None, owner=None, group=None):
          self.perms_s = perms
          self.perms = self.perms_s_to_bits(perms)
          self.owner = owner
          self.group = group
  
      def __repr__(self):
          ret = '<FileMode: {!r} owner={} group={}'
          return ret.format(self.perms_s, self.owner, self.group)
  
      @classmethod
      def perms_s_to_bits(cls, perms_s):
          '''
          Does the opposite of stat.filemode(), converts strings of the form
          'rwxr-xr-x' to st_mode enums which can be passed to os.chmod()
          '''
          if perms_s is None:
              # No perms specified, we will not touch the permissions
              return -1
          eg = 'rwxr-xr-x'
          if not isinstance(perms_s, str):
              msg = 'Install perms must be a string. For example, {!r}'
              raise MesonException(msg.format(eg))
          if len(perms_s) != 9 or not cls.symbolic_perms_regex.match(perms_s):
              msg = 'File perms {!r} must be exactly 9 chars. For example, {!r}'
              raise MesonException(msg.format(perms_s, eg))
          perms = 0
          # Owner perms
          if perms_s[0] == 'r':
              perms |= stat.S_IRUSR
          if perms_s[1] == 'w':
              perms |= stat.S_IWUSR
          if perms_s[2] == 'x':
              perms |= stat.S_IXUSR
          elif perms_s[2] == 'S':
              perms |= stat.S_ISUID
          elif perms_s[2] == 's':
              perms |= stat.S_IXUSR
              perms |= stat.S_ISUID
          # Group perms
          if perms_s[3] == 'r':
              perms |= stat.S_IRGRP
          if perms_s[4] == 'w':
              perms |= stat.S_IWGRP
          if perms_s[5] == 'x':
              perms |= stat.S_IXGRP
          elif perms_s[5] == 'S':
              perms |= stat.S_ISGID
          elif perms_s[5] == 's':
              perms |= stat.S_IXGRP
              perms |= stat.S_ISGID
          # Others perms
          if perms_s[6] == 'r':
              perms |= stat.S_IROTH
          if perms_s[7] == 'w':
              perms |= stat.S_IWOTH
          if perms_s[8] == 'x':
              perms |= stat.S_IXOTH
          elif perms_s[8] == 'T':
              perms |= stat.S_ISVTX
          elif perms_s[8] == 't':
              perms |= stat.S_IXOTH
              perms |= stat.S_ISVTX
          return perms
  
  class File:
      def __init__(self, is_built: bool, subdir: str, fname: str):
          self.is_built = is_built
          self.subdir = subdir
          self.fname = fname
          assert(isinstance(self.subdir, str))
          assert(isinstance(self.fname, str))
  
      def __str__(self) -> str:
          return self.relative_name()
  
      def __repr__(self) -> str:
          ret = '<File: {0}'
          if not self.is_built:
              ret += ' (not built)'
          ret += '>'
          return ret.format(self.relative_name())
  
      @staticmethod
      @lru_cache(maxsize=None)
      def from_source_file(source_root: str, subdir: str, fname: str):
          if not os.path.isfile(os.path.join(source_root, subdir, fname)):
              raise MesonException('File %s does not exist.' % fname)
          return File(False, subdir, fname)
  
      @staticmethod
      def from_built_file(subdir: str, fname: str):
          return File(True, subdir, fname)
  
      @staticmethod
      def from_absolute_file(fname: str):
          return File(False, '', fname)
  
      @lru_cache(maxsize=None)
      def rel_to_builddir(self, build_to_src: str) -> str:
          if self.is_built:
              return self.relative_name()
          else:
              return os.path.join(build_to_src, self.subdir, self.fname)
  
      @lru_cache(maxsize=None)
      def absolute_path(self, srcdir: str, builddir: str) -> str:
          absdir = srcdir
          if self.is_built:
              absdir = builddir
          return os.path.join(absdir, self.relative_name())
  
      def endswith(self, ending: str) -> bool:
          return self.fname.endswith(ending)
  
      def split(self, s: str) -> T.List[str]:
          return self.fname.split(s)
  
      def __eq__(self, other) -> bool:
          return (self.fname, self.subdir, self.is_built) == (other.fname, other.subdir, other.is_built)
  
      def __hash__(self) -> int:
          return hash((self.fname, self.subdir, self.is_built))
  
      @lru_cache(maxsize=None)
      def relative_name(self) -> str:
          return os.path.join(self.subdir, self.fname)
  
  
  def get_compiler_for_source(compilers, src):
      for comp in compilers:
          if comp.can_compile(src):
              return comp
      raise MesonException('No specified compiler can handle file {!s}'.format(src))
  
  def classify_unity_sources(compilers, sources):
      compsrclist = {}
      for src in sources:
          comp = get_compiler_for_source(compilers, src)
          if comp not in compsrclist:
              compsrclist[comp] = [src]
          else:
              compsrclist[comp].append(src)
      return compsrclist
  
  class OrderedEnum(Enum):
      """
      An Enum which additionally offers homogeneous ordered comparison.
      """
      def __ge__(self, other):
          if self.__class__ is other.__class__:
              return self.value >= other.value
          return NotImplemented
  
      def __gt__(self, other):
          if self.__class__ is other.__class__:
              return self.value > other.value
          return NotImplemented
  
      def __le__(self, other):
          if self.__class__ is other.__class__:
              return self.value <= other.value
          return NotImplemented
  
      def __lt__(self, other):
          if self.__class__ is other.__class__:
              return self.value < other.value
          return NotImplemented
  
  
  class MachineChoice(OrderedEnum):
  
      """Enum class representing one of the two abstract machine names used in
      most places: the build, and host, machines.
      """
  
      BUILD = 0
      HOST = 1
  
      def get_lower_case_name(self):
          return PerMachine('build', 'host')[self]
  
      def get_prefix(self):
          return PerMachine('build.', '')[self]
  
  
  class PerMachine(T.Generic[_T]):
      def __init__(self, build: _T, host: _T):
          self.build = build
          self.host = host
  
      def __getitem__(self, machine: MachineChoice) -> _T:
          return {
              MachineChoice.BUILD:  self.build,
              MachineChoice.HOST:   self.host,
          }[machine]
  
      def __setitem__(self, machine: MachineChoice, val: _T) -> None:
          setattr(self, machine.get_lower_case_name(), val)
  
      def miss_defaulting(self) -> "PerMachineDefaultable[T.Optional[_T]]":
          """Unset definition duplicated from their previous to None
  
          This is the inverse of ''default_missing''. By removing defaulted
          machines, we can elaborate the original and then redefault them and thus
          avoid repeating the elaboration explicitly.
          """
          unfreeze = PerMachineDefaultable() # type: PerMachineDefaultable[T.Optional[_T]]
          unfreeze.build = self.build
          unfreeze.host = self.host
          if unfreeze.host == unfreeze.build:
              unfreeze.host = None
          return unfreeze
  
  
  class PerThreeMachine(PerMachine[_T]):
      """Like `PerMachine` but includes `target` too.
  
      It turns out just one thing do we need track the target machine. There's no
      need to computer the `target` field so we don't bother overriding the
      `__getitem__`/`__setitem__` methods.
      """
      def __init__(self, build: _T, host: _T, target: _T):
          super().__init__(build, host)
          self.target = target
  
      def miss_defaulting(self) -> "PerThreeMachineDefaultable[T.Optional[_T]]":
          """Unset definition duplicated from their previous to None
  
          This is the inverse of ''default_missing''. By removing defaulted
          machines, we can elaborate the original and then redefault them and thus
          avoid repeating the elaboration explicitly.
          """
          unfreeze = PerThreeMachineDefaultable() # type: PerThreeMachineDefaultable[T.Optional[_T]]
          unfreeze.build = self.build
          unfreeze.host = self.host
          unfreeze.target = self.target
          if unfreeze.target == unfreeze.host:
              unfreeze.target = None
          if unfreeze.host == unfreeze.build:
              unfreeze.host = None
          return unfreeze
  
      def matches_build_machine(self, machine: MachineChoice) -> bool:
          return self.build == self[machine]
  
  class PerMachineDefaultable(PerMachine[T.Optional[_T]]):
      """Extends `PerMachine` with the ability to default from `None`s.
      """
      def __init__(self) -> None:
          super().__init__(None, None)
  
      def default_missing(self) -> "PerMachine[T.Optional[_T]]":
          """Default host to build
  
          This allows just specifying nothing in the native case, and just host in the
          cross non-compiler case.
          """
          freeze = PerMachine(self.build, self.host)
          if freeze.host is None:
              freeze.host = freeze.build
          return freeze
  
  
  class PerThreeMachineDefaultable(PerMachineDefaultable, PerThreeMachine[T.Optional[_T]]):
      """Extends `PerThreeMachine` with the ability to default from `None`s.
      """
      def __init__(self) -> None:
          PerThreeMachine.__init__(self, None, None, None)
  
      def default_missing(self) -> "PerThreeMachine[T.Optional[_T]]":
          """Default host to build and target to host.
  
          This allows just specifying nothing in the native case, just host in the
          cross non-compiler case, and just target in the native-built
          cross-compiler case.
          """
          freeze = PerThreeMachine(self.build, self.host, self.target)
          if freeze.host is None:
              freeze.host = freeze.build
          if freeze.target is None:
              freeze.target = freeze.host
          return freeze
  
  
  def is_sunos() -> bool:
      return platform.system().lower() == 'sunos'
  
  def is_osx() -> bool:
      return platform.system().lower() == 'darwin'
  
  def is_linux() -> bool:
      return platform.system().lower() == 'linux'
  
  def is_android() -> bool:
      return platform.system().lower() == 'android'
  
  def is_haiku() -> bool:
      return platform.system().lower() == 'haiku'
  
  def is_openbsd() -> bool:
      return platform.system().lower() == 'openbsd'
  
  def is_windows() -> bool:
      platname = platform.system().lower()
      return platname == 'windows' or 'mingw' in platname
  
  def is_cygwin() -> bool:
      return platform.system().lower().startswith('cygwin')
  
  def is_debianlike() -> bool:
      return os.path.isfile('/etc/debian_version')
  
  def is_dragonflybsd() -> bool:
      return platform.system().lower() == 'dragonfly'
  
  def is_netbsd() -> bool:
      return platform.system().lower() == 'netbsd'
  
  def is_freebsd() -> bool:
      return platform.system().lower() == 'freebsd'
  
  def exe_exists(arglist: T.List[str]) -> bool:
      try:
          if subprocess.run(arglist, timeout=10).returncode == 0:
              return True
      except (FileNotFoundError, subprocess.TimeoutExpired):
          pass
      return False
  
  @lru_cache(maxsize=None)
  def darwin_get_object_archs(objpath):
      '''
      For a specific object (executable, static library, dylib, etc), run `lipo`
      to fetch the list of archs supported by it. Supports both thin objects and
      'fat' objects.
      '''
      _, stdo, stderr = Popen_safe(['lipo', '-info', objpath])
      if not stdo:
          mlog.debug('lipo {}: {}'.format(objpath, stderr))
          return None
      stdo = stdo.rsplit(': ', 1)[1]
      # Convert from lipo-style archs to meson-style CPUs
      stdo = stdo.replace('i386', 'x86')
      stdo = stdo.replace('arm64', 'aarch64')
      # Add generic name for armv7 and armv7s
      if 'armv7' in stdo:
          stdo += ' arm'
      return stdo.split()
  
  def detect_vcs(source_dir):
      vcs_systems = [
          dict(name = 'git',        cmd = 'git', repo_dir = '.git', get_rev = 'git describe --dirty=+', rev_regex = '(.*)', dep = '.git/logs/HEAD'),
          dict(name = 'mercurial',  cmd = 'hg',  repo_dir = '.hg',  get_rev = 'hg id -i',               rev_regex = '(.*)', dep = '.hg/dirstate'),
          dict(name = 'subversion', cmd = 'svn', repo_dir = '.svn', get_rev = 'svn info',               rev_regex = 'Revision: (.*)', dep = '.svn/wc.db'),
          dict(name = 'bazaar',     cmd = 'bzr', repo_dir = '.bzr', get_rev = 'bzr revno',              rev_regex = '(.*)', dep = '.bzr'),
      ]
      # FIXME: this is much cleaner with pathlib.Path
      segs = source_dir.replace('\\', '/').split('/')
      for i in range(len(segs), -1, -1):
          curdir = '/'.join(segs[:i])
          for vcs in vcs_systems:
              if os.path.isdir(os.path.join(curdir, vcs['repo_dir'])) and shutil.which(vcs['cmd']):
                  vcs['wc_dir'] = curdir
                  return vcs
      return None
  
  # a helper class which implements the same version ordering as RPM
  class Version:
      def __init__(self, s):
          self._s = s
  
          # split into numeric, alphabetic and non-alphanumeric sequences
          sequences = re.finditer(r'(\d+|[a-zA-Z]+|[^a-zA-Z\d]+)', s)
          # non-alphanumeric separators are discarded
          sequences = [m for m in sequences if not re.match(r'[^a-zA-Z\d]+', m.group(1))]
          # numeric sequences are converted from strings to ints
          sequences = [int(m.group(1)) if m.group(1).isdigit() else m.group(1) for m in sequences]
  
          self._v = sequences
  
      def __str__(self):
          return '%s (V=%s)' % (self._s, str(self._v))
  
      def __repr__(self):
          return '<Version: {}>'.format(self._s)
  
      def __lt__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.lt)
          return NotImplemented
  
      def __gt__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.gt)
          return NotImplemented
  
      def __le__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.le)
          return NotImplemented
  
      def __ge__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.ge)
          return NotImplemented
  
      def __eq__(self, other):
          if isinstance(other, Version):
              return self._v == other._v
          return NotImplemented
  
      def __ne__(self, other):
          if isinstance(other, Version):
              return self._v != other._v
          return NotImplemented
  
      def __cmp(self, other, comparator):
          # compare each sequence in order
          for ours
  
  [... the rest is omitted ...]
files: mesonbuild/mesonlib.py
files: # Copyright 2012-2015 The Meson development team
  
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  
  #     http://www.apache.org/licenses/LICENSE-2.0
  
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  """A library of random helper functionality."""
  from pathlib import Path
  import sys
  import stat
  import time
  import platform, subprocess, operator, os, shlex, shutil, re
  import collections
  from enum import Enum
  from functools import lru_cache, update_wrapper
  from itertools import tee, filterfalse
  import typing as T
  import uuid
  
  from mesonbuild import mlog
  
  _T = T.TypeVar('_T')
  _U = T.TypeVar('_U')
  
  have_fcntl = False
  have_msvcrt = False
  # {subproject: project_meson_version}
  project_meson_versions = {}
  
  try:
      import fcntl
      have_fcntl = True
  except Exception:
      pass
  
  try:
      import msvcrt
      have_msvcrt = True
  except Exception:
      pass
  
  from glob import glob
  
  if os.path.basename(sys.executable) == 'meson.exe':
      # In Windows and using the MSI installed executable.
      python_command = [sys.executable, 'runpython']
  else:
      python_command = [sys.executable]
  meson_command = None
  
  def set_meson_command(mainfile):
      global python_command
      global meson_command
      # On UNIX-like systems `meson` is a Python script
      # On Windows `meson` and `meson.exe` are wrapper exes
      if not mainfile.endswith('.py'):
          meson_command = [mainfile]
      elif os.path.isabs(mainfile) and mainfile.endswith('mesonmain.py'):
          # Can't actually run meson with an absolute path to mesonmain.py, it must be run as -m mesonbuild.mesonmain
          meson_command = python_command + ['-m', 'mesonbuild.mesonmain']
      else:
          # Either run uninstalled, or full path to meson-script.py
          meson_command = python_command + [mainfile]
      # We print this value for unit tests.
      if 'MESON_COMMAND_TESTS' in os.environ:
          mlog.log('meson_command is {!r}'.format(meson_command))
  
  def is_ascii_string(astring) -> bool:
      try:
          if isinstance(astring, str):
              astring.encode('ascii')
          elif isinstance(astring, bytes):
              astring.decode('ascii')
      except UnicodeDecodeError:
          return False
      return True
  
  def check_direntry_issues(direntry_array):
      import locale
      # Warn if the locale is not UTF-8. This can cause various unfixable issues
      # such as os.stat not being able to decode filenames with unicode in them.
      # There is no way to reset both the preferred encoding and the filesystem
      # encoding, so we can just warn about it.
      e = locale.getpreferredencoding()
      if e.upper() != 'UTF-8' and not is_windows():
          if not isinstance(direntry_array, list):
              direntry_array = [direntry_array]
          for de in direntry_array:
              if is_ascii_string(de):
                  continue
              mlog.warning('''You are using {!r} which is not a Unicode-compatible '
  locale but you are trying to access a file system entry called {!r} which is
  not pure ASCII. This may cause problems.
  '''.format(e, de), file=sys.stderr)
  
  # Put this in objects that should not get dumped to pickle files
  # by accident.
  import threading
  an_unpicklable_object = threading.Lock()
  
  class MesonException(Exception):
      '''Exceptions thrown by Meson'''
  
      def get_msg_with_context(self):
          s = ''
          if hasattr(self, 'lineno') and hasattr(self, 'file'):
              s = get_error_location_string(self.file, self.lineno) + ' '
          s += str(self)
          return s
  
  class EnvironmentException(MesonException):
      '''Exceptions thrown while processing and creating the build environment'''
  
  class FileMode:
      # The first triad is for owner permissions, the second for group permissions,
      # and the third for others (everyone else).
      # For the 1st character:
      #  'r' means can read
      #  '-' means not allowed
      # For the 2nd character:
      #  'w' means can write
      #  '-' means not allowed
      # For the 3rd character:
      #  'x' means can execute
      #  's' means can execute and setuid/setgid is set (owner/group triads only)
      #  'S' means cannot execute and setuid/setgid is set (owner/group triads only)
      #  't' means can execute and sticky bit is set ("others" triads only)
      #  'T' means cannot execute and sticky bit is set ("others" triads only)
      #  '-' means none of these are allowed
      #
      # The meanings of 'rwx' perms is not obvious for directories; see:
      # https://www.hackinglinuxexposed.com/articles/20030424.html
      #
      # For information on this notation such as setuid/setgid/sticky bits, see:
      # https://en.wikipedia.org/wiki/File_system_permissions#Symbolic_notation
      symbolic_perms_regex = re.compile('[r-][w-][xsS-]' # Owner perms
                                        '[r-][w-][xsS-]' # Group perms
                                        '[r-][w-][xtT-]') # Others perms
  
      def __init__(self, perms=None, owner=None, group=None):
          self.perms_s = perms
          self.perms = self.perms_s_to_bits(perms)
          self.owner = owner
          self.group = group
  
      def __repr__(self):
          ret = '<FileMode: {!r} owner={} group={}'
          return ret.format(self.perms_s, self.owner, self.group)
  
      @classmethod
      def perms_s_to_bits(cls, perms_s):
          '''
          Does the opposite of stat.filemode(), converts strings of the form
          'rwxr-xr-x' to st_mode enums which can be passed to os.chmod()
          '''
          if perms_s is None:
              # No perms specified, we will not touch the permissions
              return -1
          eg = 'rwxr-xr-x'
          if not isinstance(perms_s, str):
              msg = 'Install perms must be a string. For example, {!r}'
              raise MesonException(msg.format(eg))
          if len(perms_s) != 9 or not cls.symbolic_perms_regex.match(perms_s):
              msg = 'File perms {!r} must be exactly 9 chars. For example, {!r}'
              raise MesonException(msg.format(perms_s, eg))
          perms = 0
          # Owner perms
          if perms_s[0] == 'r':
              perms |= stat.S_IRUSR
          if perms_s[1] == 'w':
              perms |= stat.S_IWUSR
          if perms_s[2] == 'x':
              perms |= stat.S_IXUSR
          elif perms_s[2] == 'S':
              perms |= stat.S_ISUID
          elif perms_s[2] == 's':
              perms |= stat.S_IXUSR
              perms |= stat.S_ISUID
          # Group perms
          if perms_s[3] == 'r':
              perms |= stat.S_IRGRP
          if perms_s[4] == 'w':
              perms |= stat.S_IWGRP
          if perms_s[5] == 'x':
              perms |= stat.S_IXGRP
          elif perms_s[5] == 'S':
              perms |= stat.S_ISGID
          elif perms_s[5] == 's':
              perms |= stat.S_IXGRP
              perms |= stat.S_ISGID
          # Others perms
          if perms_s[6] == 'r':
              perms |= stat.S_IROTH
          if perms_s[7] == 'w':
              perms |= stat.S_IWOTH
          if perms_s[8] == 'x':
              perms |= stat.S_IXOTH
          elif perms_s[8] == 'T':
              perms |= stat.S_ISVTX
          elif perms_s[8] == 't':
              perms |= stat.S_IXOTH
              perms |= stat.S_ISVTX
          return perms
  
  class File:
      def __init__(self, is_built: bool, subdir: str, fname: str):
          self.is_built = is_built
          self.subdir = subdir
          self.fname = fname
          assert(isinstance(self.subdir, str))
          assert(isinstance(self.fname, str))
  
      def __str__(self) -> str:
          return self.relative_name()
  
      def __repr__(self) -> str:
          ret = '<File: {0}'
          if not self.is_built:
              ret += ' (not built)'
          ret += '>'
          return ret.format(self.relative_name())
  
      @staticmethod
      @lru_cache(maxsize=None)
      def from_source_file(source_root: str, subdir: str, fname: str):
          if not os.path.isfile(os.path.join(source_root, subdir, fname)):
              raise MesonException('File %s does not exist.' % fname)
          return File(False, subdir, fname)
  
      @staticmethod
      def from_built_file(subdir: str, fname: str):
          return File(True, subdir, fname)
  
      @staticmethod
      def from_absolute_file(fname: str):
          return File(False, '', fname)
  
      @lru_cache(maxsize=None)
      def rel_to_builddir(self, build_to_src: str) -> str:
          if self.is_built:
              return self.relative_name()
          else:
              return os.path.join(build_to_src, self.subdir, self.fname)
  
      @lru_cache(maxsize=None)
      def absolute_path(self, srcdir: str, builddir: str) -> str:
          absdir = srcdir
          if self.is_built:
              absdir = builddir
          return os.path.join(absdir, self.relative_name())
  
      def endswith(self, ending: str) -> bool:
          return self.fname.endswith(ending)
  
      def split(self, s: str) -> T.List[str]:
          return self.fname.split(s)
  
      def __eq__(self, other) -> bool:
          return (self.fname, self.subdir, self.is_built) == (other.fname, other.subdir, other.is_built)
  
      def __hash__(self) -> int:
          return hash((self.fname, self.subdir, self.is_built))
  
      @lru_cache(maxsize=None)
      def relative_name(self) -> str:
          return os.path.join(self.subdir, self.fname)
  
  
  def get_compiler_for_source(compilers, src):
      for comp in compilers:
          if comp.can_compile(src):
              return comp
      raise MesonException('No specified compiler can handle file {!s}'.format(src))
  
  def classify_unity_sources(compilers, sources):
      compsrclist = {}
      for src in sources:
          comp = get_compiler_for_source(compilers, src)
          if comp not in compsrclist:
              compsrclist[comp] = [src]
          else:
              compsrclist[comp].append(src)
      return compsrclist
  
  class OrderedEnum(Enum):
      """
      An Enum which additionally offers homogeneous ordered comparison.
      """
      def __ge__(self, other):
          if self.__class__ is other.__class__:
              return self.value >= other.value
          return NotImplemented
  
      def __gt__(self, other):
          if self.__class__ is other.__class__:
              return self.value > other.value
          return NotImplemented
  
      def __le__(self, other):
          if self.__class__ is other.__class__:
              return self.value <= other.value
          return NotImplemented
  
      def __lt__(self, other):
          if self.__class__ is other.__class__:
              return self.value < other.value
          return NotImplemented
  
  
  class MachineChoice(OrderedEnum):
  
      """Enum class representing one of the two abstract machine names used in
      most places: the build, and host, machines.
      """
  
      BUILD = 0
      HOST = 1
  
      def get_lower_case_name(self):
          return PerMachine('build', 'host')[self]
  
      def get_prefix(self):
          return PerMachine('build.', '')[self]
  
  
  class PerMachine(T.Generic[_T]):
      def __init__(self, build: _T, host: _T):
          self.build = build
          self.host = host
  
      def __getitem__(self, machine: MachineChoice) -> _T:
          return {
              MachineChoice.BUILD:  self.build,
              MachineChoice.HOST:   self.host,
          }[machine]
  
      def __setitem__(self, machine: MachineChoice, val: _T) -> None:
          setattr(self, machine.get_lower_case_name(), val)
  
      def miss_defaulting(self) -> "PerMachineDefaultable[T.Optional[_T]]":
          """Unset definition duplicated from their previous to None
  
          This is the inverse of ''default_missing''. By removing defaulted
          machines, we can elaborate the original and then redefault them and thus
          avoid repeating the elaboration explicitly.
          """
          unfreeze = PerMachineDefaultable() # type: PerMachineDefaultable[T.Optional[_T]]
          unfreeze.build = self.build
          unfreeze.host = self.host
          if unfreeze.host == unfreeze.build:
              unfreeze.host = None
          return unfreeze
  
  
  class PerThreeMachine(PerMachine[_T]):
      """Like `PerMachine` but includes `target` too.
  
      It turns out just one thing do we need track the target machine. There's no
      need to computer the `target` field so we don't bother overriding the
      `__getitem__`/`__setitem__` methods.
      """
      def __init__(self, build: _T, host: _T, target: _T):
          super().__init__(build, host)
          self.target = target
  
      def miss_defaulting(self) -> "PerThreeMachineDefaultable[T.Optional[_T]]":
          """Unset definition duplicated from their previous to None
  
          This is the inverse of ''default_missing''. By removing defaulted
          machines, we can elaborate the original and then redefault them and thus
          avoid repeating the elaboration explicitly.
          """
          unfreeze = PerThreeMachineDefaultable() # type: PerThreeMachineDefaultable[T.Optional[_T]]
          unfreeze.build = self.build
          unfreeze.host = self.host
          unfreeze.target = self.target
          if unfreeze.target == unfreeze.host:
              unfreeze.target = None
          if unfreeze.host == unfreeze.build:
              unfreeze.host = None
          return unfreeze
  
      def matches_build_machine(self, machine: MachineChoice) -> bool:
          return self.build == self[machine]
  
  class PerMachineDefaultable(PerMachine[T.Optional[_T]]):
      """Extends `PerMachine` with the ability to default from `None`s.
      """
      def __init__(self) -> None:
          super().__init__(None, None)
  
      def default_missing(self) -> "PerMachine[T.Optional[_T]]":
          """Default host to build
  
          This allows just specifying nothing in the native case, and just host in the
          cross non-compiler case.
          """
          freeze = PerMachine(self.build, self.host)
          if freeze.host is None:
              freeze.host = freeze.build
          return freeze
  
  
  class PerThreeMachineDefaultable(PerMachineDefaultable, PerThreeMachine[T.Optional[_T]]):
      """Extends `PerThreeMachine` with the ability to default from `None`s.
      """
      def __init__(self) -> None:
          PerThreeMachine.__init__(self, None, None, None)
  
      def default_missing(self) -> "PerThreeMachine[T.Optional[_T]]":
          """Default host to build and target to host.
  
          This allows just specifying nothing in the native case, just host in the
          cross non-compiler case, and just target in the native-built
          cross-compiler case.
          """
          freeze = PerThreeMachine(self.build, self.host, self.target)
          if freeze.host is None:
              freeze.host = freeze.build
          if freeze.target is None:
              freeze.target = freeze.host
          return freeze
  
  
  def is_sunos() -> bool:
      return platform.system().lower() == 'sunos'
  
  def is_osx() -> bool:
      return platform.system().lower() == 'darwin'
  
  def is_linux() -> bool:
      return platform.system().lower() == 'linux'
  
  def is_android() -> bool:
      return platform.system().lower() == 'android'
  
  def is_haiku() -> bool:
      return platform.system().lower() == 'haiku'
  
  def is_openbsd() -> bool:
      return platform.system().lower() == 'openbsd'
  
  def is_windows() -> bool:
      platname = platform.system().lower()
      return platname == 'windows' or 'mingw' in platname
  
  def is_cygwin() -> bool:
      return platform.system().lower().startswith('cygwin')
  
  def is_debianlike() -> bool:
      return os.path.isfile('/etc/debian_version')
  
  def is_dragonflybsd() -> bool:
      return platform.system().lower() == 'dragonfly'
  
  def is_netbsd() -> bool:
      return platform.system().lower() == 'netbsd'
  
  def is_freebsd() -> bool:
      return platform.system().lower() == 'freebsd'
  
  def exe_exists(arglist: T.List[str]) -> bool:
      try:
          if subprocess.run(arglist, timeout=10).returncode == 0:
              return True
      except (FileNotFoundError, subprocess.TimeoutExpired):
          pass
      return False
  
  @lru_cache(maxsize=None)
  def darwin_get_object_archs(objpath):
      '''
      For a specific object (executable, static library, dylib, etc), run `lipo`
      to fetch the list of archs supported by it. Supports both thin objects and
      'fat' objects.
      '''
      _, stdo, stderr = Popen_safe(['lipo', '-info', objpath])
      if not stdo:
          mlog.debug('lipo {}: {}'.format(objpath, stderr))
          return None
      stdo = stdo.rsplit(': ', 1)[1]
      # Convert from lipo-style archs to meson-style CPUs
      stdo = stdo.replace('i386', 'x86')
      stdo = stdo.replace('arm64', 'aarch64')
      # Add generic name for armv7 and armv7s
      if 'armv7' in stdo:
          stdo += ' arm'
      return stdo.split()
  
  def detect_vcs(source_dir):
      vcs_systems = [
          dict(name = 'git',        cmd = 'git', repo_dir = '.git', get_rev = 'git describe --dirty=+', rev_regex = '(.*)', dep = '.git/logs/HEAD'),
          dict(name = 'mercurial',  cmd = 'hg',  repo_dir = '.hg',  get_rev = 'hg id -i',               rev_regex = '(.*)', dep = '.hg/dirstate'),
          dict(name = 'subversion', cmd = 'svn', repo_dir = '.svn', get_rev = 'svn info',               rev_regex = 'Revision: (.*)', dep = '.svn/wc.db'),
          dict(name = 'bazaar',     cmd = 'bzr', repo_dir = '.bzr', get_rev = 'bzr revno',              rev_regex = '(.*)', dep = '.bzr'),
      ]
      # FIXME: this is much cleaner with pathlib.Path
      segs = source_dir.replace('\\', '/').split('/')
      for i in range(len(segs), -1, -1):
          curdir = '/'.join(segs[:i])
          for vcs in vcs_systems:
              if os.path.isdir(os.path.join(curdir, vcs['repo_dir'])) and shutil.which(vcs['cmd']):
                  vcs['wc_dir'] = curdir
                  return vcs
      return None
  
  # a helper class which implements the same version ordering as RPM
  class Version:
      def __init__(self, s):
          self._s = s
  
          # split into numeric, alphabetic and non-alphanumeric sequences
          sequences = re.finditer(r'(\d+|[a-zA-Z]+|[^a-zA-Z\d]+)', s)
          # non-alphanumeric separators are discarded
          sequences = [m for m in sequences if not re.match(r'[^a-zA-Z\d]+', m.group(1))]
          # numeric sequences are converted from strings to ints
          sequences = [int(m.group(1)) if m.group(1).isdigit() else m.group(1) for m in sequences]
  
          self._v = sequences
  
      def __str__(self):
          return '%s (V=%s)' % (self._s, str(self._v))
  
      def __repr__(self):
          return '<Version: {}>'.format(self._s)
  
      def __lt__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.lt)
          return NotImplemented
  
      def __gt__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.gt)
          return NotImplemented
  
      def __le__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.le)
          return NotImplemented
  
      def __ge__(self, other):
          if isinstance(other, Version):
              return self.__cmp(other, operator.ge)
          return NotImplemented
  
      def __eq__(self, other):
          if isinstance(other, Version):
              return self._v == other._v
          return NotImplemented
  
      def __ne__(self, other):
          if isinstance(other, Version):
              return self._v != other._v
          return NotImplemented
  
      def __cmp(self, other, comparator):
          # compare each sequence in order
          for ours, theirs in zip(self._v, other._v):
              # sort a non-digit sequence before a digit sequence
              ours_is_int = isinstance(ours, int)
              theirs_is_int = isinstance(theirs, int)
              if ours_is_int != theirs_is_int:
                  return comparator(ours_is_int, theirs_is_int)
  
              if ours != theirs:
                  return comparator(ours, theirs)
  
          # if equal length, all components have matched, so equal
          # otherwise, the version with a suffix remaining is greater
          return comparator(len(self._v), len(other._v))
  
  def _version_extract_cmpop(vstr2: str) -> T.Tuple[T.Callable[[T.Any, T.Any], bool], str]:
      if v
  
  [... the rest is omitted ...]
files: mesonbuild/mesonlib.py
files: import os, subprocess
  import argparse
  
  from . import mlog
  from .mesonlib import git, Popen_safe
  from .wrap.wrap import API_ROOT, PackageDefinition, Resolver, WrapException
  from .wrap import wraptool
  
  def update_wrapdb_file(wrap, repo_dir, options):
      patch_url = wrap.get('patch_url')
      branch, revision = wraptool.parse_patch_url(patch_url)
      new_branch, new_revision = wraptool.get_latest_version(wrap.name)
      if new_branch == branch and new_revision == revision:
          mlog.log('  -> Up to date.')
          return
      wraptool.update_wrap_file(wrap.filename, wrap.name, new_branch, new_revision)
      msg = ['  -> New wrap file downloaded.']
      # Meson reconfigure won't use the new wrap file as long as the source
      # directory exists. We don't delete it ourself to avoid data loss in case
      # user has changes in their copy.
      if os.path.isdir(repo_dir):
          msg += ['To use it, delete', mlog.bold(repo_dir), 'and run', mlog.bold('meson --reconfigure')]
      mlog.log(*msg)
  
  def update_file(wrap, repo_dir, options):
      patch_url = wrap.values.get('patch_url', '')
      if patch_url.startswith(API_ROOT):
          update_wrapdb_file(wrap, repo_dir, options)
      elif not os.path.isdir(repo_dir):
          # The subproject is not needed, or it is a tarball extracted in
          # 'libfoo-1.0' directory and the version has been bumped and the new
          # directory is 'libfoo-2.0'. In that case forcing a meson
          # reconfigure will download and use the new tarball.
          mlog.log('  -> Subproject has not been checked out. Run', mlog.bold('meson --reconfigure'), 'to fetch it if needed.')
      else:
          # The subproject has not changed, or the new source and/or patch
          # tarballs should be extracted in the same directory than previous
          # version.
          mlog.log('  -> Subproject has not changed, or the new source/patch needs to be extracted on the same location.\n' +
                   '     In that case, delete', mlog.bold(repo_dir), 'and run', mlog.bold('meson --reconfigure'))
  
  def git_output(cmd, workingdir):
      return git(cmd, workingdir, check=True, universal_newlines=True,
                 stdout=subprocess.PIPE, stderr=subprocess.STDOUT).stdout
  
  def git_show(repo_dir):
      commit_message = git_output(['show', '--quiet', '--pretty=format:%h%n%d%n%s%n[%an]'], repo_dir)
      parts = [s.strip() for s in commit_message.split('\n')]
      mlog.log('  ->', mlog.yellow(parts[0]), mlog.red(parts[1]), parts[2], mlog.blue(parts[3]))
  
  def update_git(wrap, repo_dir, options):
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not used.')
          return
      revision = wrap.get('revision')
      ret = git_output(['rev-parse', '--abbrev-ref', 'HEAD'], repo_dir).strip()
      if ret == 'HEAD':
          try:
              # We are currently in detached mode, just checkout the new revision
              git_output(['fetch'], repo_dir)
              git_output(['checkout', revision], repo_dir)
          except subprocess.CalledProcessError as e:
              out = e.output.decode().strip()
              mlog.log('  -> Could not checkout revision', mlog.cyan(revision))
              mlog.log(mlog.red(out))
              mlog.log(mlog.red(str(e)))
              return
      elif ret == revision:
          try:
              # We are in the same branch, pull latest commits
              git_output(['-c', 'rebase.autoStash=true', 'pull', '--rebase'], repo_dir)
          except subprocess.CalledProcessError as e:
              out = e.output.decode().strip()
              mlog.log('  -> Could not rebase', mlog.bold(repo_dir), 'please fix and try again.')
              mlog.log(mlog.red(out))
              mlog.log(mlog.red(str(e)))
              return
      else:
          # We are in another branch, probably user created their own branch and
          # we should rebase it on top of wrap's branch.
          if options.rebase:
              try:
                  git_output(['fetch'], repo_dir)
                  git_output(['-c', 'rebase.autoStash=true', 'rebase', revision], repo_dir)
              except subprocess.CalledProcessError as e:
                  out = e.output.decode().strip()
                  mlog.log('  -> Could not rebase', mlog.bold(repo_dir), 'please fix and try again.')
                  mlog.log(mlog.red(out))
                  mlog.log(mlog.red(str(e)))
                  return
          else:
              mlog.log('  -> Target revision is', mlog.bold(revision), 'but currently in branch is', mlog.bold(ret), '\n' +
                       '     To rebase your branch on top of', mlog.bold(revision), 'use', mlog.bold('--rebase'), 'option.')
              return
  
      git_output(['submodule', 'update', '--checkout', '--recursive'], repo_dir)
      git_show(repo_dir)
  
  def update_hg(wrap, repo_dir, options):
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not used.')
          return
      revno = wrap.get('revision')
      if revno.lower() == 'tip':
          # Failure to do pull is not a fatal error,
          # because otherwise you can't develop without
          # a working net connection.
          subprocess.call(['hg', 'pull'], cwd=repo_dir)
      else:
          if subprocess.call(['hg', 'checkout', revno], cwd=repo_dir) != 0:
              subprocess.check_call(['hg', 'pull'], cwd=repo_dir)
              subprocess.check_call(['hg', 'checkout', revno], cwd=repo_dir)
  
  def update_svn(wrap, repo_dir, options):
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not used.')
          return
      revno = wrap.get('revision')
      p, out, _ = Popen_safe(['svn', 'info', '--show-item', 'revision', repo_dir])
      current_revno = out
      if current_revno == revno:
          return
      if revno.lower() == 'head':
          # Failure to do pull is not a fatal error,
          # because otherwise you can't develop without
          # a working net connection.
          subprocess.call(['svn', 'update'], cwd=repo_dir)
      else:
          subprocess.check_call(['svn', 'update', '-r', revno], cwd=repo_dir)
  
  def update(wrap, repo_dir, options):
      mlog.log('Updating %s...' % wrap.name)
      if wrap.type == 'file':
          update_file(wrap, repo_dir, options)
      elif wrap.type == 'git':
          update_git(wrap, repo_dir, options)
      elif wrap.type == 'hg':
          update_hg(wrap, repo_dir, options)
      elif wrap.type == 'svn':
          update_svn(wrap, repo_dir, options)
      else:
          mlog.log('  -> Cannot update', wrap.type, 'subproject')
  
  def checkout(wrap, repo_dir, options):
      if wrap.type != 'git' or not os.path.isdir(repo_dir):
          return
      branch_name = options.branch_name if options.branch_name else wrap.get('revision')
      cmd = ['checkout', branch_name, '--']
      if options.b:
          cmd.insert(1, '-b')
      mlog.log('Checkout %s in %s...' % (branch_name, wrap.name))
      try:
          git_output(cmd, repo_dir)
          git_show(repo_dir)
      except subprocess.CalledProcessError as e:
          out = e.output.decode().strip()
          mlog.log('  -> ', mlog.red(out))
  
  def download(wrap, repo_dir, options):
      mlog.log('Download %s...' % wrap.name)
      if os.path.isdir(repo_dir):
          mlog.log('  -> Already downloaded')
          return
      try:
          r = Resolver(os.path.dirname(repo_dir))
          r.resolve(wrap.name, 'meson')
          mlog.log('  -> done')
      except WrapException as e:
          mlog.log('  ->', mlog.red(str(e)))
  
  def foreach(wrap, repo_dir, options):
      mlog.log('Executing command in %s' % repo_dir)
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not downloaded yet')
          return
      try:
          out = subprocess.check_output([options.command] + options.args,
                                        stderr=subprocess.STDOUT,
                                        cwd=repo_dir).decode()
          mlog.log(out, end='')
      except subprocess.CalledProcessError as e:
          err_message = "Command '%s' returned non-zero exit status %d." % (" ".join(e.cmd), e.returncode)
          out = e.output.decode()
          mlog.log('  -> ', mlog.red(err_message))
          mlog.log(out, end='')
      except Exception as e:
          mlog.log('  -> ', mlog.red(str(e)))
  
  def add_common_arguments(p):
      p.add_argument('--sourcedir', default='.',
                     help='Path to source directory')
      p.add_argument('subprojects', nargs='*',
                     help='List of subprojects (default: all)')
  
  def add_arguments(parser):
      subparsers = parser.add_subparsers(title='Commands', dest='command')
      subparsers.required = True
  
      p = subparsers.add_parser('update', help='Update all subprojects from wrap files')
      p.add_argument('--rebase', default=False, action='store_true',
                     help='Rebase your branch on top of wrap\'s revision (git only)')
      add_common_arguments(p)
      p.set_defaults(subprojects_func=update)
  
      p = subparsers.add_parser('checkout', help='Checkout a branch (git only)')
      p.add_argument('-b', default=False, action='store_true',
                     help='Create a new branch')
      p.add_argument('branch_name', nargs='?',
                     help='Name of the branch to checkout or create (default: revision set in wrap file)')
      add_common_arguments(p)
      p.set_defaults(subprojects_func=checkout)
  
      p = subparsers.add_parser('download', help='Ensure subprojects are fetched, even if not in use. ' +
                                                 'Already downloaded subprojects are not modified. ' +
                                                 'This can be used to pre-fetch all subprojects and avoid downloads during configure.')
      add_common_arguments(p)
      p.set_defaults(subprojects_func=download)
  
      p = subparsers.add_parser('foreach', help='Execute a command in each subproject directory.')
      p.add_argument('command', metavar='command ...',
                     help='Command to execute in each subproject directory')
      p.add_argument('args', nargs=argparse.REMAINDER,
                     help=argparse.SUPPRESS)
      p.add_argument('--sourcedir', default='.',
                     help='Path to source directory')
      p.set_defaults(subprojects_func=foreach)
  
  def run(options):
      src_dir = os.path.relpath(os.path.realpath(options.sourcedir))
      if not os.path.isfile(os.path.join(src_dir, 'meson.build')):
          mlog.error('Directory', mlog.bold(src_dir), 'does not seem to be a Meson source directory.')
          return 1
      subprojects_dir = os.path.join(src_dir, 'subprojects')
      if not os.path.isdir(subprojects_dir):
          mlog.log('Directory', mlog.bold(src_dir), 'does not seem to have subprojects.')
          return 0
      files = []
      if hasattr(options, 'subprojects'):
          for name in options.subprojects:
              f = os.path.join(subprojects_dir, name + '.wrap')
              if not os.path.isfile(f):
                  mlog.error('Subproject', mlog.bold(name), 'not found.')
                  return 1
              else:
                  files.append(f)
      if not files:
          for f in os.listdir(subprojects_dir):
              if f.endswith('.wrap'):
                  files.append(os.path.join(subprojects_dir, f))
      for f in files:
          wrap = PackageDefinition(f)
          directory = wrap.values.get('directory', wrap.name)
          repo_dir = os.path.join(subprojects_dir, directory)
          options.subprojects_func(wrap, repo_dir, options)
      return 0
  
  
  [... the rest is omitted ...]
files: mesonbuild/msubprojects.py
files: import os, subprocess
  import argparse
  
  from . import mlog
  from .mesonlib import Popen_safe
  from .wrap.wrap import API_ROOT, PackageDefinition, Resolver, WrapException
  from .wrap import wraptool
  
  def update_wrapdb_file(wrap, repo_dir, options):
      patch_url = wrap.get('patch_url')
      branch, revision = wraptool.parse_patch_url(patch_url)
      new_branch, new_revision = wraptool.get_latest_version(wrap.name)
      if new_branch == branch and new_revision == revision:
          mlog.log('  -> Up to date.')
          return
      wraptool.update_wrap_file(wrap.filename, wrap.name, new_branch, new_revision)
      msg = ['  -> New wrap file downloaded.']
      # Meson reconfigure won't use the new wrap file as long as the source
      # directory exists. We don't delete it ourself to avoid data loss in case
      # user has changes in their copy.
      if os.path.isdir(repo_dir):
          msg += ['To use it, delete', mlog.bold(repo_dir), 'and run', mlog.bold('meson --reconfigure')]
      mlog.log(*msg)
  
  def update_file(wrap, repo_dir, options):
      patch_url = wrap.values.get('patch_url', '')
      if patch_url.startswith(API_ROOT):
          update_wrapdb_file(wrap, repo_dir, options)
      elif not os.path.isdir(repo_dir):
          # The subproject is not needed, or it is a tarball extracted in
          # 'libfoo-1.0' directory and the version has been bumped and the new
          # directory is 'libfoo-2.0'. In that case forcing a meson
          # reconfigure will download and use the new tarball.
          mlog.log('  -> Subproject has not been checked out. Run', mlog.bold('meson --reconfigure'), 'to fetch it if needed.')
      else:
          # The subproject has not changed, or the new source and/or patch
          # tarballs should be extracted in the same directory than previous
          # version.
          mlog.log('  -> Subproject has not changed, or the new source/patch needs to be extracted on the same location.\n' +
                   '     In that case, delete', mlog.bold(repo_dir), 'and run', mlog.bold('meson --reconfigure'))
  
  def git_output(cmd, workingdir):
      return subprocess.check_output(['git', '-C', workingdir] + cmd,
                                     # Redirect stdin to DEVNULL otherwise git
                                     # messes up the console and ANSI colors stop
                                     # working on Windows.
                                     stdin=subprocess.DEVNULL,
                                     stderr=subprocess.STDOUT).decode()
  
  def git_show(repo_dir):
      commit_message = git_output(['show', '--quiet', '--pretty=format:%h%n%d%n%s%n[%an]'], repo_dir)
      parts = [s.strip() for s in commit_message.split('\n')]
      mlog.log('  ->', mlog.yellow(parts[0]), mlog.red(parts[1]), parts[2], mlog.blue(parts[3]))
  
  def update_git(wrap, repo_dir, options):
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not used.')
          return
      revision = wrap.get('revision')
      ret = git_output(['rev-parse', '--abbrev-ref', 'HEAD'], repo_dir).strip()
      if ret == 'HEAD':
          try:
              # We are currently in detached mode, just checkout the new revision
              git_output(['fetch'], repo_dir)
              git_output(['checkout', revision], repo_dir)
          except subprocess.CalledProcessError as e:
              out = e.output.decode().strip()
              mlog.log('  -> Could not checkout revision', mlog.cyan(revision))
              mlog.log(mlog.red(out))
              mlog.log(mlog.red(str(e)))
              return
      elif ret == revision:
          try:
              # We are in the same branch, pull latest commits
              git_output(['-c', 'rebase.autoStash=true', 'pull', '--rebase'], repo_dir)
          except subprocess.CalledProcessError as e:
              out = e.output.decode().strip()
              mlog.log('  -> Could not rebase', mlog.bold(repo_dir), 'please fix and try again.')
              mlog.log(mlog.red(out))
              mlog.log(mlog.red(str(e)))
              return
      else:
          # We are in another branch, probably user created their own branch and
          # we should rebase it on top of wrap's branch.
          if options.rebase:
              try:
                  git_output(['fetch'], repo_dir)
                  git_output(['-c', 'rebase.autoStash=true', 'rebase', revision], repo_dir)
              except subprocess.CalledProcessError as e:
                  out = e.output.decode().strip()
                  mlog.log('  -> Could not rebase', mlog.bold(repo_dir), 'please fix and try again.')
                  mlog.log(mlog.red(out))
                  mlog.log(mlog.red(str(e)))
                  return
          else:
              mlog.log('  -> Target revision is', mlog.bold(revision), 'but currently in branch is', mlog.bold(ret), '\n' +
                       '     To rebase your branch on top of', mlog.bold(revision), 'use', mlog.bold('--rebase'), 'option.')
              return
  
      git_output(['submodule', 'update', '--checkout', '--recursive'], repo_dir)
      git_show(repo_dir)
  
  def update_hg(wrap, repo_dir, options):
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not used.')
          return
      revno = wrap.get('revision')
      if revno.lower() == 'tip':
          # Failure to do pull is not a fatal error,
          # because otherwise you can't develop without
          # a working net connection.
          subprocess.call(['hg', 'pull'], cwd=repo_dir)
      else:
          if subprocess.call(['hg', 'checkout', revno], cwd=repo_dir) != 0:
              subprocess.check_call(['hg', 'pull'], cwd=repo_dir)
              subprocess.check_call(['hg', 'checkout', revno], cwd=repo_dir)
  
  def update_svn(wrap, repo_dir, options):
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not used.')
          return
      revno = wrap.get('revision')
      p, out, _ = Popen_safe(['svn', 'info', '--show-item', 'revision', repo_dir])
      current_revno = out
      if current_revno == revno:
          return
      if revno.lower() == 'head':
          # Failure to do pull is not a fatal error,
          # because otherwise you can't develop without
          # a working net connection.
          subprocess.call(['svn', 'update'], cwd=repo_dir)
      else:
          subprocess.check_call(['svn', 'update', '-r', revno], cwd=repo_dir)
  
  def update(wrap, repo_dir, options):
      mlog.log('Updating %s...' % wrap.name)
      if wrap.type == 'file':
          update_file(wrap, repo_dir, options)
      elif wrap.type == 'git':
          update_git(wrap, repo_dir, options)
      elif wrap.type == 'hg':
          update_hg(wrap, repo_dir, options)
      elif wrap.type == 'svn':
          update_svn(wrap, repo_dir, options)
      else:
          mlog.log('  -> Cannot update', wrap.type, 'subproject')
  
  def checkout(wrap, repo_dir, options):
      if wrap.type != 'git' or not os.path.isdir(repo_dir):
          return
      branch_name = options.branch_name if options.branch_name else wrap.get('revision')
      cmd = ['checkout', branch_name, '--']
      if options.b:
          cmd.insert(1, '-b')
      mlog.log('Checkout %s in %s...' % (branch_name, wrap.name))
      try:
          git_output(cmd, repo_dir)
          git_show(repo_dir)
      except subprocess.CalledProcessError as e:
          out = e.output.decode().strip()
          mlog.log('  -> ', mlog.red(out))
  
  def download(wrap, repo_dir, options):
      mlog.log('Download %s...' % wrap.name)
      if os.path.isdir(repo_dir):
          mlog.log('  -> Already downloaded')
          return
      try:
          r = Resolver(os.path.dirname(repo_dir))
          r.resolve(wrap.name, 'meson')
          mlog.log('  -> done')
      except WrapException as e:
          mlog.log('  ->', mlog.red(str(e)))
  
  def foreach(wrap, repo_dir, options):
      mlog.log('Executing command in %s' % repo_dir)
      if not os.path.isdir(repo_dir):
          mlog.log('  -> Not downloaded yet')
          return
      try:
          out = subprocess.check_output([options.command] + options.args,
                                        stderr=subprocess.STDOUT,
                                        cwd=repo_dir).decode()
          mlog.log(out, end='')
      except subprocess.CalledProcessError as e:
          err_message = "Command '%s' returned non-zero exit status %d." % (" ".join(e.cmd), e.returncode)
          out = e.output.decode()
          mlog.log('  -> ', mlog.red(err_message))
          mlog.log(out, end='')
      except Exception as e:
          mlog.log('  -> ', mlog.red(str(e)))
  
  def add_common_arguments(p):
      p.add_argument('--sourcedir', default='.',
                     help='Path to source directory')
      p.add_argument('subprojects', nargs='*',
                     help='List of subprojects (default: all)')
  
  def add_arguments(parser):
      subparsers = parser.add_subparsers(title='Commands', dest='command')
      subparsers.required = True
  
      p = subparsers.add_parser('update', help='Update all subprojects from wrap files')
      p.add_argument('--rebase', default=False, action='store_true',
                     help='Rebase your branch on top of wrap\'s revision (git only)')
      add_common_arguments(p)
      p.set_defaults(subprojects_func=update)
  
      p = subparsers.add_parser('checkout', help='Checkout a branch (git only)')
      p.add_argument('-b', default=False, action='store_true',
                     help='Create a new branch')
      p.add_argument('branch_name', nargs='?',
                     help='Name of the branch to checkout or create (default: revision set in wrap file)')
      add_common_arguments(p)
      p.set_defaults(subprojects_func=checkout)
  
      p = subparsers.add_parser('download', help='Ensure subprojects are fetched, even if not in use. ' +
                                                 'Already downloaded subprojects are not modified. ' +
                                                 'This can be used to pre-fetch all subprojects and avoid downloads during configure.')
      add_common_arguments(p)
      p.set_defaults(subprojects_func=download)
  
      p = subparsers.add_parser('foreach', help='Execute a command in each subproject directory.')
      p.add_argument('command', metavar='command ...',
                     help='Command to execute in each subproject directory')
      p.add_argument('args', nargs=argparse.REMAINDER,
                     help=argparse.SUPPRESS)
      p.add_argument('--sourcedir', default='.',
                     help='Path to source directory')
      p.set_defaults(subprojects_func=foreach)
  
  def run(options):
      src_dir = os.path.relpath(os.path.realpath(options.sourcedir))
      if not os.path.isfile(os.path.join(src_dir, 'meson.build')):
          mlog.error('Directory', mlog.bold(src_dir), 'does not seem to be a Meson source directory.')
          return 1
      subprojects_dir = os.path.join(src_dir, 'subprojects')
      if not os.path.isdir(subprojects_dir):
          mlog.log('Directory', mlog.bold(src_dir), 'does not seem to have subprojects.')
          return 0
      files = []
      if hasattr(options, 'subprojects'):
          for name in options.subprojects:
              f = os.path.join(subprojects_dir, name + '.wrap')
              if not os.path.isfile(f):
                  mlog.error('Subproject', mlog.bold(name), 'not found.')
                  return 1
              else:
                  files.append(f)
      if not files:
          for f in os.listdir(subprojects_dir):
              if f.endswith('.wrap'):
                  files.append(os.path.join(subprojects_dir, f))
      for f in files:
          wrap = PackageDefinition(f)
          directory = wrap.values.get('directory', wrap.name)
          repo_dir = os.path.join(subprojects_dir, directory)
          options.subprojects_func(wrap, repo_dir, options)
      return 0
  
  
  [... the rest is omitted ...]
files: mesonbuild/msubprojects.py
files: # Copyright 2015 The Meson development team
  
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  
  #     http://www.apache.org/licenses/LICENSE-2.0
  
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  from .. import mlog
  import contextlib
  import urllib.request
  import urllib.error
  import urllib.parse
  import os
  import hashlib
  import shutil
  import tempfile
  import stat
  import subprocess
  import sys
  import configparser
  import typing as T
  
  from . import WrapMode
  from ..mesonlib import git, GIT, ProgressBar, MesonException
  
  if T.TYPE_CHECKING:
      import http.client
  
  try:
      # Importing is just done to check if SSL exists, so all warnings
      # regarding 'imported but unused' can be safely ignored
      import ssl  # noqa
      has_ssl = True
      API_ROOT = 'https://wrapdb.mesonbuild.com/v1/'
  except ImportError:
      has_ssl = False
      API_ROOT = 'http://wrapdb.mesonbuild.com/v1/'
  
  REQ_TIMEOUT = 600.0
  SSL_WARNING_PRINTED = False
  WHITELIST_SUBDOMAIN = 'wrapdb.mesonbuild.com'
  
  def quiet_git(cmd: T.List[str], workingdir: str) -> T.Tuple[bool, str]:
      if not GIT:
          return False, 'Git program not found.'
      pc = git(cmd, workingdir, universal_newlines=True,
               stdout=subprocess.PIPE, stderr=subprocess.PIPE)
      if pc.returncode != 0:
          return False, pc.stderr
      return True, pc.stdout
  
  def verbose_git(cmd: T.List[str], workingdir: str, check: bool = False) -> bool:
      if not GIT:
          return False
      return git(cmd, workingdir, check=check).returncode == 0
  
  def whitelist_wrapdb(urlstr: str) -> urllib.parse.ParseResult:
      """ raises WrapException if not whitelisted subdomain """
      url = urllib.parse.urlparse(urlstr)
      if not url.hostname:
          raise WrapException('{} is not a valid URL'.format(urlstr))
      if not url.hostname.endswith(WHITELIST_SUBDOMAIN):
          raise WrapException('{} is not a whitelisted WrapDB URL'.format(urlstr))
      if has_ssl and not url.scheme == 'https':
          raise WrapException('WrapDB did not have expected SSL https url, instead got {}'.format(urlstr))
      return url
  
  def open_wrapdburl(urlstring: str) -> 'http.client.HTTPResponse':
      global SSL_WARNING_PRINTED
  
      url = whitelist_wrapdb(urlstring)
      if has_ssl:
          try:
              return urllib.request.urlopen(urllib.parse.urlunparse(url), timeout=REQ_TIMEOUT)
          except urllib.error.URLError as excp:
              raise WrapException('WrapDB connection failed to {} with error {}'.format(urlstring, excp))
  
      # following code is only for those without Python SSL
      nossl_url = url._replace(scheme='http')
      if not SSL_WARNING_PRINTED:
          mlog.warning('SSL module not available in {}: WrapDB traffic not authenticated.'.format(sys.executable))
          SSL_WARNING_PRINTED = True
      try:
          return urllib.request.urlopen(urllib.parse.urlunparse(nossl_url), timeout=REQ_TIMEOUT)
      except urllib.error.URLError as excp:
          raise WrapException('WrapDB connection failed to {} with error {}'.format(urlstring, excp))
  
  
  class WrapException(MesonException):
      pass
  
  class WrapNotFoundException(WrapException):
      pass
  
  class PackageDefinition:
      def __init__(self, fname: str):
          self.filename = fname
          self.basename = os.path.basename(fname)
          self.name = self.basename[:-5]
          try:
              self.config = configparser.ConfigParser(interpolation=None)
              self.config.read(fname)
          except configparser.Error:
              raise WrapException('Failed to parse {}'.format(self.basename))
          if len(self.config.sections()) < 1:
              raise WrapException('Missing sections in {}'.format(self.basename))
          self.wrap_section = self.config.sections()[0]
          if not self.wrap_section.startswith('wrap-'):
              m = '{!r} is not a valid first section in {}'
              raise WrapException(m.format(self.wrap_section, self.basename))
          self.type = self.wrap_section[5:]
          self.values = dict(self.config[self.wrap_section])
  
      def get(self, key: str) -> str:
          try:
              return self.values[key]
          except KeyError:
              m = 'Missing key {!r} in {}'
              raise WrapException(m.format(key, self.basename))
  
      def has_patch(self) -> bool:
          return 'patch_url' in self.values
  
  def load_wrap(subdir_root: str, packagename: str) -> PackageDefinition:
      fname = os.path.join(subdir_root, packagename + '.wrap')
      if os.path.isfile(fname):
          return PackageDefinition(fname)
      return None
  
  def get_directory(subdir_root: str, packagename: str):
      directory = packagename
      # We always have to load the wrap file, if it exists, because it could
      # override the default directory name.
      wrap = load_wrap(subdir_root, packagename)
      if wrap and 'directory' in wrap.values:
          directory = wrap.get('directory')
          if os.path.dirname(directory):
              raise WrapException('Directory key must be a name and not a path')
      return wrap, directory
  
  class Resolver:
      def __init__(self, subdir_root: str, wrap_mode=WrapMode.default):
          self.wrap_mode = wrap_mode
          self.subdir_root = subdir_root
          self.cachedir = os.path.join(self.subdir_root, 'packagecache')
  
      def resolve(self, packagename: str, method: str) -> str:
          self.packagename = packagename
          self.wrap, self.directory = get_directory(self.subdir_root, self.packagename)
          self.dirname = os.path.join(self.subdir_root, self.directory)
          meson_file = os.path.join(self.dirname, 'meson.build')
          cmake_file = os.path.join(self.dirname, 'CMakeLists.txt')
  
          if method not in ['meson', 'cmake']:
              raise WrapException('Only the methods "meson" and "cmake" are supported')
  
          # The directory is there and has meson.build? Great, use it.
          if method == 'meson' and os.path.exists(meson_file):
              return self.directory
          if method == 'cmake' and os.path.exists(cmake_file):
              return self.directory
  
          # Check if the subproject is a git submodule
          self.resolve_git_submodule()
  
          if os.path.exists(self.dirname):
              if not os.path.isdir(self.dirname):
                  raise WrapException('Path already exists but is not a directory')
          else:
              # A wrap file is required to download
              if not self.wrap:
                  m = 'Subproject directory not found and {}.wrap file not found'
                  raise WrapNotFoundException(m.format(self.packagename))
  
              if self.wrap.type == 'file':
                  self.get_file()
              else:
                  self.check_can_download()
                  if self.wrap.type == 'git':
                      self.get_git()
                  elif self.wrap.type == "hg":
                      self.get_hg()
                  elif self.wrap.type == "svn":
                      self.get_svn()
                  else:
                      raise WrapException('Unknown wrap type {!r}'.format(self.wrap.type))
  
          # A meson.build or CMakeLists.txt file is required in the directory
          if method == 'meson' and not os.path.exists(meson_file):
              raise WrapException('Subproject exists but has no meson.build file')
          if method == 'cmake' and not os.path.exists(cmake_file):
              raise WrapException('Subproject exists but has no CMakeLists.txt file')
  
          return self.directory
  
      def check_can_download(self) -> None:
          # Don't download subproject data based on wrap file if requested.
          # Git submodules are ok (see above)!
          if self.wrap_mode is WrapMode.nodownload:
              m = 'Automatic wrap-based subproject downloading is disabled'
              raise WrapException(m)
  
      def resolve_git_submodule(self) -> bool:
          if not GIT:
              raise WrapException('Git program not found.')
          # Are we in a git repository?
          ret, out = quiet_git(['rev-parse'], self.subdir_root)
          if not ret:
              return False
          # Is `dirname` a submodule?
          ret, out = quiet_git(['submodule', 'status', self.dirname], self.subdir_root)
          if not ret:
              return False
          # Submodule has not been added, add it
          if out.startswith('+'):
              mlog.warning('git submodule might be out of date')
              return True
          elif out.startswith('U'):
              raise WrapException('git submodule has merge conflicts')
          # Submodule exists, but is deinitialized or wasn't initialized
          elif out.startswith('-'):
              if verbose_git(['submodule', 'update', '--init', self.dirname], self.subdir_root):
                  return True
              raise WrapException('git submodule failed to init')
          # Submodule looks fine, but maybe it wasn't populated properly. Do a checkout.
          elif out.startswith(' '):
              verbose_git(['checkout', '.'], self.dirname)
              # Even if checkout failed, try building it anyway and let the user
              # handle any problems manually.
              return True
          elif out == '':
              # It is not a submodule, just a folder that exists in the main repository.
              return False
          m = 'Unknown git submodule output: {!r}'
          raise WrapException(m.format(out))
  
      def get_file(self) -> None:
          path = self.get_file_internal('source')
          extract_dir = self.subdir_root
          # Some upstreams ship packages that do not have a leading directory.
          # Create one for them.
          if 'lead_directory_missing' in self.wrap.values:
              os.mkdir(self.dirname)
              extract_dir = self.dirname
          shutil.unpack_archive(path, extract_dir)
          if self.wrap.has_patch():
              self.apply_patch()
  
      def get_git(self) -> None:
          if not GIT:
              raise WrapException('Git program not found.')
          revno = self.wrap.get('revision')
          is_shallow = False
          depth_option = []    # type: T.List[str]
          if self.wrap.values.get('depth', '') != '':
              is_shallow = True
              depth_option = ['--depth', self.wrap.values.get('depth')]
          # for some reason git only allows commit ids to be shallowly fetched by fetch not with clone
          if is_shallow and self.is_git_full_commit_id(revno):
              # git doesn't support directly cloning shallowly for commits,
              # so we follow https://stackoverflow.com/a/43136160
              verbose_git(['init', self.directory], self.subdir_root, check=True)
              verbose_git(['remote', 'add', 'origin', self.wrap.get('url')], self.dirname, check=True)
              revno = self.wrap.get('revision')
              verbose_git(['fetch', *depth_option, 'origin', revno], self.dirname, check=True)
              verbose_git(['checkout', revno], self.dirname, check=True)
              if self.wrap.values.get('clone-recursive', '').lower() == 'true':
                  verbose_git(['submodule', 'update', '--init', '--checkout',
                               '--recursive', *depth_option], self.dirname, check=True)
              push_url = self.wrap.values.get('push-url')
              if push_url:
                  verbose_git(['remote', 'set-url', '--push', 'origin', push_url], self.dirname, check=True)
          else:
              if not is_shallow:
                  verbose_git(['clone', self.wrap.get('url'), self.directory], self.subdir_root, check=True)
                  if revno.lower() != 'head':
                      if verbose_git(['checkout', revno], self.dirname):
                          verbose_git(['fetch', self.wrap.get('url'), revno], self.dirname, check=True)
                          verbose_git(['checkout', revno], self.dirname, check=True)
              else:
                  verbose_git(['clone', *depth_option, '--branch', revno, self.wrap.get('url'),
                               self.directory], self.subdir_root, check=True)
              if self.wrap.values.get('clone-recursive', '').lower() == 'true':
                  verbose_git(['submodule', 'update', '--init', '--checkout', '--recursive', *depth_option],
                              self.dirname, check=True)
              push_url = self.wrap.values.get('push-url')
              if push_url:
                  verbose_git(['remote', 'set-url', '--push', 'origin', push_url], self.dirname, check=True)
  
      def is_git_full_commit_id(self, revno: str) -> bool:
          result = False
          if len(revno) in (40, 64): # 40 for sha1, 64 for upcoming sha256
              result = all((ch in '0123456789AaBbCcDdEeFf' for ch in revno))
          return result
  
      def get_hg(self) -> None:
          revno = self.wrap.get('revision')
          hg = shutil.which('hg')
          if not hg:
              raise WrapException('Mercurial program not found.')
          subprocess.check_call([hg, 'clone', self.wrap.get('url'),
                                 self.directory], cwd=self.subdir_root)
          if revno.lower() != 'tip':
              subprocess.check_call([hg, 'checkout', revno],
                                    cwd=self.dirname)
  
      def get_svn(self) -> None:
          revno = self.wrap.get('revision')
          svn = shutil.which('svn')
          if not svn:
              raise WrapException('SVN program not found.')
          subprocess.check_call([svn, 'checkout', '-r', revno, self.wrap.get('url'),
                                 self.directory], cwd=self.subdir_root)
  
      def get_data(self, urlstring: str) -> T.Tuple[str, str]:
          blocksize = 10 * 1024
          h = hashlib.sha256()
          tmpfile = tempfile.NamedTemporaryFile(mode='wb', dir=self.cachedir, delete=False)
          url = urllib.parse.urlparse(urlstring)
          if url.hostname and url.hostname.endswith(WHITELIST_SUBDOMAIN):
              resp = open_wrapdburl(urlstring)
          elif WHITELIST_SUBDOMAIN in urlstring:
              raise WrapException('{} may be a WrapDB-impersonating URL'.format(urlstring))
          else:
              try:
                  resp = urllib.request.urlopen(urlstring, timeout=REQ_TIMEOUT)
              except urllib.error.URLError:
                  raise WrapException('could not get {} is the internet available?'.format(urlstring))
          with contextlib.closing(resp) as resp:
              try:
                  dlsize = int(resp.info()['Content-Length'])
              except TypeError:
                  dlsize = None
              if dlsize is None:
                  print('Downloading file of unknown size.')
                  while True:
                      block = resp.read(blocksize)
                      if block == b'':
                          break
                      h.update(block)
                      tmpfile.write(block)
                  hashvalue = h.hexdigest()
                  return hashvalue, tmpfile.name
              sys.stdout.flush()
              progress_bar = ProgressBar(bar_type='download', total=dlsize,
                                         desc='Downloading')
              while True:
                  block = resp.read(blocksize)
                  if block == b'':
                      break
                  h.update(block)
                  tmpfile.write(block)
                  progress_bar.update(len(block))
              progress_bar.close()
              hashvalue = h.hexdigest()
          return hashvalue, tmpfile.name
  
      def check_hash(self, what: str, path: str) -> None:
          expected = self.wrap.get(what + '_hash')
          h = hashlib.sha256()
          with open(path, 'rb') as f:
              h.update(f.read())
          dhash = h.hexdigest()
          if dhash != expected:
              raise WrapException('Incorrect hash for {}:\n {} expected\n {} actual.'.format(what, expected, dhash))
  
      def download(self, what: str, ofname: str) -> None:
          self.check_can_download()
          srcurl = self.wrap.get(what + '_url')
          mlog.log('Downloading', mlog.bold(self.packagename), what, 'from', mlog.bold(srcurl))
          dhash, tmpfile = self.get_data(srcurl)
          expected = self.wrap.get(what + '_hash')
          if dhash != expected:
              os.remove(tmpfile)
              raise WrapException('Incorrect hash for {}:\n {} expected\n {} actual.'.format(what, expected, dhash))
          os.rename(tmpfile, ofname)
  
      def get_file_internal(self, what: str) -> str:
          filename = self.wrap.get(what + '_filename')
          cache_path = os.path.join(self.cachedir, filename)
  
          if os.path.exists(cache_path):
              self.check_hash(what, cache_path)
              mlog.log('Using', mlog.bold(self.packagename), what, 'from cache.')
              return cache_path
  
          if not os.path.isdir(self.cachedir):
              os.mkdir(self.cachedir)
          self.download(what, cache_path)
          return cache_path
  
      def apply_patch(self) -> None:
          path = self.get_file_internal('patch')
          try:
              shutil.unpack_archive(path, self.subdir_root)
          except Exception:
              with tempfile.TemporaryDirectory() as workdir:
                  shutil.unpack_archive(path, workdir)
                  self.copy_tree(workdir, self.subdir_root)
  
      def copy_tree(self, root_src_dir: str, root_dst_dir: str) -> None:
          """
          Copy directory tree. Overwrites also read only files.
          """
          for src_dir, _, files in os.walk(root_src_dir):
              dst_dir = src_dir.replace(root_src_dir, root_dst_dir, 1)
              if not os.path.exists(dst_dir):
                  os.makedirs(dst_dir)
              for file_ in files:
                  src_file = os.path.join(src_dir, file_)
                  dst_file = os.path.join(dst_dir, file_)
                  if os.path.exists(dst_file):
                      try:
                          os.remove(dst_file)
                      except PermissionError:
                          os.chmod(dst_file, stat.S_IWUSR)
                          os.remove(dst_file)
                  shutil.copy2(src_file, dst_dir)
  
  
  [... the rest is omitted ...]
files: mesonbuild/wrap/wrap.py
files: # Copyright 2015 The Meson development team
  
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  
  #     http://www.apache.org/licenses/LICENSE-2.0
  
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  from .. import mlog
  import contextlib
  import urllib.request
  import urllib.error
  import urllib.parse
  import os
  import platform
  import hashlib
  import shutil
  import tempfile
  import stat
  import subprocess
  import sys
  import configparser
  import typing as T
  
  from . import WrapMode
  from ..mesonlib import ProgressBar, MesonException
  
  if T.TYPE_CHECKING:
      import http.client
  
  try:
      # Importing is just done to check if SSL exists, so all warnings
      # regarding 'imported but unused' can be safely ignored
      import ssl  # noqa
      has_ssl = True
      API_ROOT = 'https://wrapdb.mesonbuild.com/v1/'
  except ImportError:
      has_ssl = False
      API_ROOT = 'http://wrapdb.mesonbuild.com/v1/'
  
  GIT = shutil.which('git')
  REQ_TIMEOUT = 600.0
  SSL_WARNING_PRINTED = False
  WHITELIST_SUBDOMAIN = 'wrapdb.mesonbuild.com'
  
  def git(cmd: T.List[str], workingdir: str, **kwargs) -> subprocess.CompletedProcess:
      pc = subprocess.run([GIT, '-C', workingdir] + cmd,
                          # Redirect stdin to DEVNULL otherwise git messes up the
                          # console and ANSI colors stop working on Windows.
                          stdin=subprocess.DEVNULL, **kwargs)
      # Sometimes git calls git recursively, such as `git submodule update
      # --recursive` which will be without the above workaround, so set the
      # console mode again just in case.
      if platform.system().lower() == 'windows':
          mlog._windows_ansi()
      return pc
  
  def quiet_git(cmd: T.List[str], workingdir: str) -> T.Tuple[bool, str]:
      if not GIT:
          return False, 'Git program not found.'
      pc = git(cmd, workingdir, universal_newlines=True,
               stdout=subprocess.PIPE, stderr=subprocess.PIPE)
      if pc.returncode != 0:
          return False, pc.stderr
      return True, pc.stdout
  
  def verbose_git(cmd: T.List[str], workingdir: str, check: bool = False) -> bool:
      if not GIT:
          return False
      return git(cmd, workingdir, check=check).returncode == 0
  
  def whitelist_wrapdb(urlstr: str) -> urllib.parse.ParseResult:
      """ raises WrapException if not whitelisted subdomain """
      url = urllib.parse.urlparse(urlstr)
      if not url.hostname:
          raise WrapException('{} is not a valid URL'.format(urlstr))
      if not url.hostname.endswith(WHITELIST_SUBDOMAIN):
          raise WrapException('{} is not a whitelisted WrapDB URL'.format(urlstr))
      if has_ssl and not url.scheme == 'https':
          raise WrapException('WrapDB did not have expected SSL https url, instead got {}'.format(urlstr))
      return url
  
  def open_wrapdburl(urlstring: str) -> 'http.client.HTTPResponse':
      global SSL_WARNING_PRINTED
  
      url = whitelist_wrapdb(urlstring)
      if has_ssl:
          try:
              return urllib.request.urlopen(urllib.parse.urlunparse(url), timeout=REQ_TIMEOUT)
          except urllib.error.URLError as excp:
              raise WrapException('WrapDB connection failed to {} with error {}'.format(urlstring, excp))
  
      # following code is only for those without Python SSL
      nossl_url = url._replace(scheme='http')
      if not SSL_WARNING_PRINTED:
          mlog.warning('SSL module not available in {}: WrapDB traffic not authenticated.'.format(sys.executable))
          SSL_WARNING_PRINTED = True
      try:
          return urllib.request.urlopen(urllib.parse.urlunparse(nossl_url), timeout=REQ_TIMEOUT)
      except urllib.error.URLError as excp:
          raise WrapException('WrapDB connection failed to {} with error {}'.format(urlstring, excp))
  
  
  class WrapException(MesonException):
      pass
  
  class WrapNotFoundException(WrapException):
      pass
  
  class PackageDefinition:
      def __init__(self, fname: str):
          self.filename = fname
          self.basename = os.path.basename(fname)
          self.name = self.basename[:-5]
          try:
              self.config = configparser.ConfigParser(interpolation=None)
              self.config.read(fname)
          except configparser.Error:
              raise WrapException('Failed to parse {}'.format(self.basename))
          if len(self.config.sections()) < 1:
              raise WrapException('Missing sections in {}'.format(self.basename))
          self.wrap_section = self.config.sections()[0]
          if not self.wrap_section.startswith('wrap-'):
              m = '{!r} is not a valid first section in {}'
              raise WrapException(m.format(self.wrap_section, self.basename))
          self.type = self.wrap_section[5:]
          self.values = dict(self.config[self.wrap_section])
  
      def get(self, key: str) -> str:
          try:
              return self.values[key]
          except KeyError:
              m = 'Missing key {!r} in {}'
              raise WrapException(m.format(key, self.basename))
  
      def has_patch(self) -> bool:
          return 'patch_url' in self.values
  
  def load_wrap(subdir_root: str, packagename: str) -> PackageDefinition:
      fname = os.path.join(subdir_root, packagename + '.wrap')
      if os.path.isfile(fname):
          return PackageDefinition(fname)
      return None
  
  def get_directory(subdir_root: str, packagename: str):
      directory = packagename
      # We always have to load the wrap file, if it exists, because it could
      # override the default directory name.
      wrap = load_wrap(subdir_root, packagename)
      if wrap and 'directory' in wrap.values:
          directory = wrap.get('directory')
          if os.path.dirname(directory):
              raise WrapException('Directory key must be a name and not a path')
      return wrap, directory
  
  class Resolver:
      def __init__(self, subdir_root: str, wrap_mode=WrapMode.default):
          self.wrap_mode = wrap_mode
          self.subdir_root = subdir_root
          self.cachedir = os.path.join(self.subdir_root, 'packagecache')
  
      def resolve(self, packagename: str, method: str) -> str:
          self.packagename = packagename
          self.wrap, self.directory = get_directory(self.subdir_root, self.packagename)
          self.dirname = os.path.join(self.subdir_root, self.directory)
          meson_file = os.path.join(self.dirname, 'meson.build')
          cmake_file = os.path.join(self.dirname, 'CMakeLists.txt')
  
          if method not in ['meson', 'cmake']:
              raise WrapException('Only the methods "meson" and "cmake" are supported')
  
          # The directory is there and has meson.build? Great, use it.
          if method == 'meson' and os.path.exists(meson_file):
              return self.directory
          if method == 'cmake' and os.path.exists(cmake_file):
              return self.directory
  
          # Check if the subproject is a git submodule
          self.resolve_git_submodule()
  
          if os.path.exists(self.dirname):
              if not os.path.isdir(self.dirname):
                  raise WrapException('Path already exists but is not a directory')
          else:
              # A wrap file is required to download
              if not self.wrap:
                  m = 'Subproject directory not found and {}.wrap file not found'
                  raise WrapNotFoundException(m.format(self.packagename))
  
              if self.wrap.type == 'file':
                  self.get_file()
              else:
                  self.check_can_download()
                  if self.wrap.type == 'git':
                      self.get_git()
                  elif self.wrap.type == "hg":
                      self.get_hg()
                  elif self.wrap.type == "svn":
                      self.get_svn()
                  else:
                      raise WrapException('Unknown wrap type {!r}'.format(self.wrap.type))
  
          # A meson.build or CMakeLists.txt file is required in the directory
          if method == 'meson' and not os.path.exists(meson_file):
              raise WrapException('Subproject exists but has no meson.build file')
          if method == 'cmake' and not os.path.exists(cmake_file):
              raise WrapException('Subproject exists but has no CMakeLists.txt file')
  
          return self.directory
  
      def check_can_download(self) -> None:
          # Don't download subproject data based on wrap file if requested.
          # Git submodules are ok (see above)!
          if self.wrap_mode is WrapMode.nodownload:
              m = 'Automatic wrap-based subproject downloading is disabled'
              raise WrapException(m)
  
      def resolve_git_submodule(self) -> bool:
          if not GIT:
              raise WrapException('Git program not found.')
          # Are we in a git repository?
          ret, out = quiet_git(['rev-parse'], self.subdir_root)
          if not ret:
              return False
          # Is `dirname` a submodule?
          ret, out = quiet_git(['submodule', 'status', self.dirname], self.subdir_root)
          if not ret:
              return False
          # Submodule has not been added, add it
          if out.startswith('+'):
              mlog.warning('git submodule might be out of date')
              return True
          elif out.startswith('U'):
              raise WrapException('git submodule has merge conflicts')
          # Submodule exists, but is deinitialized or wasn't initialized
          elif out.startswith('-'):
              if verbose_git(['submodule', 'update', '--init', self.dirname], self.subdir_root):
                  return True
              raise WrapException('git submodule failed to init')
          # Submodule looks fine, but maybe it wasn't populated properly. Do a checkout.
          elif out.startswith(' '):
              verbose_git(['checkout', '.'], self.dirname)
              # Even if checkout failed, try building it anyway and let the user
              # handle any problems manually.
              return True
          elif out == '':
              # It is not a submodule, just a folder that exists in the main repository.
              return False
          m = 'Unknown git submodule output: {!r}'
          raise WrapException(m.format(out))
  
      def get_file(self) -> None:
          path = self.get_file_internal('source')
          extract_dir = self.subdir_root
          # Some upstreams ship packages that do not have a leading directory.
          # Create one for them.
          if 'lead_directory_missing' in self.wrap.values:
              os.mkdir(self.dirname)
              extract_dir = self.dirname
          shutil.unpack_archive(path, extract_dir)
          if self.wrap.has_patch():
              self.apply_patch()
  
      def get_git(self) -> None:
          if not GIT:
              raise WrapException('Git program not found.')
          revno = self.wrap.get('revision')
          is_shallow = False
          depth_option = []    # type: T.List[str]
          if self.wrap.values.get('depth', '') != '':
              is_shallow = True
              depth_option = ['--depth', self.wrap.values.get('depth')]
          # for some reason git only allows commit ids to be shallowly fetched by fetch not with clone
          if is_shallow and self.is_git_full_commit_id(revno):
              # git doesn't support directly cloning shallowly for commits,
              # so we follow https://stackoverflow.com/a/43136160
              verbose_git(['init', self.directory], self.subdir_root, check=True)
              verbose_git(['remote', 'add', 'origin', self.wrap.get('url')], self.dirname, check=True)
              revno = self.wrap.get('revision')
              verbose_git(['fetch', *depth_option, 'origin', revno], self.dirname, check=True)
              verbose_git(['checkout', revno], self.dirname, check=True)
              if self.wrap.values.get('clone-recursive', '').lower() == 'true':
                  verbose_git(['submodule', 'update', '--init', '--checkout',
                               '--recursive', *depth_option], self.dirname, check=True)
              push_url = self.wrap.values.get('push-url')
              if push_url:
                  verbose_git(['remote', 'set-url', '--push', 'origin', push_url], self.dirname, check=True)
          else:
              if not is_shallow:
                  verbose_git(['clone', self.wrap.get('url'), self.directory], self.subdir_root, check=True)
                  if revno.lower() != 'head':
                      if verbose_git(['checkout', revno], self.dirname):
                          verbose_git(['fetch', self.wrap.get('url'), revno], self.dirname, check=True)
                          verbose_git(['checkout', revno], self.dirname, check=True)
              else:
                  verbose_git(['clone', *depth_option, '--branch', revno, self.wrap.get('url'),
                               self.directory], self.subdir_root, check=True)
              if self.wrap.values.get('clone-recursive', '').lower() == 'true':
                  verbose_git(['submodule', 'update', '--init', '--checkout', '--recursive', *depth_option],
                              self.dirname, check=True)
              push_url = self.wrap.values.get('push-url')
              if push_url:
                  verbose_git(['remote', 'set-url', '--push', 'origin', push_url], self.dirname, check=True)
  
      def is_git_full_commit_id(self, revno: str) -> bool:
          result = False
          if len(revno) in (40, 64): # 40 for sha1, 64 for upcoming sha256
              result = all((ch in '0123456789AaBbCcDdEeFf' for ch in revno))
          return result
  
      def get_hg(self) -> None:
          revno = self.wrap.get('revision')
          hg = shutil.which('hg')
          if not hg:
              raise WrapException('Mercurial program not found.')
          subprocess.check_call([hg, 'clone', self.wrap.get('url'),
                                 self.directory], cwd=self.subdir_root)
          if revno.lower() != 'tip':
              subprocess.check_call([hg, 'checkout', revno],
                                    cwd=self.dirname)
  
      def get_svn(self) -> None:
          revno = self.wrap.get('revision')
          svn = shutil.which('svn')
          if not svn:
              raise WrapException('SVN program not found.')
          subprocess.check_call([svn, 'checkout', '-r', revno, self.wrap.get('url'),
                                 self.directory], cwd=self.subdir_root)
  
      def get_data(self, urlstring: str) -> T.Tuple[str, str]:
          blocksize = 10 * 1024
          h = hashlib.sha256()
          tmpfile = tempfile.NamedTemporaryFile(mode='wb', dir=self.cachedir, delete=False)
          url = urllib.parse.urlparse(urlstring)
          if url.hostname and url.hostname.endswith(WHITELIST_SUBDOMAIN):
              resp = open_wrapdburl(urlstring)
          elif WHITELIST_SUBDOMAIN in urlstring:
              raise WrapException('{} may be a WrapDB-impersonating URL'.format(urlstring))
          else:
              try:
                  resp = urllib.request.urlopen(urlstring, timeout=REQ_TIMEOUT)
              except urllib.error.URLError:
                  raise WrapException('could not get {} is the internet available?'.format(urlstring))
          with contextlib.closing(resp) as resp:
              try:
                  dlsize = int(resp.info()['Content-Length'])
              except TypeError:
                  dlsize = None
              if dlsize is None:
                  print('Downloading file of unknown size.')
                  while True:
                      block = resp.read(blocksize)
                      if block == b'':
                          break
                      h.update(block)
                      tmpfile.write(block)
                  hashvalue = h.hexdigest()
                  return hashvalue, tmpfile.name
              sys.stdout.flush()
              progress_bar = ProgressBar(bar_type='download', total=dlsize,
                                         desc='Downloading')
              while True:
                  block = resp.read(blocksize)
                  if block == b'':
                      break
                  h.update(block)
                  tmpfile.write(block)
                  progress_bar.update(len(block))
              progress_bar.close()
              hashvalue = h.hexdigest()
          return hashvalue, tmpfile.name
  
      def check_hash(self, what: str, path: str) -> None:
          expected = self.wrap.get(what + '_hash')
          h = hashlib.sha256()
          with open(path, 'rb') as f:
              h.update(f.read())
          dhash = h.hexdigest()
          if dhash != expected:
              raise WrapException('Incorrect hash for {}:\n {} expected\n {} actual.'.format(what, expected, dhash))
  
      def download(self, what: str, ofname: str) -> None:
          self.check_can_download()
          srcurl = self.wrap.get(what + '_url')
          mlog.log('Downloading', mlog.bold(self.packagename), what, 'from', mlog.bold(srcurl))
          dhash, tmpfile = self.get_data(srcurl)
          expected = self.wrap.get(what + '_hash')
          if dhash != expected:
              os.remove(tmpfile)
              raise WrapException('Incorrect hash for {}:\n {} expected\n {} actual.'.format(what, expected, dhash))
          os.rename(tmpfile, ofname)
  
      def get_file_internal(self, what: str) -> str:
          filename = self.wrap.get(what + '_filename')
          cache_path = os.path.join(self.cachedir, filename)
  
          if os.path.exists(cache_path):
              self.check_hash(what, cache_path)
              mlog.log('Using', mlog.bold(self.packagename), what, 'from cache.')
              return cache_path
  
          if not os.path.isdir(self.cachedir):
              os.mkdir(self.cachedir)
          self.download(what, cache_path)
          return cache_path
  
      def apply_patch(self) -> None:
          path = self.get_file_internal('patch')
          try:
              shutil.unpack_archive(path, self.subdir_root)
          except Exception:
              with tempfile.TemporaryDirectory() as workdir:
                  shutil.unpack_archive(path, workdir)
                  self.copy_tree(workdir, self.subdir_root)
  
      def copy_tree(self, root_src_dir: str, root_dst_dir: str) -> None:
          """
          Copy directory tree. Overwrites also read only files.
          """
          for src_dir, _, files in os.walk(root_src_dir):
              dst_dir = src_dir.replace(root_src_dir, root_dst_dir, 1)
              if not os.path.exists(dst_dir):
                  os.makedirs(dst_dir)
              for file_ in files:
                  src_file = os.path.join(src_dir, file_)
                  dst_file = os.path.join(dst_dir, file_)
                  if os.path.exists(dst_file):
                      try:
                          os.remove(dst_file)
                      except PermissionError:
                          os.chmod(dst_file, stat.S_IWUSR)
                          os.remove(dst_file)
                  shutil.copy2(src_file, dst_dir)
  
  
  [... the rest is omitted ...]
files: mesonbuild/wrap/wrap.py