[Android] Add sharding for AMP instrumentation tests.

build/android/pylib/remote/device/remote_device_test_run.py

 # Copyright 2014 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.
 
 """Run specific test on specific environment."""
 
 import json
 import logging
 import os
+import random
 import re
 import shutil
 import string
 import tempfile
 import time
 import zipfile
 
+from devil.utils import parallelizer
 from devil.utils import zip_utils
 from pylib.base import base_test_result
 from pylib.base import test_run
-from pylib.remote.device import appurify_constants
 from pylib.remote.device import appurify_sanitized
 from pylib.remote.device import remote_device_helper
 
 _DEVICE_OFFLINE_RE = re.compile('error: device not found')
 _LONG_MSG_RE = re.compile('longMsg=(.*)$')
 _SHORT_MSG_RE = re.compile('shortMsg=(.*)$')
 
 class RemoteDeviceTestRun(test_run.TestRun):
   """Run tests on a remote device."""
 
-  _TEST_RUN_KEY = 'test_run'
-  _TEST_RUN_ID_KEY = 'test_run_id'
-
-  WAIT_TIME = 5
+  WAIT_TIME = 10
   COMPLETE = 'complete'
   HEARTBEAT_INTERVAL = 300
 
+  _TEST_RUN_KEY = 'test_run'
+  _TEST_RUN_IDS_KEY = 'test_run_ids'
+
   def __init__(self, env, test_instance):
     """Constructor.
 
     Args:
       env: Environment the tests will run in.
       test_instance: The test that will be run.
     """
     super(RemoteDeviceTestRun, self).__init__(env, test_instance)
+    self._app_id = None
+    self._appurify_configs = None
+    self._base_tempfile_dir = ''
+    self._device_ids = None
     self._env = env
     self._test_instance = test_instance
-    self._app_id = ''
-    self._test_id = ''
-    self._results = ''
-    self._test_run_id = ''
-    self._results_temp_dir = None
+    self._test_package = None
+    self._shard_framework_configs = None
+    self._test_run_ids = []
+
+  # pylint: disable=no-self-use
+  def _GetAdditionalApks(self):
+    return NotImplementedError
+
+  def _GetAppPath(self):
+    raise NotImplementedError
+
+  def _GetTestPath(self):
+    raise NotImplementedError
+
+  def _GetTestRunnerName(self):
+    raise NotImplementedError
+
+  # pylint: disable=no-self-use
+  def _GetTestFramework(self):
+    raise NotImplementedError
+
+  # pylint: disable=no-self-use
+  def _GetFrameworkConfigs(self):
+    return NotImplementedError
+
+  # pylint: disable=unused-argument
+  def _GetShardEnvironmentVars(self, num_shards):
+    return NotImplementedError
 
   #override
   def SetUp(self):
     """Set up a test run."""
+    self._base_tempfile_dir = tempfile.mkdtemp()
     if self._env.trigger:
       self._TriggerSetUp()
     elif self._env.collect:
       assert isinstance(self._env.collect, basestring), (
                         'File for storing test_run_id must be a string.')
       with open(self._env.collect, 'r') as persisted_data_file:
         persisted_data = json.loads(persisted_data_file.read())
-        self._env.LoadFrom(persisted_data)
-        self.LoadFrom(persisted_data)
+        self._LoadFrom(persisted_data)
+
+  def _ShouldShard(self):
+    raise NotImplementedError
 
   def _TriggerSetUp(self):
     """Set up the triggering of a test run."""
-    raise NotImplementedError
+    self._app_id = self._UploadAppToDevice(self._GetAppPath())
 
-  #override
+    if self._ShouldShard():
+      num_shards = min(self._env.num_shards, len(self._env.device_ids))
+      # TODO(mikecase): Change to always run with the number of shards passed.
+      # Will need some shards to wait for new devices to become available.
+      if num_shards < self._env.num_shards:
+        logging.warning(
+            'Requested to shard across %s devices, only %s devices availible.',
+            self._env.num_shards, num_shards)
+    else:
+      num_shards = 1
+    logging.critical('Sharding test run across %s device(s).', num_shards)

[jbudorick, 2015/10/26 22:23:30]

This is info, not critical.

[mikecase (-- gone --), 2015/10/27 01:27:43]

Done.

