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, self.cached_property_factory))
  91
  92     def test_reset_cached_property(self):
  93         Check = CheckFactory(self.cached_property_factory)
  94         check = Check()
  95
  96         # Run standard cache assertion
  97         self.assert_cached(check, 1)
  98         self.assert_cached(check, 1)
  99
 100         # Clear the cache
 101         del check.add_cached
 102
 103         # Value is cached again after the next access
 104         self.assert_cached(check, 2)
 105         self.assert_cached(check, 2)
 106
 107     def test_none_cached_property(self):
 108
 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), threadsafe=True)
 196         check = Check()
 197         num_threads = 5
 198
 199         # Same as in test_threads
 200         check.run_threads(num_threads)
 201         self.assert_cached(check, num_threads)
 202         self.assert_cached(check, num_threads)
 203
 204         # The cache expires in the future
 205         with freeze_time("9999-01-01"):
 206             check.run_threads(num_threads)
 207             self.assert_cached(check, 2 * num_threads)
 208             self.assert_cached(check, 2 * num_threads)
 209
 210         # Things are not reverted when we are back to the present
 211         self.assert_cached(check, 2 * num_threads)
 212         self.assert_cached(check, 2 * num_threads)
 213
 214
 215 class TestThreadedCachedPropertyWithTTL(
 216     TestThreadedCachedProperty, TestCachedPropertyWithTTL
 217 ):
 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), threadsafe=True)
 224         check = Check()
 225         num_threads = 5
 226
 227         # Same as in test_threads
 228         check.run_threads(num_threads)
 229         self.assert_cached(check, 1)
 230         self.assert_cached(check, 1)
 231
 232         # The cache expires in the future
 233         with freeze_time("9999-01-01"):
 234             check.run_threads(num_threads)
 235             self.assert_cached(check, 2)
 236             self.assert_cached(check, 2)
 237
 238         # Things are not reverted when we are back to the present
 239         self.assert_cached(check, 2)
 240         self.assert_cached(check, 2)