text: # Copyright 2015 The Chromium Authors. All rights reserved.
  # Use of this source code is governed by a BSD-style license that can be
  # found in the LICENSE file.
  
  import logging
  import time
  
  from pylib import flag_changer
  from pylib.base import base_test_result
  from pylib.base import test_run
  from pylib.local.device import local_device_test_run
  
  
  TIMEOUT_ANNOTATIONS = [
    ('Manual', 10 * 60 * 60),
    ('IntegrationTest', 30 * 60),
    ('External', 10 * 60),
    ('EnormousTest', 10 * 60),
    ('LargeTest', 5 * 60),
    ('MediumTest', 3 * 60),
    ('SmallTest', 1 * 60),
  ]
  
  
  # TODO(jbudorick): Make this private once the instrumentation test_runner is
  # deprecated.
  def DidPackageCrashOnDevice(package_name, device):
    # Dismiss any error dialogs. Limit the number in case we have an error
    # loop or we are failing to dismiss.
    for _ in xrange(10):
      package = device.old_interface.DismissCrashDialogIfNeeded()
      if not package:
        return False
      # Assume test package convention of ".test" suffix
      if package in package_name:
        return True
    return False
  
  
  class LocalDeviceInstrumentationTestRun(
      local_device_test_run.LocalDeviceTestRun):
    def __init__(self, env, test_instance):
      super(LocalDeviceInstrumentationTestRun, self).__init__(env, test_instance)
      self._flag_changers = {}
  
    def TestPackage(self):
      return None
  
    def SetUp(self):
      def substitute_external_storage(d, external_storage):
        if not d:
          return external_storage
        elif isinstance(d, list):
          return '/'.join(p if p else external_storage for p in d)
        else:
          return d
  
      def individual_device_set_up(dev, host_device_tuples):
        dev.Install(self._test_instance.apk_under_test)
        dev.Install(self._test_instance.test_apk)
  
        external_storage = dev.GetExternalStoragePath()
        host_device_tuples = [
            (h, substitute_external_storage(d, external_storage))
            for h, d in host_device_tuples]
        logging.info('instrumentation data deps:')
        for h, d in host_device_tuples:
          logging.info('%r -> %r', h, d)
        dev.PushChangedFiles(host_device_tuples)
        if self._test_instance.flags:
          if not self._test_instance.package_info:
            logging.error("Couldn't set flags: no package info")
          elif not self._test_instance.package_info.cmdline_file:
            logging.error("Couldn't set flags: no cmdline_file")
          else:
            self._flag_changers[str(dev)] = flag_changer.FlagChanger(
                dev, self._test_instance.package_info.cmdline_file)
            logging.debug('Attempting to set flags: %r',
                          self._test_instance.flags)
            self._flag_changers[str(dev)].AddFlags(self._test_instance.flags)
  
      self._env.parallel_devices.pMap(
          individual_device_set_up,
          self._test_instance.GetDataDependencies())
  
    def TearDown(self):
      def individual_device_tear_down(dev):
        if str(dev) in self._flag_changers:
          self._flag_changers[str(dev)].Restore()
  
      self._env.parallel_devices.pMap(individual_device_tear_down)
  
    #override
    def _CreateShards(self, tests):
      return tests
  
    #override
    def _GetTests(self):
      return self._test_instance.GetTests()
  
    #override
    def _GetTestName(self, test):
      return '%s#%s' % (test['class'], test['method'])
  
    #override
    def _RunTest(self, device, test):
      test_name = self._GetTestName(test)
      logging.info('preparing to run %s: %s' % (test_name, test))
  
      extras = {
        'class': test_name,
        'org.chromium.chrome.test.ChromeInstrumentationTestRunner'
            '.EnableTestHttpServer': '',
      }
      timeout = self._GetTimeoutFromAnnotations(test['annotations'], test_name)
  
      time_ms = lambda: int(time.time() * 1e3)
      start_ms = time_ms()
      output = device.StartInstrumentation(
          '%s/%s' % (self._test_instance.test_package,
                     self._test_instance.test_runner),
          raw=True, extras=extras, timeout=timeout, retries=0)
      duration_ms = time_ms() - start_ms
  
      # TODO(jbudorick): Make instrumentation tests output a JSON so this
      # doesn't have to parse the output.
      logging.info('output from %s:' % test_name)
      for l in output:
        logging.info('  %s' % l)
  
      _, _, statuses = self._test_instance.ParseAmInstrumentRawOutput(output)
      result = self._test_instance.GenerateTestResult(
          test_name, statuses, start_ms, duration_ms)
      if DidPackageCrashOnDevice(self._test_instance.test_package, device):
        result.SetType(base_test_result.ResultType.CRASH)
      return result
  
    #override
    def _ShouldShard(self):
      return True
  
    @staticmethod
    def _GetTimeoutFromAnnotations(annotations, test_name):
      for k, v in TIMEOUT_ANNOTATIONS:
        if k in annotations:
          timeout = v
      else:
        logging.warning('Using default 1 minute timeout for %s' % test_name)
        timeout = 60
  
      try:
        scale = int(annotations.get('TimeoutScale', 1))
      except ValueError as e:
        logging.warning("Non-integer value of TimeoutScale ignored. (%s)", str(e))
        scale = 1
      timeout *= scale
  
      return timeout
  
  