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