# Copyright 2015 Google Inc. All rights reserved.
#
# 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.

"""Unit tests for stubify_incremental_install."""

import os
import unittest
import zipfile

from tools.android import incremental_install
from third_party.py import mock


class MockAdb(object):
  """Mocks the Android ADB binary."""

  def __init__(self):
    # Map of file name -> contents.
    self.files = {}
    self._error = None
    self.package_timestamp = None
    self._last_package_timestamp = 0
    self.shell_cmdlns = []

  def Exec(self, args):
    if self._error:
      error_info, arg = self._error  # pylint: disable=unpacking-non-sequence
      if not arg or arg in args:
        return self._CreatePopenMock(*error_info)

    returncode = 0
    stdout = ""
    stderr = ""
    cmd = args[1]
    if cmd == "push":
      # "/test/adb push local remote"
      with open(args[2]) as f:
        content = f.read()
      self.files[args[3]] = content
    elif cmd == "pull":
      # "/test/adb pull remote local"
      remote = args[2]
      local = args[3]
      content = self.files.get(remote)
      if content is not None:
        with open(local, "w") as f:
          f.write(content)
      else:
        returncode = 1
        stderr = "remote object '%s' does not exist\n" % remote
    elif cmd == "install":
      self.package_timestamp = self._last_package_timestamp
      self._last_package_timestamp += 1
      return self._CreatePopenMock(0, "Success", "")
    elif cmd == "shell":
      # "/test/adb shell ..."
      # mkdir, rm, am (application manager), or monkey
      shell_cmdln = args[2]
      self.shell_cmdlns.append(shell_cmdln)
      if shell_cmdln.startswith(("mkdir", "am", "monkey")):
        pass
      elif shell_cmdln.startswith("dumpsys package "):
        return self._CreatePopenMock(
            0,
            "lastUpdateTime=%s" % self.package_timestamp,
            "")
      elif shell_cmdln.startswith("rm"):
        file_path = shell_cmdln.split()[2]
        self.files.pop(file_path, None)
      else:
        raise Exception("Unknown shell command line: %s" % shell_cmdln)
    # Return a mock subprocess.Popen object
    return self._CreatePopenMock(returncode, stdout, stderr)

  def _CreatePopenMock(self, returncode, stdout, stderr):
    return mock.Mock(
        returncode=returncode, communicate=lambda: (stdout, stderr))

  def SetError(self, returncode, stdout, stderr, for_arg=None):
    self._error = ((returncode, stdout, stderr), for_arg)


