FILEMOVER/test/mapping_manager.py


import os
import sys
import sets
import time
import unittest
import threading
import ConfigParser

if os.path.exists("src"):
    sys.path.append("src")
if os.path.exists('../src'):
    sys.path.append('../src')

from CmsFileServer.MappingManager import MappingManager

class BlockingMappingManager(MappingManager):
    """
    A mapping manager which can be made to block on acquire and release
    (so we can test thread-safety a bit).
    """
    
    def __init__(self):
        super(BlockingMappingManager, self).__init__()
        self._acquire_cond = None
        self._release_cond = None
        self._acquire_count = 0
        self._release_count = 0
        self._acquire_error = None
        
    def blockOnAcquire(self, cond):
        self._acquire_cond = cond
        
    def errorOnAcquire(self, error):
        self._acquire_error = error

    def _lookup(self, key):
        if self._acquire_cond:
            self._acquire_cond.acquire()
            self._acquire_cond.wait()
            self._acquire_cond.release()
        self._acquire_count += 1
        if self._acquire_error:
            raise self._acquire_error
        return key**2

    def blockOnRelease(self, cond):
        self._acquire_cond = cond
        
    def _release(self, key):
        if self._release_cond:
            self._release_cond.acquire()
            self._release_cond.wait()
            self._release_cond.release()
        self._release_count += 1
            
class TestMappingManager(unittest.TestCase):
    """
    A test case for the MappingManager
    """
    
    def setUp(self):
        self.manager = BlockingMappingManager()
    
    def test_acquire_release_serial(self):
        value = self.manager.acquireValue(3)
        self.failUnless(value==9, "Returned the incorrect value.")
        self.manager.releaseKey(3)
        value = self.manager.acquireValue(3)
        self.failUnless(value==9, "Returned the incorrect value.")
        self.manager.releaseKey(3)
        self.failUnless(self.manager._acquire_count == 2)
        self.failUnless(self.manager._release_count == 2)
        
    def test_acquire_release_overlapping(self):
        value = self.manager.acquireValue(3)
        self.failUnless(value==9, "Returned the incorrect value.")
        self.failUnless(self.manager._keys_refcount[3] == 1)
        value = self.manager.acquireValue(3)
        self.failUnless(self.manager._keys_refcount[3] == 2)
        self.manager.releaseKey(3)
        self.failUnless(self.manager._keys_refcount[3] == 1)
        self.failUnless(value==9, "Returned the incorrect value.")
        self.manager.releaseKey(3)
        self.failUnless(3 not in self.manager._keys_refcount)
        self.failUnless(self.manager._acquire_count == 1)
        self.failUnless(self.manager._release_count == 2)
     
    def _acquire_key_3(self):
        self.manager.acquireValue(3)
     
    def test_acquire_release_blocking(self):
        cond = threading.Condition()
        self.manager.blockOnAcquire(cond)
        t1 = threading.Thread(target=self._acquire_key_3)
        t2 = threading.Thread(target=self._acquire_key_3)
        self.failUnless(3 not in self.manager._lookup_attempts)
        self.failUnless(3 not in self.manager._keys_refcount)
        t1.start()
        self.failUnless(3 in self.manager._lookup_attempts)
        self.failUnless(3 not in self.manager._keys_refcount)
        t2.start()
        time.sleep(.2)
        cond.acquire()
        cond.notifyAll()
        cond.release()
        cond.acquire()
        cond.notifyAll()
        cond.release()
        t1.join()
        t2.join()
        self.failUnless(self.manager._acquire_count == 1)
        self.failUnless(self.manager._keys_refcount[3] == 2)
        self.manager.releaseKey(3)
        self.manager.releaseKey(3)
        self.failUnless(3 not in self.manager._keys_refcount)
        
    def test_error(self):
        self.manager.errorOnAcquire(Exception("known error."))
        did_fail = False
        try:
            self.manager.acquireValue()
        except:
            did_fail = True
        self.failUnless(did_fail, msg="Error on acquire failed.")
        self.manager.errorOnAcquire(None)
        self.failUnless(self.manager.acquireValue(3) == 9)
        self.failUnless(self.manager._acquire_count == 1)
        self.failUnless(self.manager._keys_refcount[3] == 1)
        self.manager.releaseKey(3)
        self.failUnless(3 not in self.manager._keys_refcount)

if __name__ == '__main__':
    unittest.main()

