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

Last change on this file since 4336 was 4336, checked in by ole, 17 years ago

Fix a bug in test_caching.py so that it will work with test_all.py
File size: 11.8 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
22class Dummy:
23  def __init__(self, value, another):
24    self.value = value
25
26
27class Dummy_memorytest:
28  def __init__(self, value, another):
29    self.value = value
30
31
32def clear_and_create_cache(Dummy, verbose=False):
33  a = cache(Dummy, 'clear', verbose=verbose)
34       
35  a = cache(Dummy, args=(9,10),
36            verbose=verbose)
37     
38
39def retrieve_cache(Dummy, verbose=False):
40  if verbose: print 'Check that cache is there'
41  assert cache(Dummy, args=(9,10), test=1,
42               verbose=verbose)     
43     
44
45   
46
47class Test_Caching(unittest.TestCase):
48    def setUp(self):
49        set_option('verbose', 0)  #Why
50
51        pass
52
53    def tearDown(self):
54        pass
55
56    def test_simple(self):
57        """Check set_option (and switch stats off)
58        """
59
60        set_option('savestat', 0)
61        assert options['savestat'] == 0
62        set_option('verbose', 0)
63        assert options['verbose'] == 0       
64       
65
66    def test_basic_caching(self):
67
68        verbose=False
69        # Make some test input arguments
70        #
71        N = 5000  #Make N fairly small here
72
73        a = [1,2]
74        b = ('Thou shalt count the number three',4)
75        c = {'Five is right out': 6, (7,8): 9}
76        x = 3
77        y = 'holy hand granate'
78
79        # Test caching
80        #
81
82        comprange = 2
83
84        for comp in range(comprange):
85 
86            # Evaluate and store
87            #
88            T1 = cache(f, (a,b,c,N), {'x':x, 'y':y}, evaluate=1, \
89                       compression=comp, verbose=verbose)
90
91            # Retrieve
92            #                           
93            T2 = cache(f, (a,b,c,N), {'x':x, 'y':y}, compression=comp) 
94
95            # Reference result
96            #   
97            T3 = f(a,b,c,N,x=x,y=y)  # Compute without caching
98
99
100            assert T1 == T2, 'Cached result does not match computed result'
101            assert T2 == T3, 'Cached result does not match computed result'
102           
103
104    def test_cachefiles(self):
105        """Test existence of cachefiles
106        """       
107        N = 5000  #Make N fairly small here
108
109        a = [1,2]
110        b = ('Thou shalt count the number three',4)
111        c = {'Five is right out': 6, (7,8): 9}
112        x = 3
113        y = 'holy hand granate'
114
115       
116        FN = cache(f,(a,b,c,N), {'x':x, 'y':y}, verbose=0, \
117                  return_filename = 1)
118
119
120        assert FN[:2] == 'f['
121
122        CD = checkdir(cachedir)
123        compression = 1
124
125        (datafile,compressed0) = myopen(CD+FN+'_'+file_types[0],"rb",compression)
126        (argsfile,compressed1) = myopen(CD+FN+'_'+file_types[1],"rb",compression)
127        (admfile,compressed2) =  myopen(CD+FN+'_'+file_types[2],"rb",compression)
128
129        datafile.close()
130        argsfile.close()
131        admfile.close()
132
133    def test_test(self):       
134        """Test 'test' function when cache is present
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        T4 = cache(f,(a,b,c,N), {'x':x, 'y':y}, test=1)
148        assert T1 == T4, "Option 'test' when cache file present failed"     
149
150
151    def test_clear(self):       
152        """Test that 'clear' works
153        """
154
155        N = 5000  #Make N fairly small here
156
157        a = [1,2]
158        b = ('Thou shalt count the number three',4)
159        c = {'Five is right out': 6, (7,8): 9}
160        x = 3
161        y = 'holy hand granate'
162       
163
164        T1 = cache(f,(a,b,c,N), {'x':x, 'y':y}, evaluate = 1)
165       
166       
167        cache(f, (a,b,c,N), {'x':x, 'y':y}, clear = 1)   
168
169 
170        # Test 'test' function when cache is absent
171       
172       
173        T4 = cache(f, (a,b,c,N), {'x':x, 'y':y}, test=1)
174        #print 'T4', T4
175        assert T4 is None, "Option 'test' when cache absent failed"
176
177
178    def test_dependencies(self):
179       
180        # Make a dependency file
181        CD = checkdir(cachedir)       
182
183        DepFN = CD + 'testfile.tmp'
184        DepFN_wildcard = CD + 'test*.tmp'
185        Depfile = open(DepFN,'w')
186        Depfile.write('We are the knights who say NI!')
187        Depfile.close()
188
189       
190        # Test
191        #
192
193        N = 5000  #Make N fairly small here
194
195        a = [1,2]
196        b = ('Thou shalt count the number three',4)
197        c = {'Five is right out': 6, (7,8): 9}
198        x = 3
199        y = 'holy hand granate'
200       
201        T1 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN) 
202        T2 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN)                     
203                       
204        assert T1 == T2, 'Dependencies do not work'
205
206
207        # Test basic wildcard dependency
208        T3 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN_wildcard)                     
209   
210        assert T1 == T3, 'Dependencies with wildcards do not work'
211
212
213        # Test that changed timestamp in dependencies triggers recomputation
214 
215        # Modify dependency file
216        Depfile = open(DepFN,'a')
217        Depfile.write('You must cut down the mightiest tree in the forest with a Herring')
218        Depfile.close()
219 
220        T3 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN, test = 1)
221       
222        assert T3 is None, 'Changed dependencies not recognised'
223 
224        # Test recomputation when dependencies have changed
225        #
226        T3 = cache(f,(a,b,c,N), {'x':x, 'y':y}, dependencies=DepFN)                       
227        assert T1 == T3, 'Recomputed value with changed dependencies failed'
228
229    #def test_performance(self):       
230    #    """Performance test (with statistics)
231    #    Don't really rely on this as it will depend on specific computer.
232    #    """
233    #
234    #    import time
235    #    set_option('savestat', 1)
236    #
237    #    N = 300000   #Should be large on fast computers...
238    #    a = [1,2]
239    #    b = ('Thou shalt count the number three',4)
240    #    c = {'Five is right out': 6, (7,8): 9}
241    #    x = 3
242    #    y = 'holy hand granate'
243    #     
244    #   
245    #    tt = time.time()
246    #    T1 = cache(f,(a,b,c,N), {'x':x, 'y':y})
247    #    t1 = time.time() - tt
248    #
249    #    tt = time.time()
250    #    T2 = cache(f,(a,b,c,N), {'x':x, 'y':y})
251    #    t2 = time.time() - tt
252     
253    #    assert T1 == T2
254    #     assert t1 > t2, 'WARNING: Performance a bit low - this could be specific to current platform. Try to increase N in this test'
255    #    #test_OK('Performance test: relative time saved = %s pct' \
256    #    #        %str(round((t1-t2)*100/t1,2)))
257
258
259    def test_statsfile(self):                   
260        """Test presence of statistics file
261        """
262        import os, string
263        statsfile  = '.cache_stat'  # Basefilename for cached statistics.
264       
265        CD = checkdir(cachedir)               
266        DIRLIST = os.listdir(CD)
267        SF = []
268        for FN in DIRLIST:
269            if string.find(FN,statsfile) >= 0:
270                try:
271                    fid = open(CD+FN,'r')
272                    fid.close()
273                except:
274                    raise 'Statistics files cannot be opened'         
275 
276         
277
278    # def test_network_cachedir(self):
279
280#         #set_option('cachedir', 'H:\\.python_cache\\')
281#         set_option('cachedir', 'V:\\2\\cit\\.python_cache\\')
282#         set_option('verbose', 1)
283
284       
285#         # Make some test input arguments
286#         #
287#         N = 5000  #Make N fairly small here
288
289#         a = [1,2]
290#         b = ('Thou shalt count the number three',4)
291#         c = {'Five is right out': 6, (7,8): 9}
292#         x = 3
293#         y = 'holy hand granate'
294
295#         # Test caching
296#         #
297
298#         comprange = 2
299
300#         for comp in range(comprange):
301 
302#             # Evaluate and store
303#             #
304#             T1 = cache(f, (a,b,c,N), {'x':x, 'y':y}, evaluate=1, \
305#                        compression=comp)
306
307#             # Retrieve
308#             #                           
309#             T2 = cache(f, (a,b,c,N), {'x':x, 'y':y}, compression=comp)
310
311#             # Reference result
312#             #   
313#             T3 = f(a,b,c,N,x=x,y=y)  # Compute without caching
314
315
316#             assert T1 == T2, 'Cached result does not match computed result'
317#             assert T2 == T3, 'Cached result does not match computed result'
318             
319
320
321    def Will_fail_test_objects(self):
322      """
323      This test shows how instances can't be effectively cached.
324      myhash uses hash which uses id which uses the memory address.
325      """
326      verbose = True
327      #verbose = False
328
329      for i in range(2):
330        if verbose: print "clear cache"
331        a = cache(Dummy, 'clear')
332       
333        if verbose: print "cache for first time"
334        a = cache(Dummy, args=(9,10), verbose=verbose)
335        hash_value = myhash(a)
336       
337        #print "hash_value",hash_value
338        if verbose: print "cache for second time"
339        a = cache(Dummy, args=(9,10), verbose=verbose)
340       
341        #print "myhash(a)",myhash(a)
342        assert hash_value == myhash(a)
343
344
345    def test_objects_are_created(self):
346      """
347      However, this test shows how instances can be created from cache
348      as long as input arguments are unchanged.
349
350      Such instances will have different id's and cannot be used as input
351      arguments in subsequent caches. However, this is still useful.
352
353      Do it for all combinations of compression
354
355      """
356
357      verbose = False
358
359      for compression_store in [False, True]:
360        for compression_retrieve in [False, True]:       
361       
362          if verbose: print 'clear cache'
363          a = cache(Dummy, 'clear')
364       
365          if verbose: print 'cache for first time'
366          a = cache(Dummy, args=(9,10),
367                    compression=compression_store,
368                    verbose=verbose)
369         
370          if verbose: print 'Check that cache is there'
371          assert cache(Dummy, args=(9,10), test=1,
372                       compression=compression_retrieve,
373                       verbose=verbose)
374
375
376
377
378    def test_objects_are_created_memory(self):
379      """
380     
381      This test shows how instances can be created from cache
382      as long as input arguments are unchanged - even if the class
383      lives in different memory locations.
384
385      This is using cache created in the main program
386
387      """
388
389      verbose = False
390
391      # Redefine class Dummy_memorytest
392      class Dummy_memorytest:
393        def __init__(self, value, another):
394          self.value = value     
395
396      # Make sure that class has been redefined to another address
397      #print
398      #print 'Initial_addr  ', initial_addr
399      #print 'Redefined addr', `Dummy_memorytest`
400      msg = 'Redefined class ended up at same memory location as '
401      msg += 'original class making this test irrelevant. Try to run '
402      msg += 'it again and see if this error goes away.'
403      msg += 'If it persists contact Ole.Nielsen@ga.gov.au'
404      assert initial_addr != `Dummy_memorytest`, msg   
405
406     
407      retrieve_cache(Dummy_memorytest, verbose=verbose)     
408         
409
410# Define class Dummy_memorytest before any tests are run
411# to make sure it has a different memory address
412# to the one defined in test 'test_objects_are_created_memory'
413class Dummy_memorytest:
414  def __init__(self, value, another):
415    self.value = value     
416
417# Cache created for use with 'test_objects_are_created_memory'
418initial_addr = `Dummy_memorytest`
419clear_and_create_cache(Dummy_memorytest, verbose=False)
420 
421     
422
423
424
425     
426       
427
428
429#-------------------------------------------------------------
430if __name__ == "__main__":
431    suite = unittest.makeSuite(Test_Caching,'test')
432    runner = unittest.TextTestRunner()
433    runner.run(suite)
Note: See TracBrowser for help on using the repository browser.