+    self._device_ids = random.sample(self._env.device_ids, num_shards)
+
+    shard_env_vars = (self._GetShardEnvironmentVars(num_shards)

[jbudorick, 2015/10/26 22:23:30]

I don't think this is structured properly. Most of the environment variables
should be common across all shards. The only one that shouldn't be is the list
of tests. That should be handled separately from the common environment
variables.

[mikecase (-- gone --), 2015/10/27 01:27:43]

Yeah, but it even gets messier if you want some of the environment variables to
be shard dependent and some to be general. Im very open to suggestions, I just
couldn't think of a better way to do it.

+                      or [{} for _ in range(num_shards)])

[jbudorick, 2015/10/26 22:23:30]

:/ this is ugly. Maybe [{}] * num_shards

also: when you think range, think xrange

[mikecase (-- gone --), 2015/10/27 01:27:43]

Done.

+    framework_configs = self._GetFrameworkConfigs() or {}
+
+    if self._GetTestPath():
+      self._test_package, test_package_configs = self._PackageTest(
+          test_path=self._GetTestPath(),
+          extra_apks=self._GetAdditionalApks())
+    else:
+      test_package_configs = {}
+    framework_configs.update(test_package_configs)
+
+    self._appurify_configs = {
+      'network': self._env.network_config,
+      'pcap': self._env.pcap_config,
+      'profiler': self._env.profiler_config,
+      'videocapture': self._env.videocapture_config,
+    }
+
+    self._shard_framework_configs = []
+    for env_vars in shard_env_vars:
+      shard_framework_config = dict(framework_configs)
+      shard_framework_config['environment_vars'] = ','.join(
+          '%s=%s' % (k, v) for k, v in env_vars.iteritems())
+      self._shard_framework_configs.append(shard_framework_config)
+
+  def _Trigger(self, device_id, test_id):
+    with appurify_sanitized.SanitizeLogging(
+        verbose_count=self._env.verbose_count,
+        level=logging.WARNING):
+      test_start_response = appurify_sanitized.api.tests_run(
+          access_token=self._env.token,
+          device_type_id=device_id,
+          app_id=self._app_id,
+          test_id=test_id)
+    remote_device_helper.TestHttpResponse(
+        response=test_start_response,
+        error_msg='Unable to run test.')
+    return test_start_response.json()['response']['test_run_id']
+
   def RunTests(self):
     """Run the test."""
+
+    def trigger(trigger_args):

[jbudorick, 2015/10/26 22:23:30]

these should be defined in their respective if blocks.

+      device_id = trigger_args[0]
+      configs = trigger_args[1]
+      test_id = self._UploadTestToDevice()
+      self._UploadTestConfigToDevice(
+          test_id=test_id,
+          framework_configs=configs,
+          appurify_configs=self._appurify_configs)
+      test_run_id = self._Trigger(
+          device_id=device_id,
+          test_id=test_id)
+      self._test_run_ids.append(test_run_id)
+
+    def collect(test_run_id, parsed_results):
+      download_results_file = tempfile.NamedTemporaryFile(
+          suffix='.zip',
+          dir=self._base_tempfile_dir,
+          delete=False)
+      test_output = self._Collect(test_run_id=test_run_id)
+      self._DownloadTestResults(test_output, download_results_file.name)
+      parsed_results.AddTestRunResults(self._ParseTestResults(
+          test_output=test_output,
+          results_zip=download_results_file.name))
+
     if self._env.trigger:
-      with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
-                                              logging.WARNING):
-        test_start_res = appurify_sanitized.api.tests_run(
-            self._env.token, self._env.device_type_id, self._app_id,
-            self._test_id)
-      remote_device_helper.TestHttpResponse(
-        test_start_res, 'Unable to run test.')
-      self._test_run_id = test_start_res.json()['response']['test_run_id']
-      logging.info('Test run id: %s', self._test_run_id)
-
+      parallelizer.SyncParallelizer(
+          zip(self._device_ids, self._shard_framework_configs)).pMap(trigger)

[jbudorick, 2015/10/26 22:23:30]

O_O

I... I don't know how I feel about this.

[mikecase (-- gone --), 2015/10/27 01:27:43]

Ack

     if self._env.collect:
