tests/test_async_cached_property.py

   1 # -*- coding: utf-8 -*-
   2 import asyncio
   3 import time
   4 import unittest
   5 from threading import Lock, Thread
   6 from freezegun import freeze_time
   7 import cached_property
   8
   9
  10 def unittest_run_loop(f):
  11     def wrapper(*args, **kwargs):
  12         coro = asyncio.coroutine(f)
  13         future = coro(*args, **kwargs)
  14         loop = asyncio.get_event_loop()
  15         loop.run_until_complete(future)
  16
  17     return wrapper
  18
  19
  20 def CheckFactory(cached_property_decorator, threadsafe=False):
  21     """
  22     Create dynamically a Check class whose add_cached method is decorated by
  23     the cached_property_decorator.
  24     """
  25
  26     class Check(object):
  27         def __init__(self):
  28             self.control_total = 0
  29             self.cached_total = 0
  30             self.lock = Lock()
  31
  32         async def add_control(self): # noqa
  33             self.control_total += 1
  34             return self.control_total
  35
  36         @cached_property_decorator
  37         async def add_cached(self):
  38             if threadsafe:
  39                 time.sleep(1)
  40                 # Need to guard this since += isn't atomic.
  41                 with self.lock:
  42                     self.cached_total += 1
  43             else:
  44                 self.cached_total += 1
  45
  46             return self.cached_total
  47
  48         def run_threads(self, num_threads):
  49             threads = []
  50             for _ in range(num_threads):
  51
  52                 def call_add_cached():
  53                     loop = asyncio.new_event_loop()
  54                     asyncio.set_event_loop(loop)
  55                     loop.run_until_complete(self.add_cached)
  56
  57                 thread = Thread(target=call_add_cached)
  58                 thread.start()
  59                 threads.append(thread)
  60             for thread in threads:
  61                 thread.join()
  62
  63     return Check
  64
  65
  66 class TestCachedProperty(unittest.TestCase):
  67     """Tests for cached_property"""
  68
  69     cached_property_factory = cached_property.cached_property
  70
  71     async def assert_control(self, check, expected):
  72         """
  73         Assert that both `add_control` and 'control_total` equal `expected`
  74         """
  75         self.assertEqual(await check.add_control(), expected)
  76         self.assertEqual(check.control_total, expected)
  77
  78     async def assert_cached(self, check, expected):
  79         """
  80         Assert that both `add_cached` and 'cached_total` equal `expected`
  81         """
  82         print("assert_cached", check.add_cached)
  83         self.assertEqual(await check.add_cached, expected)
  84         self.assertEqual(check.cached_total, expected)
  85
  86     @unittest_run_loop
  87     async def test_cached_property(self):
  88         Check = CheckFactory(self.cached_property_factory)
  89         check = Check()
  90
  91         # The control shows that we can continue to add 1
  92         await self.assert_control(check, 1)
  93         await self.assert_control(check, 2)
  94
  95         # The cached version demonstrates how nothing is added after the first
  96         await self.assert_cached(check, 1)
  97         await self.assert_cached(check, 1)
  98
  99         # The cache does not expire
 100         with freeze_time("9999-01-01"):
 101             await self.assert_cached(check, 1)
 102
 103         # Typically descriptors return themselves if accessed though the class
 104         # rather than through an instance.
 105         self.assertTrue(isinstance(Check.add_cached, self.cached_property_factory))
 106
 107     @unittest_run_loop
 108     async def test_reset_cached_property(self):
 109         Check = CheckFactory(self.cached_property_factory)
 110         check = Check()
 111
 112         # Run standard cache assertion
 113         await self.assert_cached(check, 1)
 114         await self.assert_cached(check, 1)
 115
 116         # Clear the cache
 117         del check.add_cached
 118
 119         # Value is cached again after the next access
 120         await self.assert_cached(check, 2)
 121         await self.assert_cached(check, 2)
 122
 123     @unittest_run_loop
 124     async def test_none_cached_property(self):
 125         class Check(object):
 126             def __init__(self):
 127                 self.cached_total = None
 128
 129             @self.cached_property_factory
 130             async def add_cached(self):
 131                 return self.cached_total
 132
 133         await self.assert_cached(Check(), None)