tests/test_cached_property.py

   1 # -*- coding: utf-8 -*-
   2
   3 """
   4 tests.py
   5 ----------------------------------
   6
   7 Tests for `cached-property` module.
   8 """
   9
  10 from time import sleep
  11 from threading import Lock, Thread
  12 import unittest
  13
  14 from cached_property import cached_property
  15
  16
  17 class TestCachedProperty(unittest.TestCase):
  18
  19     def test_cached_property(self):
  20
  21         class Check(object):
  22
  23             def __init__(self):
  24                 self.total1 = 0
  25                 self.total2 = 0
  26
  27             @property
  28             def add_control(self):
  29                 self.total1 += 1
  30                 return self.total1
  31
  32             @cached_property
  33             def add_cached(self):
  34                 self.total2 += 1
  35                 return self.total2
  36
  37         c = Check()
  38
  39         # The control shows that we can continue to add 1.
  40         self.assertEqual(c.add_control, 1)
  41         self.assertEqual(c.add_control, 2)
  42
  43         # The cached version demonstrates how nothing new is added
  44         self.assertEqual(c.add_cached, 1)
  45         self.assertEqual(c.add_cached, 1)
  46
  47     def test_reset_cached_property(self):
  48
  49         class Check(object):
  50
  51             def __init__(self):
  52                 self.total = 0
  53
  54             @cached_property
  55             def add_cached(self):
  56                 self.total += 1
  57                 return self.total
  58
  59         c = Check()
  60
  61         # Run standard cache assertion
  62         self.assertEqual(c.add_cached, 1)
  63         self.assertEqual(c.add_cached, 1)
  64
  65         # Reset the cache.
  66         del c.add_cached
  67         self.assertEqual(c.add_cached, 2)
  68         self.assertEqual(c.add_cached, 2)
  69
  70     def test_none_cached_property(self):
  71
  72         class Check(object):
  73
  74             def __init__(self):
  75                 self.total = None
  76
  77             @cached_property
  78             def add_cached(self):
  79                 return self.total
  80
  81         c = Check()
  82
  83         # Run standard cache assertion
  84         self.assertEqual(c.add_cached, None)
  85
  86
  87 class TestThreadingIssues(unittest.TestCase):
  88
  89     def test_threads(self):
  90         """ How well does the standard cached_property implementation work with threads?
  91             Short answer: It doesn't! Use threaded_cached_property instead!
  92         """
  93
  94         class Check(object):
  95
  96             def __init__(self):
  97                 self.total = 0
  98                 self.lock = Lock()
  99
 100             @cached_property
 101             def add_cached(self):
 102                 sleep(1)
 103                 # Need to guard this since += isn't atomic.
 104                 with self.lock:
 105                     self.total += 1
 106                 return self.total
 107
 108         c = Check()
 109         threads = []
 110         num_threads = 10
 111         for x in range(num_threads):
 112             thread = Thread(target=lambda: c.add_cached)
 113             thread.start()
 114             threads.append(thread)
 115
 116         for thread in threads:
 117             thread.join()
 118
 119         # Threads means that caching is bypassed.
 120         self.assertNotEqual(c.add_cached, 1)
 121
 122         # This assertion hinges on the fact the system executing the test can
 123         # spawn and start running num_threads threads within the sleep period
 124         # (defined in the Check class as 1 second). If num_threads were to be
 125         # massively increased (try 10000), the actual value returned would be
 126         # between 1 and num_threads, depending on thread scheduling and
 127         # preemption.
 128         self.assertEqual(c.add_cached, num_threads)