tests/test_async_cached_property.py

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