-      current_status = ''
-      timeout_counter = 0
-      heartbeat_counter = 0
-      while self._GetTestStatus(self._test_run_id) != self.COMPLETE:
-        if self._results['detailed_status'] != current_status:
-          logging.info('Test status: %s', self._results['detailed_status'])
-          current_status = self._results['detailed_status']
-          timeout_counter = 0
-          heartbeat_counter = 0
-        if heartbeat_counter > self.HEARTBEAT_INTERVAL:
-          logging.info('Test status: %s', self._results['detailed_status'])
-          heartbeat_counter = 0
-
-        timeout = self._env.timeouts.get(
-            current_status, self._env.timeouts['unknown'])
-        if timeout_counter > timeout:
-          raise remote_device_helper.RemoteDeviceError(
-              'Timeout while in %s state for %s seconds'
-              % (current_status, timeout),
-              is_infra_error=True)
-        time.sleep(self.WAIT_TIME)
-        timeout_counter += self.WAIT_TIME
-        heartbeat_counter += self.WAIT_TIME
-      self._DownloadTestResults(self._env.results_path)
-
-      if self._results['results']['exception']:
-        raise remote_device_helper.RemoteDeviceError(
-            self._results['results']['exception'], is_infra_error=True)
-
-      return self._ParseTestResults()
+      parsed_results = base_test_result.TestRunResults()
+      parallelizer.SyncParallelizer(
+          self._test_run_ids).pMap(collect, parsed_results)

[jbudorick, 2015/10/26 22:23:30]

or, unsurprisingly, this.

[mikecase (-- gone --), 2015/10/27 01:27:43]

Ack

+      return parsed_results
 
   #override
   def TearDown(self):
     """Tear down the test run."""
+    if self._base_tempfile_dir:
+      shutil.rmtree(self._base_tempfile_dir)
+
     if self._env.collect:
       self._CollectTearDown()
     elif self._env.trigger:
       assert isinstance(self._env.trigger, basestring), (
                         'File for storing test_run_id must be a string.')
       with open(self._env.trigger, 'w') as persisted_data_file:
         persisted_data = {}
-        self.DumpTo(persisted_data)
-        self._env.DumpTo(persisted_data)
+        self._DumpTo(persisted_data)
         persisted_data_file.write(json.dumps(persisted_data))
 
+  def _Collect(self, test_run_id):
+    current_status = ''
+    timeout_counter = 0
+    heartbeat_counter = 0
+    results = self._CheckTestResults(test_run_id)
+    while results['status'] != self.COMPLETE:
+      if results['detailed_status'] != current_status:
+        logging.info('Test status: %s', results['detailed_status'])
+        current_status = results['detailed_status']
+        timeout_counter = 0
+        heartbeat_counter = 0
+      if heartbeat_counter > self.HEARTBEAT_INTERVAL:
+        logging.info('Test status: %s', results['detailed_status'])
+        heartbeat_counter = 0
+
+      timeout = self._env.timeouts.get(
+          current_status, self._env.timeouts['unknown'])
+      if timeout_counter > timeout:
+        raise remote_device_helper.RemoteDeviceError(
+            message='Timeout while in %s state for %s seconds'
+                    % (current_status, timeout),
+            is_infra_error=True)
+      time.sleep(self.WAIT_TIME)
+      timeout_counter += self.WAIT_TIME
+      heartbeat_counter += self.WAIT_TIME
+      results = self._CheckTestResults(test_run_id)
+    if results['results']['exception']:
+      raise remote_device_helper.RemoteDeviceError(
+          message=results['results']['exception'],
+          is_infra_error=True)
+    return results
+
   def _CollectTearDown(self):
-    if self._GetTestStatus(self._test_run_id) != self.COMPLETE:
-      with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
-                                              logging.WARNING):
-        test_abort_res = appurify_sanitized.api.tests_abort(
-            self._env.token, self._test_run_id, reason='Test runner exiting.')
-      remote_device_helper.TestHttpResponse(test_abort_res,
-                                            'Unable to abort test.')
-    if self._results_temp_dir:
-      shutil.rmtree(self._results_temp_dir)
+    def abort_test_run(test_run_id):

[rnephew (Reviews Here), 2015/10/26 20:42:18]

Is there a reason you can't just pass self._AbortTestRun directly into
parallelizer? 

[mikecase (-- gone --), 2015/10/27 01:27:43]

Hmmmmmmmmmm, it seems to work your way. I assumed you couldn't do that. Because
I assumed the "self" argument would get lost if I just grabbed the
self._AbortTestRun function and passed it to the parallelizer. This doesn't seem
to be the case.

Done

