build/android/pylib/perf/setup.py

# Copyright 2013 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.

"""Generates test runner factory and tests for performance tests."""

import json
import fnmatch
import logging
import os
import shutil

from pylib import constants
from pylib import forwarder
from pylib.device import device_list
from pylib.device import device_utils
from pylib.perf import test_runner
from pylib.utils import test_environment


def _GetAllDevices():
  devices_path = os.path.join(os.environ.get('CHROMIUM_OUT_DIR', 'out'),
                              device_list.LAST_DEVICES_FILENAME)
  try:
    devices = [device_utils.DeviceUtils(s)
               for s in device_list.GetPersistentDeviceList(devices_path)]
  except IOError as e:
    logging.error('Unable to find %s [%s]', devices_path, e)
    devices = device_utils.DeviceUtils.HealthyDevices()
  return sorted(devices)


def _GetStepsDictFromSingleStep(test_options):
  # Running a single command, build the tests structure.
  steps_dict = {
    'version': 1,
    'steps': {
        'single_step': {
          'device_affinity': 0,
          'cmd': test_options.single_step
        },
    }
  }
  return steps_dict


def _GetStepsDict(test_options):
  if test_options.single_step:
    return _GetStepsDictFromSingleStep(test_options)
  if test_options.steps:
    with file(test_options.steps, 'r') as f:
      steps = json.load(f)

      # Already using the new format.
      assert steps['version'] == 1
      return steps


def Setup(test_options):
  """Create and return the test runner factory and tests.

  Args:
    test_options: A PerformanceOptions object.

  Returns:
    A tuple of (TestRunnerFactory, tests, devices).
  """
  # TODO(bulach): remove this once the bot side lands. BUG=318369
  constants.SetBuildType('Release')
  if os.path.exists(constants.PERF_OUTPUT_DIR):
    shutil.rmtree(constants.PERF_OUTPUT_DIR)
  os.makedirs(constants.PERF_OUTPUT_DIR)

  # Before running the tests, kill any leftover server.
  test_environment.CleanupLeftoverProcesses()

  # We want to keep device affinity, so return all devices ever seen.
  all_devices = _GetAllDevices()

  steps_dict = _GetStepsDict(test_options)
  sorted_step_names = sorted(steps_dict['steps'].keys())

  if test_options.test_filter:
    sorted_step_names = fnmatch.filter(sorted_step_names,
                                       test_options.test_filter)

  flaky_steps = []
  if test_options.flaky_steps:
    with file(test_options.flaky_steps, 'r') as f:
      flaky_steps = json.load(f)

  def TestRunnerFactory(device, shard_index):
    return test_runner.TestRunner(
        test_options, device, shard_index, len(all_devices),
        steps_dict, flaky_steps)

  return (TestRunnerFactory, sorted_step_names, all_devices)