cli/threads_unittest.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright 2008 Google Inc. All Rights Reserved.
   4
   5 """Tests for thread."""
   6
   7 import unittest, sys, os
   8
   9 import threading, Queue
  10
  11 import common
  12 from autotest_lib.cli import cli_mock, threads
  13
  14
  15 class thread_unittest(cli_mock.cli_unittest):
  16     results = Queue.Queue()
  17
  18     def _workload(self, i):
  19         self.results.put(i*i)
  20
  21
  22     def test_starting(self):
  23         self.god.stub_class_method(threading.Thread, 'start')
  24         threading.Thread.start.expect_call().and_return(None)
  25         threading.Thread.start.expect_call().and_return(None)
  26         threading.Thread.start.expect_call().and_return(None)
  27         threading.Thread.start.expect_call().and_return(None)
  28         threading.Thread.start.expect_call().and_return(None)
  29         th = threads.ThreadPool(self._workload, numthreads=5)
  30         self.god.check_playback()
  31
  32
  33     def test_one_thread(self):
  34         th = threads.ThreadPool(self._workload, numthreads=1)
  35         th.queue_work(range(10))
  36         th.wait()
  37         res = []
  38         while not self.results.empty():
  39             res.append(self.results.get())
  40         self.assertEqualNoOrder([0, 1, 4, 9, 16, 25, 36, 49, 64, 81], res)
  41
  42
  43     def _threading(self, numthreads, count):
  44         th = threads.ThreadPool(self._workload, numthreads=numthreads)
  45         th.queue_work(range(count))
  46         th.wait()
  47         res = []
  48         while not self.results.empty():
  49             res.append(self.results.get())
  50         self.assertEqualNoOrder([i*i for i in xrange(count)], res)
  51
  52
  53     def test_threading(self):
  54         self._threading(10, 10)
  55
  56
  57     def test_threading_lots(self):
  58         self._threading(100, 100)
  59
  60
  61     def test_threading_multi_queueing(self):
  62         th = threads.ThreadPool(self._workload, numthreads=5)
  63         th.queue_work(range(5))
  64         th.queue_work(range(5, 10))
  65         th.wait()
  66         res = []
  67         while not self.results.empty():
  68             res.append(self.results.get())
  69         self.assertEqualNoOrder([i*i for i in xrange(10)], res)
  70
  71
  72 if __name__ == '__main__':
  73     unittest.main()