+      self._AbortTestRun(test_run_id)
+    if self._test_run_ids:
+      parallelizer.SyncParallelizer(self._test_run_ids).pMap(abort_test_run)
 
   def __enter__(self):
     """Set up the test run when used as a context manager."""
     self.SetUp()
     return self
 
   def __exit__(self, exc_type, exc_val, exc_tb):
     """Tear down the test run when used as a context manager."""
     self.TearDown()
 
-  def DumpTo(self, persisted_data):
-    test_run_data = {
-      self._TEST_RUN_ID_KEY: self._test_run_id,
-    }
+  def _DumpTo(self, persisted_data):
+    """Dump test run information to dict."""
+    test_run_data = {self._TEST_RUN_IDS_KEY: self._test_run_ids}
     persisted_data[self._TEST_RUN_KEY] = test_run_data
 
-  def LoadFrom(self, persisted_data):
+  def _LoadFrom(self, persisted_data):
+    """Load test run information."""
     test_run_data = persisted_data[self._TEST_RUN_KEY]
-    self._test_run_id = test_run_data[self._TEST_RUN_ID_KEY]
+    self._test_run_ids = test_run_data[self._TEST_RUN_IDS_KEY]
 
-  def _ParseTestResults(self):
+  def _ParseTestResults(self, test_output, results_zip):
     raise NotImplementedError
 
-  def _GetTestByName(self, test_name):
-    """Gets test_id for specific test.
-
-    Args:
-      test_name: Test to find the ID of.
-    """
-    with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
-                                            logging.WARNING):
-      test_list_res = appurify_sanitized.api.tests_list(self._env.token)
-    remote_device_helper.TestHttpResponse(test_list_res,
-                                          'Unable to get tests list.')
-    for test in test_list_res.json()['response']:
-      if test['test_type'] == test_name:
-        return test['test_id']
-    raise remote_device_helper.RemoteDeviceError(
-        'No test found with name %s' % (test_name))
-
-  def _DownloadTestResults(self, results_path):
+  def _DownloadTestResults(self, results, download_path):
     """Download the test results from remote device service.
 
-    Downloads results in temporary location, and then copys results
-    to results_path if results_path is not set to None.
-
-    Args:
-      results_path: Path to download appurify results zipfile.
-
     Returns:
       Path to downloaded file.
     """
+    logging.info('Downloading results to %s.', download_path)
+    with appurify_sanitized.SanitizeLogging(
+        verbose_count=self._env.verbose_count,
+        level=logging.WARNING):
+      appurify_sanitized.utils.wget(
+          url=results['results']['url'],
+          path=download_path)
 
-    if self._results_temp_dir is None:
-      self._results_temp_dir = tempfile.mkdtemp()
-      logging.info('Downloading results to %s.', self._results_temp_dir)
-      with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
-                                              logging.WARNING):
-        appurify_sanitized.utils.wget(self._results['results']['url'],
-                                      self._results_temp_dir + '/results')
-    if results_path:
-      logging.info('Copying results to %s', results_path)
-      if not os.path.exists(os.path.dirname(results_path)):
-        os.makedirs(os.path.dirname(results_path))
-      shutil.copy(self._results_temp_dir + '/results', results_path)
-    return self._results_temp_dir + '/results'
-
-  def _GetTestStatus(self, test_run_id):
-    """Checks the state of the test, and sets self._results
+  def _CheckTestResults(self, test_run_id):
+    """Checks the state of the test.
 
     Args:
       test_run_id: Id of test on on remote service.
     """
+    with appurify_sanitized.SanitizeLogging(
+        verbose_count=self._env.verbose_count,
+        level=logging.WARNING):
+      test_check_response = appurify_sanitized.api.tests_check_result(
+          access_token=self._env.token,
+          test_run_id=test_run_id)
+    remote_device_helper.TestHttpResponse(
+        response=test_check_response,
+        error_msg='Unable to get test status.')
+    return test_check_response.json()['response']
 
-    with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
-                                            logging.WARNING):
-      test_check_res = appurify_sanitized.api.tests_check_result(
-          self._env.token, test_run_id)
-    remote_device_helper.TestHttpResponse(test_check_res,
-                                          'Unable to get test status.')
-    self._results = test_check_res.json()['response']
-    return self._results['status']
+  def _PackageTest(self, test_path, extra_apks):
+    """Packages the test and dependencies into a zip.
 
