From 4f4225138a9f92e35c2d2a84176efdd2901de92d Mon Sep 17 00:00:00 2001 From: Daniel Roy Greenfeld Date: Wed, 4 Apr 2018 22:03:02 -0500 Subject: [PATCH] Don't test asyncio on older versions of Python Per #93, asyncio tests need to be bypassed for older versions of Python --- tests/test_async_cached_property.py | 228 ++++++++++++++-------------- 1 file changed, 117 insertions(+), 111 deletions(-) diff --git a/tests/test_async_cached_property.py b/tests/test_async_cached_property.py index 36892be..6a9fb87 100644 --- a/tests/test_async_cached_property.py +++ b/tests/test_async_cached_property.py @@ -1,135 +1,141 @@ # -*- coding: utf-8 -*- -import time -import unittest -import asyncio -from threading import Lock, Thread -from freezegun import freeze_time - -import cached_property - - -def unittest_run_loop(f): - def wrapper(*args, **kwargs): - coro = asyncio.coroutine(f) - future = coro(*args, **kwargs) - loop = asyncio.get_event_loop() - loop.run_until_complete(future) - return wrapper - - -def CheckFactory(cached_property_decorator, threadsafe=False): - """ - Create dynamically a Check class whose add_cached method is decorated by - the cached_property_decorator. - """ - - class Check(object): - - def __init__(self): - self.control_total = 0 - self.cached_total = 0 - self.lock = Lock() - - async def add_control(self): - self.control_total += 1 - return self.control_total - - @cached_property_decorator - async 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 - - return self.cached_total - - def run_threads(self, num_threads): - threads = [] - for _ in range(num_threads): - def call_add_cached(): - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - loop.run_until_complete(self.add_cached) - thread = Thread(target=call_add_cached) - thread.start() - threads.append(thread) - for thread in threads: - thread.join() - return Check +try: + import asyncio + import time + import unittest + from threading import Lock, Thread + from freezegun import freeze_time + import cached_property -class TestCachedProperty(unittest.TestCase): - """Tests for cached_property""" + def unittest_run_loop(f): + def wrapper(*args, **kwargs): + coro = asyncio.coroutine(f) + future = coro(*args, **kwargs) + loop = asyncio.get_event_loop() + loop.run_until_complete(future) + return wrapper - cached_property_factory = cached_property.cached_property - async def assert_control(self, check, expected): + def CheckFactory(cached_property_decorator, threadsafe=False): """ - Assert that both `add_control` and 'control_total` equal `expected` + Create dynamically a Check class whose add_cached method is decorated by + the cached_property_decorator. """ - self.assertEqual(await check.add_control(), expected) - self.assertEqual(check.control_total, expected) - async def assert_cached(self, check, expected): - """ - Assert that both `add_cached` and 'cached_total` equal `expected` - """ - print('assert_cached', check.add_cached) - self.assertEqual(await check.add_cached, expected) - self.assertEqual(check.cached_total, expected) + class Check(object): + + def __init__(self): + self.control_total = 0 + self.cached_total = 0 + self.lock = Lock() - @unittest_run_loop - async def test_cached_property(self): - Check = CheckFactory(self.cached_property_factory) - check = Check() + async def add_control(self): + self.control_total += 1 + return self.control_total - # The control shows that we can continue to add 1 - await self.assert_control(check, 1) - await self.assert_control(check, 2) + @cached_property_decorator + async 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 - # The cached version demonstrates how nothing is added after the first - await self.assert_cached(check, 1) - await self.assert_cached(check, 1) + return self.cached_total - # The cache does not expire - with freeze_time("9999-01-01"): + def run_threads(self, num_threads): + threads = [] + for _ in range(num_threads): + def call_add_cached(): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + loop.run_until_complete(self.add_cached) + thread = Thread(target=call_add_cached) + thread.start() + threads.append(thread) + for thread in threads: + thread.join() + + return Check + + + class TestCachedProperty(unittest.TestCase): + """Tests for cached_property""" + + cached_property_factory = cached_property.cached_property + + async def assert_control(self, check, expected): + """ + Assert that both `add_control` and 'control_total` equal `expected` + """ + self.assertEqual(await check.add_control(), expected) + self.assertEqual(check.control_total, expected) + + async def assert_cached(self, check, expected): + """ + Assert that both `add_cached` and 'cached_total` equal `expected` + """ + print('assert_cached', check.add_cached) + self.assertEqual(await check.add_cached, expected) + self.assertEqual(check.cached_total, expected) + + @unittest_run_loop + async def test_cached_property(self): + Check = CheckFactory(self.cached_property_factory) + check = Check() + + # The control shows that we can continue to add 1 + await self.assert_control(check, 1) + await self.assert_control(check, 2) + + # The cached version demonstrates how nothing is added after the first + await self.assert_cached(check, 1) await self.assert_cached(check, 1) - # Typically descriptors return themselves if accessed though the class - # rather than through an instance. - self.assertTrue(isinstance(Check.add_cached, - self.cached_property_factory)) + # The cache does not expire + with freeze_time("9999-01-01"): + await self.assert_cached(check, 1) - @unittest_run_loop - async def test_reset_cached_property(self): - Check = CheckFactory(self.cached_property_factory) - check = Check() + # Typically descriptors return themselves if accessed though the class + # rather than through an instance. + self.assertTrue(isinstance(Check.add_cached, + self.cached_property_factory)) - # Run standard cache assertion - await self.assert_cached(check, 1) - await self.assert_cached(check, 1) + @unittest_run_loop + async def test_reset_cached_property(self): + Check = CheckFactory(self.cached_property_factory) + check = Check() - # Clear the cache - del check.add_cached + # Run standard cache assertion + await self.assert_cached(check, 1) + await self.assert_cached(check, 1) - # Value is cached again after the next access - await self.assert_cached(check, 2) - await self.assert_cached(check, 2) + # Clear the cache + del check.add_cached - @unittest_run_loop - async def test_none_cached_property(self): - class Check(object): + # Value is cached again after the next access + await self.assert_cached(check, 2) + await self.assert_cached(check, 2) - def __init__(self): - self.cached_total = None + @unittest_run_loop + async def test_none_cached_property(self): + class Check(object): - @self.cached_property_factory - async def add_cached(self): - return self.cached_total + def __init__(self): + self.cached_total = None + + @self.cached_property_factory + async def add_cached(self): + return self.cached_total + + await self.assert_cached(Check(), None) - await self.assert_cached(Check(), None) + +except ImportError: + pass # Running older version of Python that doesn't support asyncio + -- 2.30.2