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 def unittest_run_loop(f):
  10
  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 def CheckFactory(cached_property_decorator, threadsafe=False):
  20     """
  21     Create dynamically a Check class whose add_cached method is decorated by
  22     the cached_property_decorator.
  23     """
  24
  25     class Check(object):
  26
  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):
  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 class TestCachedProperty(unittest.TestCase):
  66     """Tests for cached_property"""
  67
  68     cached_property_factory = cached_property.cached_property
  69
  70     async def assert_control(self, check, expected):
  71         """
  72         Assert that both `add_control` and 'control_total` equal `expected`
  73         """
  74         self.assertEqual(await check.add_control(), expected)
  75         self.assertEqual(check.control_total, expected)
  76
  77     async def assert_cached(self, check, expected):
  78         """
  79         Assert that both `add_cached` and 'cached_total` equal `expected`
  80         """
  81         print("assert_cached", check.add_cached)
  82         self.assertEqual(await check.add_cached, expected)
  83         self.assertEqual(check.cached_total, expected)
  84
  85     @unittest_run_loop
  86     async def test_cached_property(self):
  87         Check = CheckFactory(self.cached_property_factory)
  88         check = Check()
  89
  90         # The control shows that we can continue to add 1
  91         await self.assert_control(check, 1)
  92         await self.assert_control(check, 2)
  93
  94         # The cached version demonstrates how nothing is added after the first
  95         await self.assert_cached(check, 1)
  96         await self.assert_cached(check, 1)
  97
  98         # The cache does not expire
  99         with freeze_time("9999-01-01"):
 100             await self.assert_cached(check, 1)
 101
 102         # Typically descriptors return themselves if accessed though the class
 103         # rather than through an instance.
 104         self.assertTrue(isinstance(Check.add_cached, self.cached_property_factory))
 105
 106     @unittest_run_loop
 107     async def test_reset_cached_property(self):
 108         Check = CheckFactory(self.cached_property_factory)
 109         check = Check()
 110
 111         # Run standard cache assertion
 112         await self.assert_cached(check, 1)
 113         await self.assert_cached(check, 1)
 114
 115         # Clear the cache
 116         del check.add_cached
 117
 118         # Value is cached again after the next access
 119         await self.assert_cached(check, 2)
 120         await self.assert_cached(check, 2)
 121
 122     @unittest_run_loop
 123     async def test_none_cached_property(self):
 124
 125         class Check(object):
 126
 127             def __init__(self):
 128                 self.cached_total = None
 129
 130             @self.cached_property_factory
 131             async def add_cached(self):
 132                 return self.cached_total
 133
 134         await self.assert_cached(Check(), None)