-  def _AmInstrumentTestSetup(self, app_path, test_path, runner_package,
-                             environment_variables, extra_apks=None):
-    config = {'runner': runner_package}
-    if environment_variables:
-      config['environment_vars'] = ','.join(
-          '%s=%s' % (k, v) for k, v in environment_variables.iteritems())
-
-    self._app_id = self._UploadAppToDevice(app_path)
-
+    Returns:
+      A tuple of the path to the packaged test and a dictionary containing
+      configs about how the test was packaged.
+    """
+    package_configs = {}
     data_deps = self._test_instance.GetDataDependencies()
     if data_deps:
-      with tempfile.NamedTemporaryFile(suffix='.zip') as test_with_deps:
-        sdcard_files = []
-        additional_apks = []
-        host_test = os.path.basename(test_path)
-        with zipfile.ZipFile(test_with_deps.name, 'w') as zip_file:
-          zip_file.write(test_path, host_test, zipfile.ZIP_DEFLATED)
-          for h, _ in data_deps:
-            if os.path.isdir(h):
-              zip_utils.WriteToZipFile(zip_file, h, '.')
-              sdcard_files.extend(os.listdir(h))
-            else:
-              zip_utils.WriteToZipFile(zip_file, h, os.path.basename(h))
-              sdcard_files.append(os.path.basename(h))
-          for a in extra_apks or ():
-            zip_utils.WriteToZipFile(zip_file, a, os.path.basename(a))
-            additional_apks.append(os.path.basename(a))
+      test_package = tempfile.NamedTemporaryFile(
+          suffix='.zip',
+          delete=False,
+          dir=self._base_tempfile_dir)

[jbudorick, 2015/10/26 22:23:30]

oh. This is how you're deleting the temporary files.

I would still vastly prefer a context manager.

[mikecase (-- gone --), 2015/10/27 01:27:43]

Ack. All my temporary files are put into _base_tempfile_dir. This is created in
Setup() and deleted in TearDown(). As I said in some earlier comments, a context
manager with #(num_shard) number of tempfiles is super awkward.

+      sdcard_files = []
+      additional_apks = []
+      host_test = os.path.basename(test_path)
+      with zipfile.ZipFile(test_package.name, 'w') as zip_file:
+        zip_file.write(self._GetTestPath(), host_test, zipfile.ZIP_DEFLATED)
+        for h, _ in data_deps:
+          if os.path.isdir(h):
+            zip_utils.WriteToZipFile(zip_file, h, '.')
+            sdcard_files.extend(os.listdir(h))
+          else:
+            zip_utils.WriteToZipFile(zip_file, h, os.path.basename(h))
+            sdcard_files.append(os.path.basename(h))
+        for a in extra_apks or ():
+          zip_utils.WriteToZipFile(zip_file, a, os.path.basename(a))
+          additional_apks.append(os.path.basename(a))
 
-        config['sdcard_files'] = ','.join(sdcard_files)
-        config['host_test'] = host_test
-        if additional_apks:
-          config['additional_apks'] = ','.join(additional_apks)
-        self._test_id = self._UploadTestToDevice(
-            'robotium', test_with_deps.name, app_id=self._app_id)
+      package_configs['sdcard_files'] = ','.join(sdcard_files)
+      package_configs['host_test'] = host_test
+      if additional_apks:
+        package_configs['additional_apks'] = ','.join(additional_apks)
+      return (test_package.name, package_configs)
     else:
-      self._test_id = self._UploadTestToDevice('robotium', test_path)
-
-    logging.info('Setting config: %s', config)
-    appurify_configs = {}
-    if self._env.network_config:
-      appurify_configs['network'] = self._env.network_config
-    self._SetTestConfig('robotium', config, **appurify_configs)
+      return (self._GetTestPath(), package_configs)
 
   def _UploadAppToDevice(self, app_path):
     """Upload app to device."""
     logging.info('Uploading %s to remote service as %s.', app_path,
                  self._test_instance.suite)
     with open(app_path, 'rb') as apk_src:
       with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
                                               logging.WARNING):
         upload_results = appurify_sanitized.api.apps_upload(
             self._env.token, apk_src, 'raw', name=self._test_instance.suite)
       remote_device_helper.TestHttpResponse(
           upload_results, 'Unable to upload %s.' % app_path)
       return upload_results.json()['response']['app_id']
 
-  def _UploadTestToDevice(self, test_type, test_path, app_id=None):
-    """Upload test to device
-    Args:
-      test_type: Type of test that is being uploaded. Ex. uirobot, gtest..
-    """
-    logging.info('Uploading %s to remote service.', test_path)
-    with open(test_path, 'rb') as test_src:
-      with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
-                                              logging.WARNING):
-        upload_results = appurify_sanitized.api.tests_upload(
-            self._env.token, test_src, 'raw', test_type, app_id=app_id)
-      remote_device_helper.TestHttpResponse(upload_results,
-          'Unable to upload %s.' % test_path)
-      return upload_results.json()['response']['test_id']
+  def _UploadTestToDevice(self):
+    """Upload test to device."""
+    logging.info('Shard uploading %s to remote service.', self._test_package)
+    with open(self._test_package, 'rb') as test_src:
+      with appurify_sanitized.SanitizeLogging(
+          verbose_count=self._env.verbose_count,
+          level=logging.WARNING):
+        upload_results_response = appurify_sanitized.api.tests_upload(
+            access_token=self._env.token,

[jbudorick, 2015/10/26 22:23:30]

You sure added a lot of kwarg labels. Are they really necessary...?

[mikecase (-- gone --), 2015/10/27 01:27:43]

Added them because it made reading the code far easier.
But seeing as you don't like them and you're the boss I'll remove (most) of
them...  o(╥﹏╥)o
Leaving kwargs I added for calls to appurify apis.

[jbudorick, 2015/10/27 01:30:32]

to be fair, you were also outvoted 2-1 :P

+            test_source=test_src,
+            test_source_type='raw',
+            test_type=self._GetTestFramework(),
+            app_id=self._app_id)
+      remote_device_helper.TestHttpResponse(
+          response=upload_results_response,
+          error_msg='Unable to upload %s.' % self._test_package)
+      return upload_results_response.json()['response']['test_id']
 
-  def _SetTestConfig(self, runner_type, runner_configs,
-                     network=appurify_constants.NETWORK.WIFI_1_BAR,
-                     pcap=0, profiler=0, videocapture=0):
-    """Generates and uploads config file for test.
-    Args:
-      runner_configs: Configs specific to the runner you are using.
-      network: Config to specify the network environment the devices running
-          the tests will be in.
-      pcap: Option to set the recording the of network traffic from the device.
-      profiler: Option to set the recording of CPU, memory, and network
-          transfer usage in the tests.
-      videocapture: Option to set video capture during the tests.
-
-    """
+  def _UploadTestConfigToDevice(
+      self, test_id, framework_configs, appurify_configs):
+    """Generates and uploads config file for test."""
     logging.info('Generating config file for test.')
+    config_data = ['[appurify]']
+    config_data.extend(
+        '%s=%s' % (k, v) for k, v in appurify_configs.iteritems())
+    config_data.append('[%s]' % self._GetTestFramework())
+    config_data.extend(
+        '%s=%s' % (k, v) for k, v in framework_configs.iteritems())
     with tempfile.TemporaryFile() as config:
-      config_data = [
-          '[appurify]',
-          'network=%s' % network,
-          'pcap=%s' % pcap,
-          'profiler=%s' % profiler,
-          'videocapture=%s' % videocapture,
-          '[%s]' % runner_type
-      ]
-      config_data.extend(
-          '%s=%s' % (k, v) for k, v in runner_configs.iteritems())
       config.write(''.join('%s\n' % l for l in config_data))
       config.flush()
       config.seek(0)
-      with appurify_sanitized.SanitizeLogging(self._env.verbose_count,
-                                              logging.WARNING):
-        config_response = appurify_sanitized.api.config_upload(
-            self._env.token, config, self._test_id)
+      with appurify_sanitized.SanitizeLogging(
+          verbose_count=self._env.verbose_count,
+          level=logging.WARNING):
+        config_upload_response = appurify_sanitized.api.config_upload(
+            access_token=self._env.token,
+            config_src=config,
+            test_id=test_id)
       remote_device_helper.TestHttpResponse(
-          config_response, 'Unable to upload test config.')
+          response=config_upload_response,
+          error_msg='Unable to upload test config.')
 
-  def _LogLogcat(self, level=logging.CRITICAL):
+  def _DetectPlatformErrors(self, results, test_output, results_zip):
+    if not test_output['results']['pass']:
+      crash_msg = None
+      for line in test_output['results']['output'].splitlines():
+        m = _LONG_MSG_RE.search(line)
+        if m:
+          crash_msg = m.group(1)
+          break
+        m = _SHORT_MSG_RE.search(line)
+        if m:
+          crash_msg = m.group(1)
+      if crash_msg:
+        self._LogLogcat(results_zip)
+        results.AddResult(base_test_result.BaseTestResult(
+            crash_msg, base_test_result.ResultType.CRASH))
+      elif self._DidDeviceGoOffline(results_zip):
+        self._LogLogcat(results_zip)
+        self._LogAdbTraceLog(results_zip)
+        raise remote_device_helper.RemoteDeviceError(
+            message='Remote service unable to reach device.',
+            is_infra_error=True)
+      else:
+        # Remote service is reporting a failure, but no failure in results obj.
+        if results.DidRunPass():
+          results.AddResult(base_test_result.BaseTestResult(
+              'Remote service detected error.',
+              base_test_result.ResultType.UNKNOWN))
+
+  def _AbortTestRun(self, test_run_id):
+    if self._CheckTestResults(test_run_id)['status'] != self.COMPLETE:
+      with appurify_sanitized.SanitizeLogging(
+          verbose_count=self._env.verbose_count,
+          level=logging.WARNING):
+        test_abort_response = appurify_sanitized.api.tests_abort(
+            access_token=self._env.token,
+            test_run_id=test_run_id,
+            reason='Test runner exiting.')
+      remote_device_helper.TestHttpResponse(
+          response=test_abort_response,
+          error_msg='Unable to abort test.')
+
+  # pylint: disable=no-self-use
+  def _LogLogcat(self, results_zip, level=logging.CRITICAL):

[jbudorick, 2015/10/26 22:23:30]

methods that don't use self should usually be moved to module-scope private
functions.

[mikecase (-- gone --), 2015/10/27 01:27:43]

Will do. Was kinda unsure since only this class really has any use for these
functions.

     """Prints out logcat downloaded from remote service.
     Args:
