tests/test_cached_property.py

   1 # -*- coding: utf-8 -*-
   2
   3 import time
   4 import unittest
   5 from threading import Lock, Thread
   6 from freezegun import freeze_time
   7
   8 import cached_property
   9
  10
  11 def CheckFactory(cached_property_decorator, threadsafe=False):
  12     """
  13     Create dynamically a Check class whose add_cached method is decorated by
  14     the cached_property_decorator.
  15     """
  16
  17     class Check(object):
  18
  19         def __init__(self):
  20             self.control_total = 0
  21             self.cached_total = 0
  22             self.lock = Lock()
  23
  24         @property
  25         def add_control(self):
  26             self.control_total += 1
  27             return self.control_total
  28
  29         @cached_property_decorator
  30         def add_cached(self):
  31             if threadsafe:
  32                 time.sleep(1)
  33                 # Need to guard this since += isn't atomic.
  34                 with self.lock:
  35                     self.cached_total += 1
  36             else:
  37                 self.cached_total += 1
  38
  39             return self.cached_total
  40
  41         def run_threads(self, num_threads):
  42             threads = []
  43             for _ in range(num_threads):
  44                 thread = Thread(target=lambda: self.add_cached)
  45                 thread.start()
  46                 threads.append(thread)
  47             for thread in threads:
  48                 thread.join()
  49
  50     return Check
  51
  52
  53 class TestCachedProperty(unittest.TestCase):
  54     """Tests for cached_property"""
  55
  56     cached_property_factory = cached_property.cached_property
  57
  58     def assert_control(self, check, expected):
  59         """
  60         Assert that both `add_control` and 'control_total` equal `expected`
  61         """
  62         self.assertEqual(check.add_control, expected)
  63         self.assertEqual(check.control_total, expected)
  64
  65     def assert_cached(self, check, expected):
  66         """
  67         Assert that both `add_cached` and 'cached_total` equal `expected`
  68         """
  69         self.assertEqual(check.add_cached, expected)
  70         self.assertEqual(check.cached_total, expected)
  71
  72     def test_cached_property(self):
  73         Check = CheckFactory(self.cached_property_factory)
  74         check = Check()
  75
  76         # The control shows that we can continue to add 1
  77         self.assert_control(check, 1)
  78         self.assert_control(check, 2)
  79
  80         # The cached version demonstrates how nothing is added after the first
  81         self.assert_cached(check, 1)
  82         self.assert_cached(check, 1)
  83
  84         # The cache does not expire
  85         with freeze_time("9999-01-01"):
  86             self.assert_cached(check, 1)
  87
  88         # Typically descriptors return themselves if accessed though the class
  89         # rather than through an instance.
  90         self.assertTrue(isinstance(Check.add_cached,
  91                                    self.cached_property_factory))
  92
  93     def test_reset_cached_property(self):
  94         Check = CheckFactory(self.cached_property_factory)
  95         check = Check()
  96
  97         # Run standard cache assertion
  98         self.assert_cached(check, 1)
  99         self.assert_cached(check, 1)
 100
 101         # Clear the cache
 102         del check.add_cached
 103
 104         # Value is cached again after the next access
 105         self.assert_cached(check, 2)
 106         self.assert_cached(check, 2)
 107
 108     def test_none_cached_property(self):
 109         class Check(object):
 110
 111             def __init__(self):
 112                 self.cached_total = None
 113
 114             @self.cached_property_factory
 115             def add_cached(self):
 116                 return self.cached_total
 117
 118         self.assert_cached(Check(), None)
 119
 120     def test_set_cached_property(self):
 121         Check = CheckFactory(self.cached_property_factory)
 122         check = Check()
 123         check.add_cached = 'foo'
 124         self.assertEqual(check.add_cached, 'foo')
 125         self.assertEqual(check.cached_total, 0)
 126
 127     def test_threads(self):
 128         Check = CheckFactory(self.cached_property_factory, threadsafe=True)
 129         check = Check()
 130         num_threads = 5
 131
 132         # cached_property_with_ttl is *not* thread-safe!
 133         check.run_threads(num_threads)
 134         # This assertion hinges on the fact the system executing the test can
 135         # spawn and start running num_threads threads within the sleep period
 136         # (defined in the Check class as 1 second). If num_threads were to be
 137         # massively increased (try 10000), the actual value returned would be
 138         # between 1 and num_threads, depending on thread scheduling and
 139         # preemption.
 140         self.assert_cached(check, num_threads)
 141         self.assert_cached(check, num_threads)
 142
 143         # The cache does not expire
 144         with freeze_time("9999-01-01"):
 145             check.run_threads(num_threads)
 146             self.assert_cached(check, num_threads)
 147             self.assert_cached(check, num_threads)
 148
 149
 150 class TestThreadedCachedProperty(TestCachedProperty):
 151     """Tests for threaded_cached_property"""
 152
 153     cached_property_factory = cached_property.threaded_cached_property
 154
 155     def test_threads(self):
 156         Check = CheckFactory(self.cached_property_factory, threadsafe=True)
 157         check = Check()
 158         num_threads = 5
 159
 160         # threaded_cached_property_with_ttl is thread-safe
 161         check.run_threads(num_threads)
 162         self.assert_cached(check, 1)
 163         self.assert_cached(check, 1)
 164
 165         # The cache does not expire
 166         with freeze_time("9999-01-01"):
 167             check.run_threads(num_threads)
 168             self.assert_cached(check, 1)
 169             self.assert_cached(check, 1)
 170
 171
 172 class TestCachedPropertyWithTTL(TestCachedProperty):
 173     """Tests for cached_property_with_ttl"""
 174
 175     cached_property_factory = cached_property.cached_property_with_ttl
 176
 177     def test_ttl_expiry(self):
 178         Check = CheckFactory(self.cached_property_factory(ttl=100000))
 179         check = Check()
 180
 181         # Run standard cache assertion
 182         self.assert_cached(check, 1)
 183         self.assert_cached(check, 1)
 184
 185         # The cache expires in the future
 186         with freeze_time("9999-01-01"):
 187             self.assert_cached(check, 2)
 188             self.assert_cached(check, 2)
 189
 190         # Things are not reverted when we are back to the present
 191         self.assert_cached(check, 2)
 192         self.assert_cached(check, 2)
 193
 194     def test_threads_ttl_expiry(self):
 195         Check = CheckFactory(self.cached_property_factory(ttl=100000),
 196                              threadsafe=True)
 197         check = Check()
 198         num_threads = 5
 199
 200         # Same as in test_threads
 201         check.run_threads(num_threads)
 202         self.assert_cached(check, num_threads)
 203         self.assert_cached(check, num_threads)
 204
 205         # The cache expires in the future
 206         with freeze_time("9999-01-01"):
 207             check.run_threads(num_threads)
 208             self.assert_cached(check, 2 * num_threads)
 209             self.assert_cached(check, 2 * num_threads)
 210
 211         # Things are not reverted when we are back to the present
 212         self.assert_cached(check, 2 * num_threads)
 213         self.assert_cached(check, 2 * num_threads)
 214
 215
 216 class TestThreadedCachedPropertyWithTTL(TestThreadedCachedProperty,
 217                                         TestCachedPropertyWithTTL):
 218     """Tests for threaded_cached_property_with_ttl"""
 219
 220     cached_property_factory = cached_property.threaded_cached_property_with_ttl
 221
 222     def test_threads_ttl_expiry(self):
 223         Check = CheckFactory(self.cached_property_factory(ttl=100000),
 224                              threadsafe=True)
 225         check = Check()
 226         num_threads = 5
 227
 228         # Same as in test_threads
 229         check.run_threads(num_threads)
 230         self.assert_cached(check, 1)
 231         self.assert_cached(check, 1)
 232
 233         # The cache expires in the future
 234         with freeze_time("9999-01-01"):
 235             check.run_threads(num_threads)
 236             self.assert_cached(check, 2)
 237             self.assert_cached(check, 2)
 238
 239         # Things are not reverted when we are back to the present
 240         self.assert_cached(check, 2)
 241         self.assert_cached(check, 2)