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 android_commands
from pylib import constants
from pylib import forwarder
from pylib.device import device_list
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_list.GetPersistentDeviceList(devices_path)
  except IOError as e:
    logging.error('Unable to find %s [%s]', devices_path, e)
    devices = android_commands.GetAttachedDevices()
  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

# TODO(bulach): remove once it rolls downstream, crbug.com/378862.
def _GetStepsDictFromV0(steps_v0):
  steps_dict = {
    'version': 1,
    'steps': {},
  }
  affinity = 0
  for step in steps_v0:
    steps_dict['steps'][step[0]] = {
        'device_affinity': affinity,
        'cmd': step[1],
    }
    affinity += 1
  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)
      # TODO(bulach): remove once it rolls downstream, crbug.com/378862.
      if isinstance(steps, list):
        return _GetStepsDictFromV0(steps)

      # 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()
  forwarder.Forwarder.UseMultiprocessing()

  # 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)