mbligh | be630eb | 2008-08-01 16:41:48 +0000 | [diff] [blame] | 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 | |
mbligh | be630eb | 2008-08-01 16:41:48 +0000 | [diff] [blame] | 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() |