Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/repoze/lru/tests.py @ 1:56ad4e20f292 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
comparison
equal deleted inserted replaced
0:d30785e31577 1:56ad4e20f292
1 import random
2 import time
3 import unittest
4
5 try:
6 range = xrange
7 except NameError: # pragma: NO COVER (Python3)
8 pass
9
10
11 class UnboundedCacheTests(unittest.TestCase):
12
13 def _getTargetClass(self):
14 from repoze.lru import UnboundedCache
15 return UnboundedCache
16
17 def _makeOne(self):
18 return self._getTargetClass()()
19
20 def test_ctor(self):
21 cache = self._makeOne()
22 self.assertEqual(cache._data, {})
23
24 def test_get_miss_no_default(self):
25 cache = self._makeOne()
26 self.assertIsNone(cache.get('nonesuch'))
27
28 def test_get_miss_explicit_default(self):
29 cache = self._makeOne()
30 default = object()
31 self.assertIs(cache.get('nonesuch', default), default)
32
33 def test_get_hit(self):
34 cache = self._makeOne()
35 extant = cache._data['extant'] = object()
36 self.assertIs(cache.get('extant'), extant)
37
38 def test_clear(self):
39 cache = self._makeOne()
40 extant = cache._data['extant'] = object()
41 cache.clear()
42 self.assertIsNone(cache.get('extant'))
43
44 def test_invalidate_miss(self):
45 cache = self._makeOne()
46 cache.invalidate('nonesuch') # does not raise
47
48 def test_invalidate_hit(self):
49 cache = self._makeOne()
50 extant = cache._data['extant'] = object()
51 cache.invalidate('extant')
52 self.assertIsNone(cache.get('extant'))
53
54 def test_put(self):
55 cache = self._makeOne()
56 extant = object()
57 cache.put('extant', extant)
58 self.assertIs(cache._data['extant'], extant)
59
60
61 class LRUCacheTests(unittest.TestCase):
62
63 def _getTargetClass(self):
64 from repoze.lru import LRUCache
65 return LRUCache
66
67 def _makeOne(self, size):
68 return self._getTargetClass()(size)
69
70 def check_cache_is_consistent(self, cache):
71 #Return if cache is consistent, else raise fail test case.
72 # cache.hand/maxpos/size
73 self.assertTrue(cache.hand < len(cache.clock_keys))
74 self.assertTrue(cache.hand >= 0)
75 self.assertEqual(cache.maxpos, cache.size - 1)
76 self.assertEqual(len(cache.clock_keys), cache.size)
77
78 # lengths of data structures
79 self.assertEqual(len(cache.clock_keys), len(cache.clock_refs))
80 self.assertTrue(len(cache.data) <= len(cache.clock_refs))
81
82 # For each item in cache.data
83 # 1. pos must be a valid index
84 # 2. clock_keys must point back to the entry
85 for key, value in cache.data.items():
86 pos, val = value
87 self.assertTrue(
88 type(pos) == type(42) or
89 type(pos) == type(2 ** 128))
90 self.assertTrue(pos >= 0)
91 self.assertTrue(pos <= cache.maxpos)
92
93 clock_key = cache.clock_keys[pos]
94 self.assertTrue(clock_key is key)
95 clock_ref = cache.clock_refs[pos]
96
97 # All clock_refs must be True or False, nothing else.
98 for clock_ref in cache.clock_refs:
99 self.assertTrue(clock_ref is True or clock_ref is False)
100
101 def test_size_lessthan_1(self):
102 self.assertRaises(ValueError, self._makeOne, 0)
103
104 def test_get(self):
105 cache = self._makeOne(1)
106 # Must support different types of keys
107 self.assertIsNone(cache.get("foo"))
108 self.assertIsNone(cache.get(42))
109 self.assertIsNone(cache.get(("foo", 42)))
110 self.assertIsNone(cache.get(None))
111 self.assertIsNone(cache.get(""))
112 self.assertIsNone(cache.get(object()))
113 # Check if default value is used
114 self.assertEqual(cache.get("foo", "bar"), "bar")
115 self.assertEqual(cache.get("foo", default="bar"), "bar")
116
117 self.check_cache_is_consistent(cache)
118
119 def test_put(self):
120 cache = self._makeOne(8)
121 self.check_cache_is_consistent(cache)
122 # Must support different types of keys
123 cache.put("foo", "FOO")
124 cache.put(42, "fortytwo")
125 cache.put( ("foo", 42), "tuple_as_key")
126 cache.put(None, "None_as_key")
127 cache.put("", "empty_string_as_key")
128 cache.put(3.141, "float_as_key")
129 my_object = object()
130 cache.put(my_object, "object_as_key")
131
132 self.check_cache_is_consistent(cache)
133
134 self.assertEqual(cache.get("foo"), "FOO")
135 self.assertEqual(cache.get(42), "fortytwo")
136 self.assertEqual(cache.get(("foo", 42), "fortytwo"), "tuple_as_key")
137 self.assertEqual(cache.get(None), "None_as_key")
138 self.assertEqual(cache.get(""), "empty_string_as_key")
139 self.assertEqual(cache.get(3.141), "float_as_key")
140 self.assertEqual(cache.get(my_object), "object_as_key")
141
142 # put()ing again must overwrite
143 cache.put(42, "fortytwo again")
144 self.assertEqual(cache.get(42), "fortytwo again")
145
146 self.check_cache_is_consistent(cache)
147
148 def test_invalidate(self):
149 cache = self._makeOne(3)
150 cache.put("foo", "bar")
151 cache.put("FOO", "BAR")
152
153 cache.invalidate("foo")
154 self.assertIsNone(cache.get("foo"))
155 self.assertEqual(cache.get("FOO"), "BAR")
156 self.check_cache_is_consistent(cache)
157
158 cache.invalidate("FOO")
159 self.assertIsNone(cache.get("foo"))
160 self.assertIsNone(cache.get("FOO"))
161 self.assertEqual(cache.data, {})
162 self.check_cache_is_consistent(cache)
163
164 cache.put("foo", "bar")
165 cache.invalidate("nonexistingkey")
166 self.assertEqual(cache.get("foo"), "bar")
167 self.assertIsNone(cache.get("FOO"))
168 self.check_cache_is_consistent(cache)
169
170 def test_small_cache(self):
171 #Cache of size 1 must work
172 cache = self._makeOne(1)
173
174 cache.put("foo", "bar")
175 self.assertEqual(cache.get("foo"), "bar")
176 self.check_cache_is_consistent(cache)
177
178 cache.put("FOO", "BAR")
179 self.assertEqual(cache.get("FOO"), "BAR")
180 self.assertIsNone(cache.get("foo"))
181 self.check_cache_is_consistent(cache)
182
183 # put() again
184 cache.put("FOO", "BAR")
185 self.assertEqual(cache.get("FOO"), "BAR")
186 self.assertIsNone(cache.get("foo"))
187 self.check_cache_is_consistent(cache)
188
189 # invalidate()
190 cache.invalidate("FOO")
191 self.check_cache_is_consistent(cache)
192 self.assertIsNone(cache.get("FOO"))
193 self.assertIsNone(cache.get("foo"))
194
195 # clear()
196 cache.put("foo", "bar")
197 self.assertEqual(cache.get("foo"), "bar")
198 cache.clear()
199 self.check_cache_is_consistent(cache)
200 self.assertIsNone(cache.get("FOO"))
201 self.assertIsNone(cache.get("foo"))
202
203 def test_equal_but_not_identical(self):
204 #equal but not identical keys must be treated the same
205 cache = self._makeOne(1)
206 tuple_one = (1, 1)
207 tuple_two = (1, 1)
208 cache.put(tuple_one, 42)
209
210 self.assertEqual(cache.get(tuple_one), 42)
211 self.assertEqual(cache.get(tuple_two), 42)
212 self.check_cache_is_consistent(cache)
213
214 cache = self._makeOne(1)
215 cache.put(tuple_one, 42)
216 cache.invalidate(tuple_two)
217 self.assertIsNone(cache.get(tuple_one))
218 self.assertIsNone(cache.get(tuple_two))
219
220 def test_perfect_hitrate(self):
221 #If cache size equals number of items, expect 100% cache hits
222 size = 1000
223 cache = self._makeOne(size)
224
225 for count in range(size):
226 cache.put(count, "item%s" % count)
227
228 for cache_op in range(10000):
229 item = random.randrange(0, size - 1)
230 if random.getrandbits(1):
231 self.assertEqual(cache.get(item), "item%s" % item)
232 else:
233 cache.put(item, "item%s" % item)
234
235 self.assertEqual(cache.misses, 0)
236 self.assertEqual(cache.evictions, 0)
237
238 self.check_cache_is_consistent(cache)
239
240 def test_imperfect_hitrate(self):
241 #If cache size == half the number of items -> hit rate ~50%
242 size = 1000
243 cache = self._makeOne(size / 2)
244
245 for count in range(size):
246 cache.put(count, "item%s" % count)
247
248 hits = 0
249 misses = 0
250 total_gets = 0
251 for cache_op in range(10000):
252 item = random.randrange(0, size - 1)
253 if random.getrandbits(1):
254 entry = cache.get(item)
255 total_gets += 1
256 self.assertTrue(
257 (entry == "item%s" % item) or
258 entry is None)
259 if entry is None:
260 misses += 1
261 else:
262 hits += 1
263 else:
264 cache.put(item, "item%s" % item)
265
266 # Cache hit rate should be roughly 50%
267 hit_ratio = hits / float(total_gets) * 100
268 self.assertTrue(hit_ratio > 45)
269 self.assertTrue(hit_ratio < 55)
270
271 # The internal cache counters should have the same information
272 internal_hit_ratio = 100 * cache.hits / cache.lookups
273 self.assertTrue(internal_hit_ratio > 45)
274 self.assertTrue(internal_hit_ratio < 55)
275
276 # The internal miss counters should also be around 50%
277 internal_miss_ratio = 100 * cache.misses / cache.lookups
278 self.assertTrue(internal_miss_ratio > 45)
279 self.assertTrue(internal_miss_ratio < 55)
280
281 self.check_cache_is_consistent(cache)
282
283 def test_eviction_counter(self):
284 cache = self._makeOne(2)
285 cache.put(1, 1)
286 cache.put(2, 1)
287 self.assertEqual(cache.evictions, 0)
288
289 cache.put(3, 1)
290 cache.put(4, 1)
291 self.assertEqual(cache.evictions, 2)
292
293 cache.put(3, 1)
294 cache.put(4, 1)
295 self.assertEqual(cache.evictions, 2)
296
297 cache.clear()
298 self.assertEqual(cache.evictions, 0)
299
300
301 def test_it(self):
302 cache = self._makeOne(3)
303 self.assertIsNone(cache.get('a'))
304
305 cache.put('a', '1')
306 pos, value = cache.data.get('a')
307 self.assertEqual(cache.clock_refs[pos], True)
308 self.assertEqual(cache.clock_keys[pos], 'a')
309 self.assertEqual(value, '1')
310 self.assertEqual(cache.get('a'), '1')
311 self.assertEqual(cache.hand, pos + 1)
312
313 pos, value = cache.data.get('a')
314 self.assertEqual(cache.clock_refs[pos], True)
315 self.assertEqual(cache.hand, pos + 1)
316 self.assertEqual(len(cache.data), 1)
317
318 cache.put('b', '2')
319 pos, value = cache.data.get('b')
320 self.assertEqual(cache.clock_refs[pos], True)
321 self.assertEqual(cache.clock_keys[pos], 'b')
322 self.assertEqual(len(cache.data), 2)
323
324 cache.put('c', '3')
325 pos, value = cache.data.get('c')
326 self.assertEqual(cache.clock_refs[pos], True)
327 self.assertEqual(cache.clock_keys[pos], 'c')
328 self.assertEqual(len(cache.data), 3)
329
330 pos, value = cache.data.get('a')
331 self.assertEqual(cache.clock_refs[pos], True)
332
333 cache.get('a')
334 # All items have ref==True. cache.hand points to "a". Putting
335 # "d" will set ref=False on all items and then replace "a",
336 # because "a" is the first item with ref==False that is found.
337 cache.put('d', '4')
338 self.assertEqual(len(cache.data), 3)
339 self.assertIsNone(cache.data.get('a'))
340
341 # Only item "d" has ref==True. cache.hand points at "b", so "b"
342 # will be evicted when "e" is inserted. "c" will be left alone.
343 cache.put('e', '5')
344 self.assertEqual(len(cache.data), 3)
345 self.assertIsNone(cache.data.get('b'))
346 self.assertEqual(cache.get('d'), '4')
347 self.assertEqual(cache.get('e'), '5')
348 self.assertIsNone(cache.get('a'))
349 self.assertIsNone(cache.get('b'))
350 self.assertEqual(cache.get('c'), '3')
351
352 self.check_cache_is_consistent(cache)
353
354
355 class ExpiringLRUCacheTests(LRUCacheTests):
356
357 def _getTargetClass(self):
358 from repoze.lru import ExpiringLRUCache
359 return ExpiringLRUCache
360
361 def _makeOne(self, size, default_timeout=None):
362 if default_timeout is None:
363 return self._getTargetClass()(size)
364 else:
365 return self._getTargetClass()(
366 size, default_timeout=default_timeout)
367
368 def check_cache_is_consistent(self, cache):
369 #Return if cache is consistent, else raise fail test case.
370 #
371 #This is slightly different for ExpiringLRUCache since self.data
372 #contains 3-tuples instead of 2-tuples.
373 # cache.hand/maxpos/size
374 self.assertTrue(cache.hand < len(cache.clock_keys))
375 self.assertTrue(cache.hand >= 0)
376 self.assertEqual(cache.maxpos, cache.size - 1)
377 self.assertEqual(len(cache.clock_keys), cache.size)
378
379 # lengths of data structures
380 self.assertEqual(len(cache.clock_keys), len(cache.clock_refs))
381 self.assertTrue(len(cache.data) <= len(cache.clock_refs))
382
383 # For each item in cache.data
384 # 1. pos must be a valid index
385 # 2. clock_keys must point back to the entry
386 for key, value in cache.data.items():
387 pos, val, timeout = value
388 self.assertTrue(
389 type(pos) == type(42) or type(pos) == type(2 ** 128))
390 self.assertTrue(pos >= 0)
391 self.assertTrue(pos <= cache.maxpos)
392
393 clock_key = cache.clock_keys[pos]
394 self.assertTrue(clock_key is key)
395 clock_ref = cache.clock_refs[pos]
396
397 self.assertTrue(type(timeout) == type(3.141))
398
399 # All clock_refs must be True or False, nothing else.
400 for clock_ref in cache.clock_refs:
401 self.assertTrue(clock_ref is True or clock_ref is False)
402
403 def test_it(self):
404 #Test a sequence of operations
405 #
406 # Looks at internal data, which is different for ExpiringLRUCache.
407 cache = self._makeOne(3)
408 self.assertIsNone(cache.get('a'))
409
410 cache.put('a', '1')
411 pos, value, expires = cache.data.get('a')
412 self.assertEqual(cache.clock_refs[pos], True)
413 self.assertEqual(cache.clock_keys[pos], 'a')
414 self.assertEqual(value, '1')
415 self.assertEqual(cache.get('a'), '1')
416 self.assertEqual(cache.hand, pos + 1)
417
418 pos, value, expires = cache.data.get('a')
419 self.assertEqual(cache.clock_refs[pos], True)
420 self.assertEqual(cache.hand, pos + 1)
421 self.assertEqual(len(cache.data), 1)
422
423 cache.put('b', '2')
424 pos, value, expires = cache.data.get('b')
425 self.assertEqual(cache.clock_refs[pos], True)
426 self.assertEqual(cache.clock_keys[pos], 'b')
427 self.assertEqual(len(cache.data), 2)
428
429 cache.put('c', '3')
430 pos, value, expires = cache.data.get('c')
431 self.assertEqual(cache.clock_refs[pos], True)
432 self.assertEqual(cache.clock_keys[pos], 'c')
433 self.assertEqual(len(cache.data), 3)
434
435 pos, value, expires = cache.data.get('a')
436 self.assertEqual(cache.clock_refs[pos], True)
437
438 cache.get('a')
439 # All items have ref==True. cache.hand points to "a". Putting
440 # "d" will set ref=False on all items and then replace "a",
441 # because "a" is the first item with ref==False that is found.
442 cache.put('d', '4')
443 self.assertEqual(len(cache.data), 3)
444 self.assertIsNone(cache.data.get('a'))
445
446 # Only item "d" has ref==True. cache.hand points at "b", so "b"
447 # will be evicted when "e" is inserted. "c" will be left alone.
448 cache.put('e', '5')
449 self.assertEqual(len(cache.data), 3)
450 self.assertIsNone(cache.data.get('b'))
451 self.assertEqual(cache.get('d'), '4')
452 self.assertEqual(cache.get('e'), '5')
453 self.assertIsNone(cache.get('a'))
454 self.assertIsNone(cache.get('b'))
455 self.assertEqual(cache.get('c'), '3')
456
457 self.check_cache_is_consistent(cache)
458
459 def test_default_timeout(self):
460 #Default timeout provided at init time must be applied.
461 # Provide no default timeout -> entries must remain valid
462 cache = self._makeOne(3)
463 cache.put("foo", "bar")
464
465 time.sleep(0.1)
466 cache.put("FOO", "BAR")
467 self.assertEqual(cache.get("foo"), "bar")
468 self.assertEqual(cache.get("FOO"), "BAR")
469 self.check_cache_is_consistent(cache)
470
471 # Provide short default timeout -> entries must become invalid
472 cache = self._makeOne(3, default_timeout=0.1)
473 cache.put("foo", "bar")
474
475 time.sleep(0.1)
476 cache.put("FOO", "BAR")
477 self.assertIsNone(cache.get("foo"))
478 self.assertEqual(cache.get("FOO"), "BAR")
479 self.check_cache_is_consistent(cache)
480
481 def test_different_timeouts(self):
482 #Timeouts must be per entry, default applied when none provided
483 cache = self._makeOne(3, default_timeout=0.1)
484
485 cache.put("one", 1)
486 cache.put("two", 2, timeout=0.2)
487 cache.put("three", 3, timeout=0.3)
488
489 # All entries still here
490 self.assertEqual(cache.get("one"), 1)
491 self.assertEqual(cache.get("two"), 2)
492 self.assertEqual(cache.get("three"), 3)
493
494 # Entry "one" must expire, "two"/"three" remain valid
495 time.sleep(0.1)
496 self.assertIsNone(cache.get("one"))
497 self.assertEqual(cache.get("two"), 2)
498 self.assertEqual(cache.get("three"), 3)
499
500 # Only "three" remains valid
501 time.sleep(0.1)
502 self.assertIsNone(cache.get("one"))
503 self.assertIsNone(cache.get("two"))
504 self.assertEqual(cache.get("three"), 3)
505
506 # All have expired
507 time.sleep(0.1)
508 self.assertIsNone(cache.get("one"))
509 self.assertIsNone(cache.get("two"))
510 self.assertIsNone(cache.get("three"))
511
512 self.check_cache_is_consistent(cache)
513
514 def test_renew_timeout(self):
515 #Re-putting an entry must update timeout
516 cache = self._makeOne(3, default_timeout=0.2)
517
518 cache.put("foo", "bar")
519 cache.put("foo2", "bar2", timeout=10)
520 cache.put("foo3", "bar3", timeout=10)
521
522 time.sleep(0.1)
523 # All must still be here
524 self.assertEqual(cache.get("foo"), "bar")
525 self.assertEqual(cache.get("foo2"), "bar2")
526 self.assertEqual(cache.get("foo3"), "bar3")
527 self.check_cache_is_consistent(cache)
528
529 # Set new timeouts by re-put()ing the entries
530 cache.put("foo", "bar")
531 cache.put("foo2", "bar2", timeout=0.1)
532 cache.put("foo3", "bar3")
533
534 time.sleep(0.1)
535 # "foo2" must have expired
536 self.assertEqual(cache.get("foo"), "bar")
537 self.assertIsNone(cache.get("foo2"))
538 self.assertEqual(cache.get("foo3"), "bar3")
539 self.check_cache_is_consistent(cache)
540
541
542 class DecoratorTests(unittest.TestCase):
543
544 def _getTargetClass(self):
545 from repoze.lru import lru_cache
546 return lru_cache
547
548 def _makeOne(self, *args, **kw):
549 return self._getTargetClass()(*args, **kw)
550
551 def test_ctor_no_size(self):
552 from repoze.lru import UnboundedCache
553 decorator = self._makeOne(maxsize=None)
554 self.assertIsInstance(decorator.cache, UnboundedCache)
555 self.assertEqual(decorator.cache._data, {})
556
557 def test_ctor_w_size_no_timeout(self):
558 from repoze.lru import LRUCache
559 decorator = self._makeOne(maxsize=10)
560 self.assertIsInstance(decorator.cache, LRUCache)
561 self.assertEqual(decorator.cache.size, 10)
562
563 def test_ctor_w_size_w_timeout(self):
564 from repoze.lru import ExpiringLRUCache
565 decorator = self._makeOne(maxsize=10, timeout=30)
566 self.assertIsInstance(decorator.cache, ExpiringLRUCache)
567 self.assertEqual(decorator.cache.size, 10)
568 self.assertEqual(decorator.cache.default_timeout, 30)
569
570 def test_ctor_nocache(self):
571 decorator = self._makeOne(10, None)
572 self.assertEqual(decorator.cache.size, 10)
573
574 def test_singlearg(self):
575 cache = DummyLRUCache()
576 decorator = self._makeOne(0, cache)
577 def wrapped(key):
578 return key
579 decorated = decorator(wrapped)
580 result = decorated(1)
581 self.assertEqual(cache[(1,)], 1)
582 self.assertEqual(result, 1)
583 self.assertEqual(len(cache), 1)
584 result = decorated(2)
585 self.assertEqual(cache[(2,)], 2)
586 self.assertEqual(result, 2)
587 self.assertEqual(len(cache), 2)
588 result = decorated(2)
589 self.assertEqual(cache[(2,)], 2)
590 self.assertEqual(result, 2)
591 self.assertEqual(len(cache), 2)
592
593 def test_cache_attr(self):
594 cache = DummyLRUCache()
595 decorator = self._makeOne(0, cache)
596 def wrapped(key): #pragma NO COVER
597 return key
598 decorated = decorator(wrapped)
599 self.assertTrue(decorated._cache is cache)
600
601 def test_multiargs(self):
602 cache = DummyLRUCache()
603 decorator = self._makeOne(0, cache)
604 def moreargs(*args):
605 return args
606 decorated = decorator(moreargs)
607 result = decorated(3, 4, 5)
608 self.assertEqual(cache[(3, 4, 5)], (3, 4, 5))
609 self.assertEqual(result, (3, 4, 5))
610 self.assertEqual(len(cache), 1)
611
612 def test_multiargs_keywords(self):
613 cache = DummyLRUCache()
614 decorator = self._makeOne(0, cache)
615 def moreargs(*args, **kwargs):
616 return args, kwargs
617 decorated = decorator(moreargs)
618 result = decorated(3, 4, 5, a=1, b=2, c=3)
619 self.assertEqual(
620 cache[((3, 4, 5), frozenset([ ('a',1), ('b',2), ('c',3) ]))],
621 ((3, 4, 5), {'a':1, 'b':2, 'c':3}))
622 self.assertEqual(result, ((3, 4, 5), {'a':1, 'b':2, 'c':3}))
623 self.assertEqual(len(cache), 1)
624
625 def test_multiargs_keywords_ignore_unhashable_true(self):
626 cache = DummyLRUCache()
627 decorator = self._makeOne(0, cache, ignore_unhashable_args=True)
628 def moreargs(*args, **kwargs):
629 return args, kwargs
630 decorated = decorator(moreargs)
631 result = decorated(3, 4, 5, a=1, b=[1, 2, 3])
632 self.assertEqual(len(cache), 0)
633 self.assertEqual(result, ((3, 4, 5), {'a':1, 'b':[1, 2, 3]}))
634
635 def test_multiargs_keywords_ignore_unhashable(self):
636 cache = DummyLRUCache()
637 decorator = self._makeOne(0, cache, ignore_unhashable_args=False)
638
639 def moreargs(*args, **kwargs): # pragma: NO COVER
640 return args, kwargs
641
642 decorated = decorator(moreargs)
643
644 with self.assertRaises(TypeError):
645 decorated(3, 4, 5, a=1, b=[1, 2, 3])
646
647 def test_expiry(self):
648 #When timeout is given, decorator must eventually forget entries
649 @self._makeOne(1, None, timeout=0.1)
650 def sleep_a_bit(param):
651 time.sleep(0.1)
652 return 2 * param
653
654 # First call must take at least 0.1 seconds
655 start = time.time()
656 result1 = sleep_a_bit("hello")
657 stop = time.time()
658 self.assertEqual(result1, 2 * "hello")
659 self.assertTrue(stop - start > 0.1)
660
661 # Second call must take less than 0.1 seconds.
662 start = time.time()
663 result2 = sleep_a_bit("hello")
664 stop = time.time()
665 self.assertEqual(result2, 2 * "hello")
666 self.assertTrue(stop - start < 0.1)
667
668 time.sleep(0.1)
669 # This one must calculate again and take at least 0.1 seconds
670 start = time.time()
671 result3 = sleep_a_bit("hello")
672 stop = time.time()
673 self.assertEqual(result3, 2 * "hello")
674 self.assertTrue(stop - start > 0.1)
675
676 def test_partial(self):
677 #lru_cache decorator must not crash on functools.partial instances
678 def add(a,b):
679 return a + b
680 from functools import partial
681 from repoze.lru import lru_cache
682 add_five = partial(add, 5)
683 decorated = lru_cache(20)(add_five)
684 self.assertEqual(decorated(3), 8)
685
686
687 class DummyLRUCache(dict):
688
689 def put(self, k, v):
690 return self.__setitem__(k, v)
691
692
693 class CacherMaker(unittest.TestCase):
694
695 def _getTargetClass(self):
696 from repoze.lru import CacheMaker
697 return CacheMaker
698
699 def _makeOne(self, *args, **kw):
700 return self._getTargetClass()(*args, **kw)
701
702 def test_named_cache(self):
703 maker = self._makeOne()
704 size = 10
705 name = "name"
706 decorated = maker.lrucache(maxsize=size, name=name)(_adder)
707 self.assertEqual(list(maker._cache.keys()), [name])
708 self.assertEqual(maker._cache[name].size, size)
709 decorated(10)
710 decorated(11)
711 self.assertEqual(len(maker._cache[name].data),2)
712
713 def test_exception(self):
714 maker = self._makeOne()
715 size = 10
716 name = "name"
717 decorated = maker.lrucache(maxsize=size, name=name)(_adder)
718 self.assertRaises(KeyError, maker.lrucache, maxsize=size, name=name)
719 self.assertRaises(ValueError, maker.lrucache)
720
721 def test_defaultvalue_and_clear(self):
722 size = 10
723 maker = self._makeOne(maxsize=size)
724 for i in range(100):
725 decorated = maker.lrucache()(_adder)
726 decorated(10)
727
728 self.assertEqual(len(maker._cache) , 100)
729 for _cache in maker._cache.values():
730 self.assertEqual( _cache.size,size)
731 self.assertEqual(len(_cache.data),1)
732 ## and test clear cache
733 maker.clear()
734 for _cache in maker._cache.values():
735 self.assertEqual( _cache.size,size)
736 self.assertEqual(len(_cache.data),0)
737
738 def test_clear_with_single_name(self):
739 maker = self._makeOne(maxsize=10)
740 one = maker.lrucache(name='one')(_adder)
741 two = maker.lrucache(name='two')(_adder)
742 for i in range(100):
743 _ = one(i)
744 _ = two(i)
745 self.assertEqual(len(maker._cache['one'].data), 10)
746 self.assertEqual(len(maker._cache['two'].data), 10)
747 maker.clear('one')
748 self.assertEqual(len(maker._cache['one'].data), 0)
749 self.assertEqual(len(maker._cache['two'].data), 10)
750
751 def test_clear_with_multiple_names(self):
752 maker = self._makeOne(maxsize=10)
753 one = maker.lrucache(name='one')(_adder)
754 two = maker.lrucache(name='two')(_adder)
755 three = maker.lrucache(name='three')(_adder)
756 for i in range(100):
757 _ = one(i)
758 _ = two(i)
759 _ = three(i)
760 self.assertEqual(len(maker._cache['one'].data), 10)
761 self.assertEqual(len(maker._cache['two'].data), 10)
762 self.assertEqual(len(maker._cache['three'].data), 10)
763 maker.clear('one', 'three')
764 self.assertEqual(len(maker._cache['one'].data), 0)
765 self.assertEqual(len(maker._cache['two'].data), 10)
766 self.assertEqual(len(maker._cache['three'].data), 0)
767
768 def test_memoized(self):
769 from repoze.lru import lru_cache
770 from repoze.lru import UnboundedCache
771 maker = self._makeOne(maxsize=10)
772 memo = maker.memoized('test')
773 self.assertIsInstance(memo, lru_cache)
774 self.assertIsInstance(memo.cache, UnboundedCache)
775 self.assertIs(memo.cache, maker._cache['test'])
776
777 def test_expiring(self):
778 size = 10
779 timeout = 10
780 name = "name"
781 cache = self._makeOne(maxsize=size, timeout=timeout)
782 for i in range(100):
783 if not i:
784 decorator = cache.expiring_lrucache(name=name)
785 decorated = decorator(_adder)
786 self.assertEqual( cache._cache[name].size,size)
787 else:
788 decorator = cache.expiring_lrucache()
789 decorated = decorator(_adder)
790 self.assertEqual(decorator.cache.default_timeout, timeout)
791 decorated(10)
792
793 self.assertEqual( len(cache._cache) , 100)
794 for _cache in cache._cache.values():
795 self.assertEqual( _cache.size,size)
796 self.assertEqual( _cache.default_timeout,timeout)
797 self.assertEqual(len(_cache.data),1)
798 ## and test clear cache
799 cache.clear()
800 for _cache in cache._cache.values():
801 self.assertEqual( _cache.size,size)
802 self.assertEqual(len(_cache.data),0)
803
804 def test_expiring_w_timeout(self):
805 size = 10
806 maker_timeout = 10
807 timeout = 20
808 name = "name"
809 cache = self._makeOne(maxsize=size, timeout=maker_timeout)
810 decorator = cache.expiring_lrucache(name=name, timeout=20)
811 self.assertEqual(decorator.cache.default_timeout, timeout)
812
813 def _adder(x):
814 return x + 10