+      results_zip: path to zipfile containing the results files.
       level: logging level to print at.
 
     Raises:
       KeyError: If appurify_results/logcat.txt file cannot be found in
                 downloaded zip.
     """
-    zip_file = self._DownloadTestResults(None)
-    with zipfile.ZipFile(zip_file) as z:
+    with zipfile.ZipFile(results_zip) as z:
       try:
         logcat = z.read('appurify_results/logcat.txt')
         printable_logcat = ''.join(c for c in logcat if c in string.printable)
         for line in printable_logcat.splitlines():
           logging.log(level, line)
       except KeyError:
         logging.error('No logcat found.')
 
-  def _LogAdbTraceLog(self):
-    zip_file = self._DownloadTestResults(None)
-    with zipfile.ZipFile(zip_file) as z:
+  # pylint: disable=no-self-use
+  def _LogAdbTraceLog(self, results_zip):
+    with zipfile.ZipFile(results_zip) as z:
       adb_trace_log = z.read('adb_trace.log')
       for line in adb_trace_log.splitlines():
         logging.critical(line)
 
-  def _DidDeviceGoOffline(self):
-    zip_file = self._DownloadTestResults(None)
-    with zipfile.ZipFile(zip_file) as z:
+  # pylint: disable=no-self-use
+  def _DidDeviceGoOffline(self, results_zip):
+    with zipfile.ZipFile(results_zip) as z:
       adb_trace_log = z.read('adb_trace.log')
       if any(_DEVICE_OFFLINE_RE.search(l) for l in adb_trace_log.splitlines()):
         return True
     return False
-
-  def _DetectPlatformErrors(self, results):
-    if not self._results['results']['pass']:
-      crash_msg = None
-      for line in self._results['results']['output'].splitlines():
-        m = _LONG_MSG_RE.search(line)
-        if m:
-          crash_msg = m.group(1)
-          break
-        m = _SHORT_MSG_RE.search(line)
-        if m:
-          crash_msg = m.group(1)
-      if crash_msg:
-        self._LogLogcat()
-        results.AddResult(base_test_result.BaseTestResult(
-            crash_msg, base_test_result.ResultType.CRASH))
-      elif self._DidDeviceGoOffline():
-        self._LogLogcat()
-        self._LogAdbTraceLog()
-        raise remote_device_helper.RemoteDeviceError(
-            'Remote service unable to reach device.', is_infra_error=True)
-      else:
-        # Remote service is reporting a failure, but no failure in results obj.
-        if results.DidRunPass():
-          results.AddResult(base_test_result.BaseTestResult(
-              'Remote service detected error.',
-              base_test_result.ResultType.UNKNOWN))