Revision:	1.2
Committed:	Wed Jan 14 03:20:57 2009 UTC (16 years, 3 months ago) by bbockelm
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	V01_00_33, V01_00_32, V01_00_31, V01_00_30, V01_00_29, V01_00_28, V01_00_27, V01_00_26, V01_00_25, V01_00_24, V01_00_23, V01_00_22, V01_00_21, V01_00_20, V01_00_19, V01_00_18, V01_00_17, V01_00_16, V01_00_15, V01_00_14, V01_00_13, V01_00_12, HEAD
Changes since 1.1:	+22 -0 lines
Error occurred while calculating annotation data.
Log Message:	Commit of FileMover version used for benchmarks reported to Dima.
#	Content
1
2	import os
3	import sys
4	import sets
5	import time
6	import unittest
7	import threading
8	import ConfigParser
9
10	if os.path.exists("src"):
11	sys.path.append("src")
12	if os.path.exists('../src'):
13	sys.path.append('../src')
14
15	from CmsFileServer.MappingManager import MappingManager
16
17	class BlockingMappingManager(MappingManager):
18	"""
19	A mapping manager which can be made to block on acquire and release
20	(so we can test thread-safety a bit).
21	"""
22
23	def __init__(self):
24	super(BlockingMappingManager, self).__init__()
25	self._acquire_cond = None
26	self._release_cond = None
27	self._acquire_count = 0
28	self._release_count = 0
29	self._acquire_error = None
30
31	def blockOnAcquire(self, cond):
32	self._acquire_cond = cond
33
34	def errorOnAcquire(self, error):
35	self._acquire_error = error
36
37	def _lookup(self, key):
38	if self._acquire_cond:
39	self._acquire_cond.acquire()
40	self._acquire_cond.wait()
41	self._acquire_cond.release()
42	self._acquire_count += 1
43	if self._acquire_error:
44	raise self._acquire_error
45	return key**2
46
47	def blockOnRelease(self, cond):
48	self._acquire_cond = cond
49
50	def _release(self, key):
51	if self._release_cond:
52	self._release_cond.acquire()
53	self._release_cond.wait()
54	self._release_cond.release()
55	self._release_count += 1
56
57	class TestMappingManager(unittest.TestCase):
58	"""
59	A test case for the MappingManager
60	"""
61
62	def setUp(self):
63	self.manager = BlockingMappingManager()
64
65	def test_acquire_release_serial(self):
66	value = self.manager.acquireValue(3)
67	self.failUnless(value==9, "Returned the incorrect value.")
68	self.manager.releaseKey(3)
69	value = self.manager.acquireValue(3)
70	self.failUnless(value==9, "Returned the incorrect value.")
71	self.manager.releaseKey(3)
72	self.failUnless(self.manager._acquire_count == 2)
73	self.failUnless(self.manager._release_count == 2)
74
75	def test_acquire_release_overlapping(self):
76	value = self.manager.acquireValue(3)
77	self.failUnless(value==9, "Returned the incorrect value.")
78	self.failUnless(self.manager._keys_refcount[3] == 1)
79	value = self.manager.acquireValue(3)
80	self.failUnless(self.manager._keys_refcount[3] == 2)
81	self.manager.releaseKey(3)
82	self.failUnless(self.manager._keys_refcount[3] == 1)
83	self.failUnless(value==9, "Returned the incorrect value.")
84	self.manager.releaseKey(3)
85	self.failUnless(3 not in self.manager._keys_refcount)
86	self.failUnless(self.manager._acquire_count == 1)
87	self.failUnless(self.manager._release_count == 2)
88
89	def _acquire_key_3(self):
90	self.manager.acquireValue(3)
91
92	def test_acquire_release_blocking(self):
93	cond = threading.Condition()
94	self.manager.blockOnAcquire(cond)
95	t1 = threading.Thread(target=self._acquire_key_3)
96	t2 = threading.Thread(target=self._acquire_key_3)
97	self.failUnless(3 not in self.manager._lookup_attempts)
98	self.failUnless(3 not in self.manager._keys_refcount)
99	t1.start()
100	self.failUnless(3 in self.manager._lookup_attempts)
101	self.failUnless(3 not in self.manager._keys_refcount)
102	t2.start()
103	time.sleep(.2)
104	cond.acquire()
105	cond.notifyAll()
106	cond.release()
107	cond.acquire()
108	cond.notifyAll()
109	cond.release()
110	t1.join()
111	t2.join()
112	self.failUnless(self.manager._acquire_count == 1)
113	self.failUnless(self.manager._keys_refcount[3] == 2)
114	self.manager.releaseKey(3)
115	self.manager.releaseKey(3)
116	self.failUnless(3 not in self.manager._keys_refcount)
117
118	def test_error(self):
119	self.manager.errorOnAcquire(Exception("known error."))
120	did_fail = False
121	try:
122	self.manager.acquireValue()
123	except:
124	did_fail = True
125	self.failUnless(did_fail, msg="Error on acquire failed.")
126	self.manager.errorOnAcquire(None)
127	self.failUnless(self.manager.acquireValue(3) == 9)
128	self.failUnless(self.manager._acquire_count == 1)
129	self.failUnless(self.manager._keys_refcount[3] == 1)
130	self.manager.releaseKey(3)
131	self.failUnless(3 not in self.manager._keys_refcount)
132
133	if __name__ == '__main__':
134	unittest.main()
135