releasetools: Make common Python 3 compatible.

from __future__ import print_function

import base64

import collections

import copy

import errno

import re

import shlex

import shutil

import string

import subprocess

import sys

import tempfile

    kwargs: Any additional args to be passed to subprocess.Popen(), such as env,

        stdin, etc. stdout and stderr will default to subprocess.PIPE and

        subprocess.STDOUT respectively unless caller specifies any of them.

        universal_newlines will default to True, as most of the users in

        releasetools expect string output.

  Returns:

    A subprocess.Popen object.

  if 'stdout' not in kwargs and 'stderr' not in kwargs:

    kwargs['stdout'] = subprocess.PIPE

    kwargs['stderr'] = subprocess.STDOUT

  if 'universal_newlines' not in kwargs:

    kwargs['universal_newlines'] = True

  # Don't log any if caller explicitly says so.

  if verbose != False:

    logger.info("  Running: \"%s\"", " ".join(args))

  def read_helper(fn):

    if isinstance(input_file, zipfile.ZipFile):

      return input_file.read(fn)

      return input_file.read(fn).decode()

    else:

      path = os.path.join(input_file, *fn.split("/"))

      try:

  # system. Other areas assume system is always at "/system" so point /system

  # at /.

  if system_root_image:

    assert not d.has_key("/system") and d.has_key("/")

    assert '/system' not in d and '/' in d

    d["/system"] = d["/"]

  return d

    # filename listed in system.map may contain an additional leading slash

    # (i.e. "//system/framework/am.jar"). Using lstrip to get consistent

    # results.

    arcname = string.replace(entry, which, which.upper(), 1).lstrip('/')

    arcname = entry.replace(which, which.upper(), 1).lstrip('/')

    # Special handling another case, where files not under /system

    # (e.g. "/sbin/charger") are packed under ROOT/ in a target_files.zip.

    if basename:

      installed_files.add(basename)

  for line in tf_zip.read("META/apkcerts.txt").split("\n"):

  for line in tf_zip.read('META/apkcerts.txt').decode().split('\n'):

    line = line.strip()

    if not line:

      continue

      if not first:

        print("key file %s still missing some passwords." % (self.pwfile,))

        if sys.version_info[0] >= 3:

          raw_input = input  # pylint: disable=redefined-builtin

        answer = raw_input("try to edit again? [y]> ").strip()

        if answer and answer[0] not in 'yY':

          raise RuntimeError("key passwords unavailable")

  This gives the same result as `openssl x509 -in <filename> -outform DER`.

  Returns:

    The decoded certificate string.

    The decoded certificate bytes.

  """

  cert_buffer = []

  save = False

      cert_buffer.append(line)

    if "--BEGIN CERTIFICATE--" in line:

      save = True

  cert = "".join(cert_buffer).decode('base64')

  cert = base64.b64decode("".join(cert_buffer))

  return cert

  logger.info("putting script in %s", sh_location)

  output_sink(sh_location, sh)

  output_sink(sh_location, sh.encode())

class DynamicPartitionUpdate(object):

  def test_ExtractPublicKey(self):

    cert = os.path.join(self.testdata_dir, 'testkey.x509.pem')

    pubkey = os.path.join(self.testdata_dir, 'testkey.pubkey.pem')

    with open(pubkey, 'rb') as pubkey_fp:

    with open(pubkey) as pubkey_fp:

      self.assertEqual(pubkey_fp.read(), common.ExtractPublicKey(cert))

  def test_ExtractPublicKey_invalidInput(self):

  def test_ExtractAvbPublicKey(self):

    privkey = os.path.join(self.testdata_dir, 'testkey.key')

    pubkey = os.path.join(self.testdata_dir, 'testkey.pubkey.pem')

    with open(common.ExtractAvbPublicKey(privkey)) as privkey_fp, \

        open(common.ExtractAvbPublicKey(pubkey)) as pubkey_fp:

    with open(common.ExtractAvbPublicKey(privkey), 'rb') as privkey_fp, \

        open(common.ExtractAvbPublicKey(pubkey), 'rb') as pubkey_fp:

      self.assertEqual(privkey_fp.read(), pubkey_fp.read())

  def test_ParseCertificate(self):

    cert = os.path.join(self.testdata_dir, 'testkey.x509.pem')

    cmd = ['openssl', 'x509', '-in', cert, '-outform', 'DER']

    proc = common.Run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    proc = common.Run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

                      universal_newlines=False)

    expected, _ = proc.communicate()

    self.assertEqual(0, proc.returncode)

    target_files = common.MakeTempFile(prefix='target_files-', suffix='.zip')

    with zipfile.ZipFile(target_files, 'w') as target_files_zip:

      info_values = ''.join(

          ['{}={}\n'.format(k, v) for k, v in sorted(info_dict.iteritems())])

          ['{}={}\n'.format(k, v) for k, v in sorted(info_dict.items())])

      common.ZipWriteStr(target_files_zip, 'META/misc_info.txt', info_values)

      FSTAB_TEMPLATE = "/dev/block/system {} ext4 ro,barrier=1 defaults"

    loc = os.path.join(self._tempdir, prefix, name)

    if not os.path.exists(os.path.dirname(loc)):

      os.makedirs(os.path.dirname(loc))

    with open(loc, "w+") as f:

    with open(loc, "wb") as f:

      f.write(data)

  def test_full_recovery(self):

    validate_target_files.ValidateInstallRecoveryScript(self._tempdir,

                                                        self._info)

    # Validate 'recovery-from-boot' with bonus argument.

    self._out_tmp_sink("etc/recovery-resource.dat", "bonus", "SYSTEM")

    self._out_tmp_sink("etc/recovery-resource.dat", b"bonus", "SYSTEM")

    common.MakeRecoveryPatch(self._tempdir, self._out_tmp_sink,

                             recovery_image, boot_image, self._info)

    validate_target_files.ValidateInstallRecoveryScript(self._tempdir,

class MockScriptWriter(object):

  """A class that mocks edify_generator.EdifyGenerator.

  """

  """A class that mocks edify_generator.EdifyGenerator."""

  def __init__(self, enable_comments=False):

    self.lines = []

    self.enable_comments = enable_comments

  def Comment(self, comment):

    if self.enable_comments:

      self.lines.append("# {}".format(comment))

      self.lines.append('# {}'.format(comment))

  def AppendExtra(self, extra):

    self.lines.append(extra)

  def __str__(self):

    return "\n".join(self.lines)

    return '\n'.join(self.lines)

class MockBlockDifference(object):

  def __init__(self, partition, tgt, src=None):

    self.partition = partition

    self.tgt = tgt

    self.src = src

  def WriteScript(self, script, _, progress=None,

                  write_verify_script=False):

    if progress:

    script.AppendExtra("patch({});".format(self.partition))

    if write_verify_script:

      self.WritePostInstallVerifyScript(script)

  def WritePostInstallVerifyScript(self, script):

    script.AppendExtra("verify({});".format(self.partition))

class FakeSparseImage(object):

  def __init__(self, size):

    self.blocksize = 4096

    self.total_blocks = size // 4096

class DynamicPartitionsDifferenceTest(test_utils.ReleaseToolsTestCase):

  @staticmethod

  def get_op_list(output_path):

    with zipfile.ZipFile(output_path) as output_zip:

      with output_zip.open("dynamic_partitions_op_list") as op_list:

        return [line.strip() for line in op_list.readlines()

                if not line.startswith("#")]

      with output_zip.open('dynamic_partitions_op_list') as op_list:

        return [line.decode().strip() for line in op_list.readlines()

                if not line.startswith(b'#')]

  def setUp(self):

    self.script = MockScriptWriter()

  """Constructs and returns a File object. Rounds up its size if needed."""

  assert os.path.exists(unpacked_name)

  with open(unpacked_name, 'r') as f:

  with open(unpacked_name, 'rb') as f:

    file_data = f.read()

  file_size = len(file_data)

  if round_up: