symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_dummy_threading.py
author johnathan.white@2718R8BGH51.accenture.com
Mon, 08 Mar 2010 18:45:03 +0000
changeset 46 b6935a90ca64
parent 1 2fb8b9db1c86
permissions -rw-r--r--
Modify framebuffer and NGA framebuffer to read screen size from board model dtb file. Optimise memory usuage of frame buffer Add example minigui application with hooks to profiler (which writes results to S:\). Modified NGA framebuffer to run its own dfc queue at high priority

from test import test_support
import unittest
import dummy_threading as _threading
import time

class DummyThreadingTestCase(unittest.TestCase):

    class TestThread(_threading.Thread):

        def run(self):
            global running
            global sema
            global mutex
            # Uncomment if testing another module, such as the real 'threading'
            # module.
            #delay = random.random() * 2
            delay = 0
            if test_support.verbose:
                print 'task', self.name, 'will run for', delay, 'sec'
            sema.acquire()
            mutex.acquire()
            running += 1
            if test_support.verbose:
                print running, 'tasks are running'
            mutex.release()
            time.sleep(delay)
            if test_support.verbose:
                print 'task', self.name, 'done'
            mutex.acquire()
            running -= 1
            if test_support.verbose:
                print self.name, 'is finished.', running, 'tasks are running'
            mutex.release()
            sema.release()

    def setUp(self):
        self.numtasks = 10
        global sema
        sema = _threading.BoundedSemaphore(value=3)
        global mutex
        mutex = _threading.RLock()
        global running
        running = 0
        self.threads = []

    def test_tasks(self):
        for i in range(self.numtasks):
            t = self.TestThread(name="<thread %d>"%i)
            self.threads.append(t)
            t.start()

        if test_support.verbose:
            print 'waiting for all tasks to complete'
        for t in self.threads:
            t.join()
        if test_support.verbose:
            print 'all tasks done'

def test_main():
    test_support.run_unittest(DummyThreadingTestCase)

if __name__ == '__main__':
    test_main()