class IncrementalInstallTest(unittest.TestCase):
  """Unit tests for incremental install."""

  _DEXMANIFEST = "dexmanifest.txt"
  _ADB_PATH = "/test/adb"
  _OUTPUT_MARKER = "full_deploy_marker"
  _APK = "myapp_incremental.apk"
  _RESOURCE_APK = "incremental.ap_"
  _STUB_DATAFILE = "stub_application_data.txt"
  _OLD_APP_PACKGE = "old.app.package"
  _APP_PACKAGE = "new.app.package"
  _EXEC_ROOT = "."

  def setUp(self):
    os.chdir(os.environ["TEST_TMPDIR"])

    self._mock_adb = MockAdb()

    # Write the stub datafile which contains the package name of the app.
    with open(self._STUB_DATAFILE, "w") as f:
      f.write("\n".join([self._OLD_APP_PACKGE, self._APP_PACKAGE]))

    # Write the local resource apk file.
    with open(self._RESOURCE_APK, "w") as f:
      f.write("resource apk")

    # Mock out subprocess.Popen to use our mock adb.
    self._popen_patch = mock.patch.object(incremental_install, "subprocess")
    self._popen = self._popen_patch.start().Popen
    self._popen.side_effect = lambda args, **kwargs: self._mock_adb.Exec(args)

  def tearDown(self):
    self._popen_patch.stop()

  def _CreateZip(self, name="zip1", *files):
    if not files:
      files = [("zp1", "content1"), ("zp2", "content2")]
    with zipfile.ZipFile(name, "w") as z:
      for f, content in files:
        z.writestr(f, content)

  def _CreateLocalManifest(self, *lines):
    content = "\n".join(lines)
    with open(self._DEXMANIFEST, "w") as f:
      f.write(content)
    return content

  def _CreateRemoteManifest(self, *lines):
    self._PutDeviceFile("dex/manifest", "\n".join(lines))

  def _GetDeviceAppPath(self, f):
    return os.path.join(
        incremental_install.DEVICE_DIRECTORY, self._APP_PACKAGE, f)

  def _GetDeviceFile(self, f):
    return self._mock_adb.files[self._GetDeviceAppPath(f)]

  def _PutDeviceFile(self, f, content):
    self._mock_adb.files[self._GetDeviceAppPath(f)] = content

  def _CallIncrementalInstall(self, incremental, start_app=False):
    if incremental:
      apk = None
    else:
      apk = self._APK
    incremental_install.IncrementalInstall(
        adb_path=self._ADB_PATH,
        execroot=self._EXEC_ROOT,
        stub_datafile=self._STUB_DATAFILE,
        dexmanifest=self._DEXMANIFEST,
        apk=apk,
        resource_apk=self._RESOURCE_APK,
        output_marker=self._OUTPUT_MARKER,
        adb_jobs=1,
        start_app=start_app,
        user_home_dir="/home/root")

  def testUploadToPristineDevice(self):

    self._CreateZip()

    with open("dex1", "w") as f:
      f.write("content3")

    manifest = self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zp2 ip2 0",
        "dex1 - ip3 0")

    self._CallIncrementalInstall(incremental=False)

    resources_checksum_path = self._GetDeviceAppPath("resources_checksum")
    self.assertTrue(resources_checksum_path in self._mock_adb.files)
    self.assertEquals(manifest, self._GetDeviceFile("dex/manifest"))
    self.assertEquals("content1", self._GetDeviceFile("dex/ip1"))
    self.assertEquals("content2", self._GetDeviceFile("dex/ip2"))
    self.assertEquals("content3", self._GetDeviceFile("dex/ip3"))
    self.assertEquals("resource apk", self._GetDeviceFile("resources.ap_"))

  def testUploadWithOneChangedFile(self):

    # Existing manifest from a previous install.
    self._CreateRemoteManifest(
        "zip1 zp1 ip1 0",
        "zip1 zp2 ip2 1")

    # Existing files from a previous install.
    self._PutDeviceFile("dex/ip1", "old content1")
    self._PutDeviceFile("dex/ip2", "old content2")
    self._PutDeviceFile("install_timestamp", "0")
    self._mock_adb.package_timestamp = "0"

    self._CreateZip()

    # Updated dex manifest.
    self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zp2 ip2 2")

    self._CallIncrementalInstall(incremental=True)

    # This is a bit of a dishonest test: the local content for "ip1" is
    # "content1" and the remote content for it is "old content1", but
    # the checksums for that file are the same in the local and remote manifest.
    # We just want to make sure that only one file was updated, so to
    # distinguish that we force the local and remote content to be different but
    # keep the checksum the same.
    self.assertEquals("old content1", self._GetDeviceFile("dex/ip1"))
    self.assertEquals("content2", self._GetDeviceFile("dex/ip2"))

  def testFullUploadWithOneChangedFile(self):

    # Existing manifest from a previous install.
    self._CreateRemoteManifest(
        "zip1 zp1 ip1 0",
        "zip1 zp2 ip2 1")

    self._PutDeviceFile("dex/ip1", "old content1")
    self._PutDeviceFile("dex/ip2", "old content2")
    self._PutDeviceFile("install_timestamp", "0")
    self._mock_adb.package_timestamp = "0"

    self._CreateZip()

    self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zp2 ip2 2")

    self._CallIncrementalInstall(incremental=False)

    # Even though the checksums for ip1 were the same, the file still got
    # updated. This is a bit of a dishonest test because the local and remote
    # content for ip1 were different, but their checksums were the same.
    self.assertEquals("content1", self._GetDeviceFile("dex/ip1"))
    self.assertEquals("content2", self._GetDeviceFile("dex/ip2"))

  def testUploadWithNewFile(self):

    self._CreateRemoteManifest("zip1 zp1 ip1 0")
    self._PutDeviceFile("dex/ip1", "content1")
    self._PutDeviceFile("install_timestamp", "0")
    self._mock_adb.package_timestamp = "0"

    self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zp2 ip2 1")

    self._CreateZip()

    self._CallIncrementalInstall(incremental=True)

    self.assertEquals("content1", self._GetDeviceFile("dex/ip1"))
    self.assertEquals("content2", self._GetDeviceFile("dex/ip2"))

  def testDeletesFile(self):

    self._CreateRemoteManifest(
        "zip1 zp1 ip1 0",
        "zip1 zip2 ip2 1")
    self._PutDeviceFile("dex/ip1", "content1")
    self._PutDeviceFile("dex/ip2", "content2")
    self._PutDeviceFile("install_timestamp", "0")
    self._mock_adb.package_timestamp = "0"

    self._CreateZip("zip1", ("zp1", "content1"))
    self._CreateLocalManifest("zip1 zp1 ip1 0")

    self.assertTrue(self._GetDeviceAppPath("dex/ip2") in self._mock_adb.files)
    self._CallIncrementalInstall(incremental=True)
    self.assertFalse(self._GetDeviceAppPath("dex/ip2") in self._mock_adb.files)

  def testNothingToUpdate(self):
    self._CreateRemoteManifest(
        "zip1 zp1 ip1 0",
        "zip1 zip2 ip2 1",
        "dex1 - ip3 0")
    self._PutDeviceFile("dex/ip1", "content1")
    self._PutDeviceFile("dex/ip2", "content2")
    self._PutDeviceFile("dex/ip3", "content3")
    self._PutDeviceFile("install_timestamp", "0")
    self._mock_adb.package_timestamp = "0"

    self._CreateZip()
    self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zip2 ip2 1",
        "dex1 - ip3 0")

    self._CallIncrementalInstall(incremental=True)
    self.assertEquals("content1", self._GetDeviceFile("dex/ip1"))
    self.assertEquals("content2", self._GetDeviceFile("dex/ip2"))
    self.assertEquals("content3", self._GetDeviceFile("dex/ip3"))

  def testNoResourcesToUpdate(self):
    self._CreateRemoteManifest("zip1 zp1 ip1 0")
    self._PutDeviceFile("dex/ip1", "content1")
    # The local file is actually "resource apk", but the checksum on the device
    # for the resources file is set to be the same as the checksum for the
    # local file so that we can confirm that it was not updated.
    self._PutDeviceFile("resources.ap_", "resources")
    self._PutDeviceFile("resources_checksum",
                        incremental_install.Checksum(self._RESOURCE_APK))
    self._PutDeviceFile("install_timestamp", "0")
    self._mock_adb.package_timestamp = "0"

    self._CreateZip()
    self._CreateLocalManifest("zip1 zp1 ip1 0")

    self._CallIncrementalInstall(incremental=True)
    self.assertEquals("resources", self._GetDeviceFile("resources.ap_"))

  def testUpdateResources(self):
    self._CreateRemoteManifest("zip1 zp1 ip1 0")
    self._PutDeviceFile("dex/ip1", "content1")
    self._PutDeviceFile("resources.ap_", "resources")
    self._PutDeviceFile("resources_checksum", "checksum")
    self._PutDeviceFile("install_timestamp", "0")
    self._mock_adb.package_timestamp = "0"

    self._CreateZip()
    self._CreateLocalManifest("zip1 zp1 ip1 0")

    self._CallIncrementalInstall(incremental=True)
    self.assertEquals("resource apk", self._GetDeviceFile("resources.ap_"))

  def testNoDevice(self):
    self._mock_adb.SetError(1, "", "device not found")
    try:
      self._CallIncrementalInstall(incremental=True)
      self.fail("Should have quit if there is no device")
    except SystemExit:
      pass

  def testUnauthorizedDevice(self):
    self._mock_adb.SetError(1, "", "device unauthorized. Please check the "
                            "confirmation dialog on your device")
    try:
      self._CallIncrementalInstall(incremental=True)
      self.fail("Should have quit if the device is unauthorized.")
    except SystemExit:
      pass

  def testInstallFailure(self):
    self._mock_adb.SetError(0, "Failure", "", for_arg="install")
    self._CreateZip()
    self._CreateLocalManifest("zip1 zp1 ip1 0")
    try:
      self._CallIncrementalInstall(incremental=False)
      self.fail("Should have quit if the install failed.")
    except SystemExit:
      pass

  def testStartApp(self):
    # Based on testUploadToPristineDevice
    self._CreateZip()

    with open("dex1", "w") as f:
      f.write("content3")

    self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zp2 ip2 0",
        "dex1 - ip3 0")

    self._CallIncrementalInstall(incremental=False, start_app=True)

    self.assertTrue(("monkey -p %s -c android.intent.category.LAUNCHER 1" %
                     self._APP_PACKAGE) in self._mock_adb.shell_cmdlns)

  def testMultipleDevicesError(self):
    errors = [
        "more than one device and emulator",
        "more than one device",
        "more than one emulator",
    ]
    for error in errors:
      self._mock_adb.SetError(255, "", error)
      try:
        self._CallIncrementalInstall(incremental=True)
        self.fail("Should have quit if there were multiple devices.")
      except SystemExit:
        pass

  def testIncrementalInstallOnPristineDevice(self):
    self._CreateZip()
    self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zip2 ip2 1",
        "dex1 - ip3 0")

    try:
      self._CallIncrementalInstall(incremental=True)
      self.fail("Should have quit for incremental install on pristine device")
    except SystemExit:
      pass

  def testIncrementalInstallWithWrongInstallTimestamp(self):
    self._CreateRemoteManifest(
        "zip1 zp1 ip1 0",
        "zip1 zip2 ip2 1",
        "dex1 - ip3 0")
    self._PutDeviceFile("dex/ip1", "content1")
    self._PutDeviceFile("dex/ip2", "content2")
    self._PutDeviceFile("dex/ip3", "content3")
    self._mock_adb.package_timestamp = "WRONG"

    self._CreateZip()
    self._CreateLocalManifest(
        "zip1 zp1 ip1 0",
        "zip1 zip2 ip2 1",
        "dex1 - ip3 0")

    try:
      self._CallIncrementalInstall(incremental=True)
      self.fail("Should have quit if install timestamp is wrong")
    except SystemExit:
      pass

if __name__ == "__main__":
  unittest.main()