Go to the documentation of this file.00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018 import sys, time
00019 sys.path.insert(1,"../")
00020
00021 import unittest
00022 import threading
00023 from Guard import *
00024 from Task import *
00025
00026 ShareCount = 0
00027
00028 class TestGuardTask(Task):
00029 def __init__(self):
00030 global ShareCount
00031 self.ShareCount = 1
00032 Task.__init__(self)
00033 self.flag = False
00034 self._mutex = threading.RLock()
00035
00036 def svc(self):
00037 global ShareCount
00038 guard = ScopedLock(self._mutex)
00039 for i in range(100):
00040 ShareCount += 1
00041 time.sleep(0.1)
00042 self.ShareCount = ShareCount
00043 return
00044
00045 def getShareCount(self):
00046 return self.ShareCount
00047
00048 class TestGuard(unittest.TestCase):
00049 def setUp(self):
00050 return
00051
00052 def test_case0(self):
00053 task1 = TestGuardTask()
00054 task2 = TestGuardTask()
00055 task1.activate()
00056 task2.activate()
00057 task1.wait()
00058 task2.wait()
00059 count = task1.getShareCount()
00060 self.assertEqual(count, 200)
00061 count = task2.getShareCount()
00062 self.assertEqual(count, 200)
00063 task2.finalize()
00064 task1.finalize()
00065 return
00066
00067
00068
00069
00070 if __name__ == '__main__':
00071 unittest.main()