[Android] Support generic parallel execution across devices.

build/android/pylib/device/device_utils_test.py

Index: build/android/pylib/device/device_utils_test.py
diff --git a/build/android/pylib/device/device_utils_test.py b/build/android/pylib/device/device_utils_test.py
index 0aff7ec9358d84707a554abc0dc3ae29d29003d1..03db9b80eda81ef4acb3274940eb4c776f2f068e 100644
--- a/build/android/pylib/device/device_utils_test.py
+++ b/build/android/pylib/device/device_utils_test.py
@@ -9,13 +9,72 @@ Unit tests for the contents of device_utils.py (mostly DeviceUtils).
 # pylint: disable=W0212
 # pylint: disable=W0613
 
+import os
+import tempfile
+import time
 import unittest
+
 from pylib import android_commands
 from pylib import cmd_helper
 from pylib.device import adb_wrapper
 from pylib.device import device_utils
 
 
+class _ParallelTestObject(object):
+  """ Class used to test device_utils._ParallelExecutor.
+
+  Note that this has to live at top-level so it can be pickled.

[craigdh, 2014/05/16 21:59:35]

out of date?

[jbudorick, 2014/05/21 16:42:48]

Done.

+  """
+
+  parallel = device_utils._ParallelExecutor
+
+  def __init__(self, thing, completion_file_name=None):
+    self._thing = thing
+    self._completion_file_name = completion_file_name
+    self.helper = _ParallelTestObjectHelper(thing)
+
+  @staticmethod
+  def doReturn(what):
+    return what
+
+  @classmethod
+  def doRaise(cls, what):
+    raise what
+
+  def doReturnTheThing(self):
+    return self._thing
+
+  def doRaiseTheThing(self):
+    raise self._thing
+
+  def doRaiseIfExceptionElseSleepFor(self, sleep_duration):
+    if isinstance(self._thing, Exception):
+      raise self._thing
+    time.sleep(sleep_duration)
+    self._write_completion_file()
+    return self._thing
+
+  def _write_completion_file(self):
+    if self._completion_file_name and len(self._completion_file_name):
+      with open(self._completion_file_name, 'w+b') as completion_file:
+        completion_file.write('complete')
+
+  def __getitem__(self, index):
+    return self._thing[index]
+
+  def __str__(self):
+    return type(self).__name__
+
+
+class _ParallelTestObjectHelper(object):
+
+  def __init__(self, thing):
+    self._thing = thing
+
+  def doReturnStringThing(self):
+    return str(self._thing)
+
+
 class DeviceUtilsTest(unittest.TestCase):
   def testGetAVDs(self):
     pass
@@ -63,7 +122,74 @@ class DeviceUtilsTest(unittest.TestCase):
     d = device_utils.DeviceUtils(None)
     self.assertIsNone(d.old_interface.GetDevice())
 
+  def testParallelAllReturn(self):
+    devices = [_ParallelTestObject(True) for _ in xrange(0, 10)]
+    results = _ParallelTestObject.parallel(devices).doReturnTheThing().get(1)
+    self.assertTrue(isinstance(results, list))
+    self.assertEquals(10, len(results))
+    self.assertTrue(all(results))
+
+  def testParallelAllRaise(self):
+    devices = [_ParallelTestObject(Exception('thing %d' % i))
+               for i in xrange(0, 10)]
+    p = _ParallelTestObject.parallel(devices).doRaiseTheThing()
+    with self.assertRaises(Exception):
+      p.get(1)
+
+  def testParallelOneFailOthersComplete(self):
+    parallel_device_count = 10
+    exception_index = 7
+    exception_msg = 'thing %d' % exception_index
+
+    try:
+      completion_files = [tempfile.NamedTemporaryFile(delete=False)
+                          for _ in xrange(0, parallel_device_count)]
+      devices = [
+          _ParallelTestObject(
+              i if i != exception_index else Exception(exception_msg),
+              completion_files[i].name)
+          for i in xrange(0, parallel_device_count)]
+      for f in completion_files:
+        f.close()
+      p = _ParallelTestObject.parallel(devices)
+      with self.assertRaises(Exception) as e:
+        p.doRaiseIfExceptionElseSleepFor(2).get(3)
+      self.assertTrue(exception_msg in str(e.exception))
+      for i in xrange(0, parallel_device_count):
+        with open(completion_files[i].name) as f:
+          if i == exception_index:
+            self.assertEquals('', f.read())
+          else:
+            self.assertEquals('complete', f.read())
+    finally:
+      for f in completion_files:
+        os.remove(f.name)
+
+  def testParallelReusable(self):
+    devices = [_ParallelTestObject(True) for _ in xrange(0, 10)]
+    p = _ParallelTestObject.parallel(devices)
+    results = p.doReturn(True).get(1)
+    self.assertTrue(all(results))
+    results = p.doReturn(True).get(1)
+    self.assertTrue(all(results))
+    with self.assertRaises(Exception):
+      results = p.doRaise(Exception('reusableTest')).get(1)
+
+  def testParallelContained(self):
+    devices = [_ParallelTestObject(i) for i in xrange(0, 10)]
+    results = (_ParallelTestObject.parallel(devices).helper
+        .doReturnStringThing().get(1))
+    self.assertTrue(isinstance(results, list))
+    self.assertEquals(10, len(results))
+    for i in xrange(0, 10):
+      self.assertEquals(str(i), results[i])
+
+  def testParallelGetItem(self):
+    devices = [_ParallelTestObject(range(i, i+10)) for i in xrange(0, 10)]
+    results = _ParallelTestObject.parallel(devices)[9].get(1)
+    self.assertEquals(range(9, 19), results)
+
 
 if __name__ == '__main__':
-  unittest.main()
+  unittest.main(verbosity=2)