tools/download_fuchsia_sdk.py

#!/usr/bin/env python3
# Copyright 2013 The Flutter Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

# The return code of this script will always be 0, even if there is an error,
# unless the --fail-loudly flag is passed.

import argparse
import tarfile
import json
import os
import shutil
import subprocess
import sys

SRC_ROOT = os.path.dirname(
    os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
)
FUCHSIA_SDK_DIR = os.path.join(SRC_ROOT, 'fuchsia', 'sdk')
FLUTTER_DIR = os.path.join(SRC_ROOT, 'flutter')


# Prints to stderr.
def eprint(*args, **kwargs):
  print(*args, file=sys.stderr, **kwargs)


def FileNameForSdkPath(sdk_path):
  return sdk_path.split('/')[-1]


def DownloadFuchsiaSDKFromGCS(sdk_path, verbose):
  file = FileNameForSdkPath(sdk_path)
  url = 'https://storage.googleapis.com/fuchsia-artifacts/{}'.format(sdk_path)
  dest = os.path.join(FUCHSIA_SDK_DIR, file)

  if verbose:
    print('Fuchsia SDK url: "%s"' % url)
    print('Fuchsia SDK destination path: "%s"' % dest)

  if os.path.isfile(dest):
    os.unlink(dest)

  # Ensure destination folder exists.
  os.makedirs(FUCHSIA_SDK_DIR, exist_ok=True)
  curl_command = [
      'curl',
      '--retry',
      '3',
      '--continue-at',
      '-',
      '--location',
      '--output',
      dest,
      url,
  ]
  if verbose:
    print('Running: "%s"' % (' '.join(curl_command)))
  curl_result = subprocess.run(
      curl_command,
      stdout=subprocess.PIPE,
      stderr=subprocess.PIPE,
      universal_newlines=True,
  )
  if curl_result.returncode == 0 and verbose:
    print(
        'curl output:stdout:\n{}\nstderr:\n{}'.format(
            curl_result.stdout,
            curl_result.stderr,
        )
    )
  elif curl_result.returncode != 0:
    eprint(
        'Failed to download: stdout:\n{}\nstderr:\n{}'.format(
            curl_result.stdout,
            curl_result.stderr,
        )
    )
    return None

  return dest


def OnErrorRmTree(func, path, exc_info):
  """
  Error handler for ``shutil.rmtree``.

  If the error is due to an access error (read only file)
  it attempts to add write permission and then retries.
  If the error is for another reason it re-raises the error.

  Usage : ``shutil.rmtree(path, onerror=onerror)``
  """
  import stat
  # Is the error an access error?
  if not os.access(path, os.W_OK):
    os.chmod(path, stat.S_IWUSR)
    func(path)
  else:
    raise


def ExtractGzipArchive(archive, host_os, verbose):
  sdk_dest = os.path.join(FUCHSIA_SDK_DIR, host_os)
  if os.path.isdir(sdk_dest):
    shutil.rmtree(sdk_dest, onerror=OnErrorRmTree)

  extract_dest = os.path.join(FUCHSIA_SDK_DIR, 'temp')
  if os.path.isdir(extract_dest):
    shutil.rmtree(extract_dest, onerror=OnErrorRmTree)
  os.makedirs(extract_dest, exist_ok=True)

  if verbose:
    print('Extracting "%s" to "%s"' % (archive, extract_dest))

  with tarfile.open(archive, 'r') as z:
    z.extractall(extract_dest)

  shutil.move(extract_dest, sdk_dest)


def Main():
  parser = argparse.ArgumentParser()
  parser.add_argument(
      '--fail-loudly',
      action='store_true',
      default=False,
      help="Return an error code if a prebuilt couldn't be fetched and extracted"
  )

  parser.add_argument(
      '--verbose',
      action='store_true',
      default='LUCI_CONTEXT' in os.environ,
      help='Emit verbose output'
  )

  parser.add_argument('--host-os', help='The host os')

  parser.add_argument(
      '--fuchsia-sdk-path',
      help='The path in gcs to the fuchsia sdk to download'
  )

  args = parser.parse_args()
  fail_loudly = 1 if args.fail_loudly else 0
  verbose = args.verbose
  host_os = args.host_os
  fuchsia_sdk_path = args.fuchsia_sdk_path

  if fuchsia_sdk_path is None:
    eprint('sdk_path can not be empty')
    return fail_loudly

  archive = DownloadFuchsiaSDKFromGCS(fuchsia_sdk_path, verbose)
  if archive is None:
    eprint('Failed to download SDK from %s' % fuchsia_sdk_path)
    return fail_loudly

  ExtractGzipArchive(archive, host_os, verbose)

  success = True
  return 0 if success else fail_loudly


if __name__ == '__main__':
  sys.exit(Main())