source: anuga_core/source/anuga/caching/test_caching.py @ 4154

Last change on this file since 4154 was 3715, checked in by ole, 18 years ago

Enabled caching in create_mesh_from_regions as requested in ticket:80 and
added a unittest for it.
File size: 8.2 KB
Line 
1
2import unittest
3from caching import *
4
5
6
7# Define a test function to be cached
8#
9def f(a,b,c,N,x=0,y='abcdefg'):
10  """f(a,b,c,N)
11     Do something time consuming and produce a complex result.
12  """
13
14  import string
15
16  B = []
17  for n in range(N):
18    s = str(n+2.0/(n + 4.0))+'.a'*10
19    B.append((a,b,c,s,n,x,y))
20  return(B)
21
22
23
24
25
26
27
28class Test_Caching(unittest.TestCase):
29    def setUp(self):
30        set_option('verbose', 0)  #Why
31       
32        pass
33
34    def tearDown(self):
35        pass
36
37    def test_simple(self):
38        """Check set_option (and switch stats off)
39        """
40
41        set_option('savestat', 0)
42        assert options['savestat'] == 0
43        set_option('verbose', 0)
44        assert options['verbose'] == 0       
45       
46
47    def test_basic_caching(self):
48
49       
50        # Make some test input arguments
51        #
52        N = 5000  #Make N fairly small here
53
54        a = [1,2]
55        b = ('Thou shalt count the number three',4)
56        c = {'Five is right out': 6, (7,8): 9}
57        x = 3
58        y = 'holy hand granate'
59
60        # Test caching
61        #
62
63        comprange = 2
64
65        for comp in range(comprange):
66 
67            # Evaluate and store
68            #
69            T1 = cache(f, (a,b,c,N), {'x':x, 'y':y}, evaluate=1, \
70                       compression=comp)
71
72            # Retrieve
73            #                           
74            T2 = cache(f, (a,b,c,N), {'x':x, 'y':y}, compression=comp) 
75
76            # Reference result
77            #   
78            T3 = f(a,b,c,N,x=x,y=y)  # Compute without caching
79
80
81            assert T1 == T2, 'Cached result does not match computed result'
82            assert T2 == T3, 'Cached result does not match computed result'
83           
84
85    def test_cachefiles(self):
86        """Test existence of cachefiles
87        """       
88        N = 5000  #Make N fairly small here
89
90        a = [1,2]
91        b = ('Thou shalt count the number three',4)
92        c = {'Five is right out': 6, (7,8): 9}
93        x = 3
94        y = 'holy hand granate'
95
96       
97        FN = cache(f,(a,b,c,N), {'x':x, 'y':y}, verbose=0, \
98                  return_filename = 1)
99
100
101        assert FN[:2] == 'f['
102
103        CD = checkdir(cachedir)
104        compression = 1
105
106        (datafile,compressed0) = myopen(CD+FN+'_'+file_types[0],"rb",compression)
107        (argsfile,compressed1) = myopen(CD+FN+'_'+file_types[1],"rb",compression)
108        (admfile,compressed2) =  myopen(CD+FN+'_'+file_types[2],"rb",compression)
109
110        datafile.close()
111        argsfile.close()
112        admfile.close()
113
114    def test_test(self):       
115        """Test 'test' function when cache is present
116        """
117        N = 5000  #Make N fairly small here
118
119        a = [1,2]
120        b = ('Thou shalt count the number three',4)
121        c = {'Five is right out': 6, (7,8): 9}
122        x = 3
123        y = 'holy hand granate'
124       
125
126        T1 = cache(f,(a,b,c,N), {'x':x, 'y':y}, evaluate=1)
127       
128        T4 = cache(f,(a,b,c,N), {'x':x, 'y':y}, test=1)
129        assert T1 == T4, "Option 'test' when cache file present failed"     
130
131
132    def test_clear(self):       
133        """Test that 'clear' works
134        """
135
136        N = 5000  #Make N fairly small here
137
138        a = [1,2]
139        b = ('Thou shalt count the number three',4)
140        c = {'Five is right out': 6, (7,8): 9}
141        x = 3
142        y = 'holy hand granate'
143       
144
145        T1 = cache(f,(a,b,c,N), {'x':x, 'y':y}, evaluate = 1)
146       
147       
148        cache(f, (a,b,c,N), {'x':x, 'y':y}, clear = 1)   
149
150 
151        # Test 'test' function when cache is absent
152       
153       
154        T4 = cache(f, (a,b,c,N), {'x':x, 'y':y}, test=1)
155        #print 'T4', T4
156        assert T4 is None, "Option 'test' when cache absent failed"
157
158
159    def test_dependencies(self):
160       
161        # Make a dependency file
162        CD = checkdir(cachedir)       
163
164        DepFN = CD + 'testfile.tmp'
165        DepFN_wildcard = CD + 'test*.tmp'
166        Depfile = open(DepFN,'w')
167        Depfile.write('We are the knights who say NI!')
168        Depfile.close()
169
170       
171        # Test
172        #
173
174        N = 5000  #Make N fairly small here
175
176        a = [1,2]
177        b = ('Thou shalt count the number three',4)
178        c = {'Five is right out': 6, (7,8): 9}
179        x = 3
180        y = 'holy hand granate'
181       
182        T1 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN) 
183        T2 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN)                     
184                       
185        assert T1 == T2, 'Dependencies do not work'
186
187
188        # Test basic wildcard dependency
189        T3 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN_wildcard)                     
190   
191        assert T1 == T3, 'Dependencies with wildcards do not work'
192
193
194        # Test that changed timestamp in dependencies triggers recomputation
195 
196        # Modify dependency file
197        Depfile = open(DepFN,'a')
198        Depfile.write('You must cut down the mightiest tree in the forest with a Herring')
199        Depfile.close()
200 
201        T3 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN, test = 1)
202       
203        assert T3 is None, 'Changed dependencies not recognised'
204 
205        # Test recomputation when dependencies have changed
206        #
207        T3 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN)                       
208        assert T1 == T3, 'Recomputed value with changed dependencies failed'
209
210    #def test_performance(self):       
211    #    """Performance test (with statistics)
212    #    Don't really rely on this as it will depend on specific computer.
213    #    """
214    #
215    #    import time
216    #    set_option('savestat', 1)
217    #
218    #    N = 300000   #Should be large on fast computers...
219    #    a = [1,2]
220    #    b = ('Thou shalt count the number three',4)
221    #    c = {'Five is right out': 6, (7,8): 9}
222    #    x = 3
223    #    y = 'holy hand granate'
224    #     
225    #   
226    #    tt = time.time()
227    #    T1 = cache(f,(a,b,c,N), {'x':x, 'y':y})
228    #    t1 = time.time() - tt
229    #
230    #    tt = time.time()
231    #    T2 = cache(f,(a,b,c,N), {'x':x, 'y':y})
232    #    t2 = time.time() - tt
233     
234    #    assert T1 == T2
235    #     assert t1 > t2, 'WARNING: Performance a bit low - this could be specific to current platform. Try to increase N in this test'
236    #    #test_OK('Performance test: relative time saved = %s pct' \
237    #    #        %str(round((t1-t2)*100/t1,2)))
238
239
240    def test_statsfile(self):                   
241        """Test presence of statistics file
242        """
243        import os, string
244        statsfile  = '.cache_stat'  # Basefilename for cached statistics.
245       
246        CD = checkdir(cachedir)               
247        DIRLIST = os.listdir(CD)
248        SF = []
249        for FN in DIRLIST:
250            if string.find(FN,statsfile) >= 0:
251                try:
252                    fid = open(CD+FN,'r')
253                    fid.close()
254                except:
255                    raise 'Statistics files cannot be opened'         
256 
257         
258
259    # def test_network_cachedir(self):
260
261#         #set_option('cachedir', 'H:\\.python_cache\\')
262#         set_option('cachedir', 'V:\\2\\cit\\.python_cache\\')
263#         set_option('verbose', 1)
264
265       
266#         # Make some test input arguments
267#         #
268#         N = 5000  #Make N fairly small here
269
270#         a = [1,2]
271#         b = ('Thou shalt count the number three',4)
272#         c = {'Five is right out': 6, (7,8): 9}
273#         x = 3
274#         y = 'holy hand granate'
275
276#         # Test caching
277#         #
278
279#         comprange = 2
280
281#         for comp in range(comprange):
282 
283#             # Evaluate and store
284#             #
285#             T1 = cache(f, (a,b,c,N), {'x':x, 'y':y}, evaluate=1, \
286#                        compression=comp)
287
288#             # Retrieve
289#             #                           
290#             T2 = cache(f, (a,b,c,N), {'x':x, 'y':y}, compression=comp)
291
292#             # Reference result
293#             #   
294#             T3 = f(a,b,c,N,x=x,y=y)  # Compute without caching
295
296
297#             assert T1 == T2, 'Cached result does not match computed result'
298#             assert T2 == T3, 'Cached result does not match computed result'
299             
300
301
302
303
304#-------------------------------------------------------------
305if __name__ == "__main__":
306    suite = unittest.makeSuite(Test_Caching,'test')
307    runner = unittest.TextTestRunner()
308    runner.run(suite)
Note: See TracBrowser for help on using the repository browser.