# -*- coding: utf-8 -*-
-"""Tests for cached_property and threaded_cached_property"""
-
-from time import sleep
-from threading import Lock, Thread
+import time
import unittest
+from threading import Lock, Thread
+from freezegun import freeze_time
-from cached_property import cached_property, threaded_cached_property
+import cached_property
-class TestCachedProperty(unittest.TestCase):
- """Tests for cached_property"""
+def CheckFactory(cached_property_decorator, threadsafe=False):
+ """
+ Create dynamically a Check class whose add_cached method is decorated by
+ the cached_property_decorator.
+ """
- cached_property_factory = cached_property
+ class Check(object):
- def test_cached_property(self):
+ def __init__(self):
+ self.control_total = 0
+ self.cached_total = 0
+ self.lock = Lock()
- class Check(object):
+ @property
+ def add_control(self):
+ self.control_total += 1
+ return self.control_total
- def __init__(self):
- self.total1 = 0
- self.total2 = 0
+ @cached_property_decorator
+ def add_cached(self):
+ if threadsafe:
+ time.sleep(1)
+ # Need to guard this since += isn't atomic.
+ with self.lock:
+ self.cached_total += 1
+ else:
+ self.cached_total += 1
- @property
- def add_control(self):
- self.total1 += 1
- return self.total1
+ return self.cached_total
- @self.cached_property_factory
- def add_cached(self):
- self.total2 += 1
- return self.total2
+ def run_threads(self, num_threads):
+ threads = []
+ for _ in range(num_threads):
+ thread = Thread(target=lambda: self.add_cached)
+ thread.start()
+ threads.append(thread)
+ for thread in threads:
+ thread.join()
- c = Check()
+ return Check
- # The control shows that we can continue to add 1.
- self.assertEqual(c.add_control, 1)
- self.assertEqual(c.add_control, 2)
- # The cached version demonstrates how nothing new is added
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.total2, 1)
+class TestCachedProperty(unittest.TestCase):
+ """Tests for cached_property"""
- # It's customary for descriptors to return themselves if accessed
- # though the class, rather than through an instance.
- self.assertTrue(isinstance(Check.add_cached, self.cached_property_factory))
+ cached_property_factory = cached_property.cached_property
- def test_reset_cached_property(self):
+ def assert_control(self, check, expected):
+ """
+ Assert that both `add_control` and 'control_total` equal `expected`
+ """
+ self.assertEqual(check.add_control, expected)
+ self.assertEqual(check.control_total, expected)
- class Check(object):
+ def assert_cached(self, check, expected):
+ """
+ Assert that both `add_cached` and 'cached_total` equal `expected`
+ """
+ self.assertEqual(check.add_cached, expected)
+ self.assertEqual(check.cached_total, expected)
- def __init__(self):
- self.total = 0
+ def test_cached_property(self):
+ Check = CheckFactory(self.cached_property_factory)
+ check = Check()
- @self.cached_property_factory
- def add_cached(self):
- self.total += 1
- return self.total
+ # The control shows that we can continue to add 1
+ self.assert_control(check, 1)
+ self.assert_control(check, 2)
+
+ # The cached version demonstrates how nothing is added after the first
+ self.assert_cached(check, 1)
+ self.assert_cached(check, 1)
- c = Check()
+ # The cache does not expire
+ with freeze_time("9999-01-01"):
+ self.assert_cached(check, 1)
+
+ # Typically descriptors return themselves if accessed though the class
+ # rather than through an instance.
+ self.assertIsInstance(Check.add_cached, self.cached_property_factory)
+
+ def test_reset_cached_property(self):
+ Check = CheckFactory(self.cached_property_factory)
+ check = Check()
# Run standard cache assertion
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.total, 1)
+ self.assert_cached(check, 1)
+ self.assert_cached(check, 1)
- # Reset the cache.
- del c.add_cached
- self.assertEqual(c.add_cached, 2)
- self.assertEqual(c.add_cached, 2)
- self.assertEqual(c.total, 2)
+ # Clear the cache
+ del check.add_cached
- def test_none_cached_property(self):
+ # Value is cached again after the next access
+ self.assert_cached(check, 2)
+ self.assert_cached(check, 2)
+ def test_none_cached_property(self):
class Check(object):
def __init__(self):
- self.total = None
+ self.cached_total = None
@self.cached_property_factory
def add_cached(self):
- return self.total
-
- c = Check()
+ return self.cached_total
- # Run standard cache assertion
- self.assertEqual(c.add_cached, None)
- self.assertEqual(c.total, None)
+ self.assert_cached(Check(), None)
def test_threads(self):
- """
- How well does the standard cached_property implementation work with
- threads? It doesn't, use threaded_cached_property instead!
- """
- num_threads = 10
- check = self._run_threads(num_threads)
- # Threads means that caching is bypassed.
+ Check = CheckFactory(self.cached_property_factory, threadsafe=True)
+ check = Check()
+ num_threads = 5
+
+ # cached_property_with_ttl is *not* thread-safe!
+ check.run_threads(num_threads)
# This assertion hinges on the fact the system executing the test can
# spawn and start running num_threads threads within the sleep period
# (defined in the Check class as 1 second). If num_threads were to be
# massively increased (try 10000), the actual value returned would be
# between 1 and num_threads, depending on thread scheduling and
# preemption.
- self.assertEqual(check.add_cached, num_threads)
- self.assertEqual(check.total, num_threads)
+ self.assert_cached(check, num_threads)
+ self.assert_cached(check, num_threads)
- def _run_threads(self, num_threads):
- class Check(object):
+ # The cache does not expire
+ with freeze_time("9999-01-01"):
+ check.run_threads(num_threads)
+ self.assert_cached(check, num_threads)
+ self.assert_cached(check, num_threads)
- def __init__(self):
- self.total = 0
- self.lock = Lock()
- @self.cached_property_factory
- def add_cached(self):
- sleep(1)
- # Need to guard this since += isn't atomic.
- with self.lock:
- self.total += 1
- return self.total
+class TestThreadedCachedProperty(TestCachedProperty):
+ """Tests for threaded_cached_property"""
+
+ cached_property_factory = cached_property.threaded_cached_property
- c = Check()
+ def test_threads(self):
+ Check = CheckFactory(self.cached_property_factory, threadsafe=True)
+ check = Check()
+ num_threads = 5
- threads = []
- for _ in range(num_threads):
- thread = Thread(target=lambda: c.add_cached)
- thread.start()
- threads.append(thread)
- for thread in threads:
- thread.join()
+ # threaded_cached_property_with_ttl is thread-safe
+ check.run_threads(num_threads)
+ self.assert_cached(check, 1)
+ self.assert_cached(check, 1)
- return c
+ # The cache does not expire
+ with freeze_time("9999-01-01"):
+ check.run_threads(num_threads)
+ self.assert_cached(check, 1)
+ self.assert_cached(check, 1)
-class TestThreadedCachedProperty(TestCachedProperty):
- """Tests for threaded_cached_property"""
+class TestCachedPropertyWithTTL(TestCachedProperty):
+ """Tests for cached_property_with_ttl"""
- cached_property_factory = threaded_cached_property
+ cached_property_factory = cached_property.cached_property_with_ttl
- def test_threads(self):
- """How well does this implementation work with threads?"""
- num_threads = 10
- check = self._run_threads(num_threads)
- self.assertEqual(check.add_cached, 1)
- self.assertEqual(check.total, 1)
+ def test_ttl_expiry(self):
+ Check = CheckFactory(self.cached_property_factory(ttl=100000))
+ check = Check()
+
+ # Run standard cache assertion
+ self.assert_cached(check, 1)
+ self.assert_cached(check, 1)
+
+ # The cache expires in the future
+ with freeze_time("9999-01-01"):
+ self.assert_cached(check, 2)
+ self.assert_cached(check, 2)
+
+ # Things are not reverted when we are back to the present
+ self.assert_cached(check, 2)
+ self.assert_cached(check, 2)
+
+ def test_threads_ttl_expiry(self):
+ Check = CheckFactory(self.cached_property_factory(ttl=100000),
+ threadsafe=True)
+ check = Check()
+ num_threads = 5
+
+ # Same as in test_threads
+ check.run_threads(num_threads)
+ self.assert_cached(check, num_threads)
+ self.assert_cached(check, num_threads)
+
+ # The cache expires in the future
+ with freeze_time("9999-01-01"):
+ check.run_threads(num_threads)
+ self.assert_cached(check, 2 * num_threads)
+ self.assert_cached(check, 2 * num_threads)
+
+ # Things are not reverted when we are back to the present
+ self.assert_cached(check, 2 * num_threads)
+ self.assert_cached(check, 2 * num_threads)
+
+
+class TestThreadedCachedPropertyWithTTL(TestThreadedCachedProperty,
+ TestCachedPropertyWithTTL):
+ """Tests for threaded_cached_property_with_ttl"""
+
+ cached_property_factory = cached_property.threaded_cached_property_with_ttl
+
+ def test_threads_ttl_expiry(self):
+ Check = CheckFactory(self.cached_property_factory(ttl=100000),
+ threadsafe=True)
+ check = Check()
+ num_threads = 5
+
+ # Same as in test_threads
+ check.run_threads(num_threads)
+ self.assert_cached(check, 1)
+ self.assert_cached(check, 1)
+
+ # The cache expires in the future
+ with freeze_time("9999-01-01"):
+ check.run_threads(num_threads)
+ self.assert_cached(check, 2)
+ self.assert_cached(check, 2)
+
+ # Things are not reverted when we are back to the present
+ self.assert_cached(check, 2)
+ self.assert_cached(check, 2)
+++ /dev/null
-# -*- coding: utf-8 -*-
-
-"""Tests for cached_property_with_ttl and threaded_cache_property_with_ttl"""
-
-import unittest
-from freezegun import freeze_time
-from time import sleep
-from threading import Lock, Thread
-
-from cached_property import (
- cached_property_with_ttl,
- threaded_cached_property_with_ttl
-)
-
-
-class TestCachedPropertyWithTTL(unittest.TestCase):
-
- def test_cached_property(self):
-
- class Check(object):
-
- def __init__(self):
- self.total1 = 0
- self.total2 = 0
-
- @property
- def add_control(self):
- self.total1 += 1
- return self.total1
-
- @cached_property_with_ttl
- def add_cached(self):
- self.total2 += 1
- return self.total2
-
- c = Check()
-
- # The control shows that we can continue to add 1.
- self.assertEqual(c.add_control, 1)
- self.assertEqual(c.add_control, 2)
-
- # The cached version demonstrates how nothing new is added
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.add_cached, 1)
-
- # Cannot expire the cache.
- with freeze_time("9999-01-01"):
- self.assertEqual(c.add_cached, 1)
-
- # It's customary for descriptors to return themselves if accessed
- # though the class, rather than through an instance.
- self.assertTrue(isinstance(Check.add_cached, cached_property_with_ttl))
-
- def test_reset_cached_property(self):
-
- class Check(object):
-
- def __init__(self):
- self.total = 0
-
- @cached_property_with_ttl
- def add_cached(self):
- self.total += 1
- return self.total
-
- c = Check()
-
- # Resetting the cache before it is set is a no-op
- del c.add_cached
-
- # Run standard cache assertion
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.add_cached, 1)
-
- # Reset the cache.
- del c.add_cached
- self.assertEqual(c.add_cached, 2)
- self.assertEqual(c.add_cached, 2)
-
- def test_none_cached_property(self):
-
- class Check(object):
-
- def __init__(self):
- self.total = None
-
- @cached_property_with_ttl
- def add_cached(self):
- return self.total
-
- c = Check()
-
- # Run standard cache assertion
- self.assertEqual(c.add_cached, None)
-
- def test_threads(self):
- """ How well does the standard cached_property implementation work with threads?
- Short answer: It doesn't! Use threaded_cached_property instead!
- """ # noqa
-
- class Check(object):
-
- def __init__(self):
- self.total = 0
- self.lock = Lock()
-
- @cached_property_with_ttl
- def add_cached(self):
- sleep(1)
- # Need to guard this since += isn't atomic.
- with self.lock:
- self.total += 1
- return self.total
-
- c = Check()
- threads = []
- num_threads = 10
- for x in range(num_threads):
- thread = Thread(target=lambda: c.add_cached)
- thread.start()
- threads.append(thread)
-
- for thread in threads:
- thread.join()
-
- # Threads means that caching is bypassed.
- self.assertNotEqual(c.add_cached, 1)
-
- # This assertion hinges on the fact the system executing the test can
- # spawn and start running num_threads threads within the sleep period
- # (defined in the Check class as 1 second). If num_threads were to be
- # massively increased (try 10000), the actual value returned would be
- # between 1 and num_threads, depending on thread scheduling and
- # preemption.
- self.assertEqual(c.add_cached, num_threads)
-
- def test_ttl_expiry(self):
-
- class Check(object):
-
- def __init__(self):
- self.total = 0
-
- @cached_property_with_ttl(ttl=100000)
- def add_cached(self):
- self.total += 1
- return self.total
-
- c = Check()
-
- # Run standard cache assertion
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.add_cached, 1)
-
- # Expire the cache.
- with freeze_time("9999-01-01"):
- self.assertEqual(c.add_cached, 2)
- self.assertEqual(c.add_cached, 2)
-
-
-class TestThreadedCachedPropertyWithTTL(unittest.TestCase):
-
- def test_cached_property(self):
-
- class Check(object):
-
- def __init__(self):
- self.total1 = 0
- self.total2 = 0
-
- @property
- def add_control(self):
- self.total1 += 1
- return self.total1
-
- @threaded_cached_property_with_ttl
- def add_cached(self):
- self.total2 += 1
- return self.total2
-
- c = Check()
-
- # The control shows that we can continue to add 1.
- self.assertEqual(c.add_control, 1)
- self.assertEqual(c.add_control, 2)
-
- # The cached version demonstrates how nothing new is added
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.add_cached, 1)
-
- # Cannot expire the cache.
- with freeze_time("9999-01-01"):
- self.assertEqual(c.add_cached, 1)
-
- # It's customary for descriptors to return themselves if accessed
- # though the class, rather than through an instance.
- self.assertTrue(isinstance(Check.add_cached, threaded_cached_property_with_ttl))
-
- def test_reset_cached_property(self):
-
- class Check(object):
-
- def __init__(self):
- self.total = 0
-
- @threaded_cached_property_with_ttl
- def add_cached(self):
- self.total += 1
- return self.total
-
- c = Check()
-
- # Resetting the cache before it is set is a no-op
- del c.add_cached
-
- # Run standard cache assertion
- self.assertEqual(c.add_cached, 1)
- self.assertEqual(c.add_cached, 1)
-
- # Reset the cache.
- del c.add_cached
- self.assertEqual(c.add_cached, 2)
- self.assertEqual(c.add_cached, 2)
-
- def test_none_cached_property(self):
-
- class Check(object):
-
- def __init__(self):
- self.total = None
-
- @threaded_cached_property_with_ttl
- def add_cached(self):
- return self.total
-
- c = Check()
-
- # Run standard cache assertion
- self.assertEqual(c.add_cached, None)
-
- def test_threads(self):
- """ How well does this implementation work with threads?"""
-
- class Check(object):
-
- def __init__(self):
- self.total = 0
- self.lock = Lock()
-
- @threaded_cached_property_with_ttl
- def add_cached(self):
- sleep(1)
- # Need to guard this since += isn't atomic.
- with self.lock:
- self.total += 1
- return self.total
-
- c = Check()
- threads = []
- for x in range(10):
- thread = Thread(target=lambda: c.add_cached)
- thread.start()
- threads.append(thread)
-
- for thread in threads:
- thread.join()
-
- self.assertEqual(c.add_cached, 1)
-
- def test_ttl_expiry(self):
-
- class Check(object):
-
- def __init__(self):
- self.total = 0
- self.lock = Lock()
-
- @threaded_cached_property_with_ttl(ttl=100000)
- def add_cached(self):
- sleep(1)
- # Need to guard this since += isn't atomic.
- with self.lock:
- self.total += 1
- return self.total
-
- def run_threads(check, num_threads=10):
- threads = []
- for _ in range(num_threads):
- thread = Thread(target=lambda: check.add_cached)
- thread.start()
- threads.append(thread)
- for thread in threads:
- thread.join()
-
- c = Check()
- run_threads(c)
- self.assertEqual(c.add_cached, 1)
-
- # Expire the cache.
- with freeze_time("9999-01-01"):
- run_threads(c)
- self.assertEqual(c.add_cached, 2)
-
- self.assertEqual(c.add_cached, 2)
projects / cached-property.git / commitdiff