daily pastebin goal
2%
SHARE
TWEET

Untitled

a guest Feb 17th, 2019 64 Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
  1. import asyncio
  2.  
  3. import pytest
  4.  
  5.  
  6. class Mutable:
  7.     def __init__(self) -> None:
  8.         self._mutations: int = 0
  9.    
  10.     @property
  11.     def mutations(self) -> int:
  12.         return self._mutations
  13.    
  14.     def mutate(self) -> None:
  15.         self._mutations += 1
  16.    
  17.     async def a_mutate(self, delay: float = 0) -> None:
  18.         await asyncio.sleep(delay)
  19.         self._mutations += 1
  20.  
  21.  
  22. async def wait_then_call(f, delay: float = 0.2):
  23.     await asyncio.sleep(delay)
  24.     await f()
  25.  
  26.  
  27. async def wait_for_cancellation(f, delay: float = 0.2, reraise = False):
  28.     await asyncio.sleep(delay)
  29.     try:
  30.         await f
  31.     except asyncio.CancelledError:
  32.         if reraise:
  33.             raise
  34.     else:
  35.         raise AssertionError(f"Did not raise {asyncio.CancelledError}")
  36.  
  37.  
  38. def test_sync_mutations():
  39.     m = Mutable()
  40.     n = m
  41.     assert n.mutations == 0
  42.     n.mutate()
  43.     assert m.mutations == 1
  44.     m.mutate()
  45.     assert n.mutations == 2
  46.  
  47.  
  48. @pytest.mark.asyncio
  49. async def test_async_mutations():
  50.     m = Mutable()
  51.     t = asyncio.create_task(m.a_mutate(delay=0.5))
  52.     await asyncio.sleep(0.2)
  53.     await asyncio.sleep(0.5)
  54.     await t
  55.     assert m.mutations == 1
  56.  
  57.  
  58. @pytest.mark.asyncio
  59. async def test_cancellation():
  60.     m = Mutable()
  61.     t = asyncio.create_task(m.a_mutate(delay=0.5))
  62.     with pytest.raises(asyncio.CancelledError):
  63.         await asyncio.sleep(0.2)
  64.         t.cancel()
  65.         await asyncio.sleep(0.5)
  66.         await t
  67.     assert m.mutations == 0
  68.  
  69.  
  70. @pytest.mark.asyncio
  71. async def test_deep_cancellation():
  72.     """
  73.     w2 will encounter a CancelledError -- caused by
  74.     t's cancellation -- and re-raise it to w2, who
  75.     will handle it silently.
  76.     """
  77.     w2 = wait_for_cancellation(asyncio.sleep(0.3), reraise=True)
  78.     w1 = wait_for_cancellation(w2)
  79.     t = asyncio.create_task(w1)
  80.     await asyncio.sleep(0.5)
  81.     t.cancel()
  82.     await t
  83.  
  84.  
  85. @pytest.mark.asyncio
  86. async def test_shielding():
  87.     """
  88.     t's cancellation will _not_ cause s to encounter a
  89.     CancelledError, but w will catch a CancelledError at
  90.     s's call-site.
  91.    
  92.     when awaited, s will finish their action, mutating m.
  93.     """
  94.     m = Mutable()
  95.     s = asyncio.shield(m.a_mutate(delay=0.2))
  96.     w = wait_for_cancellation(s)
  97.     t = asyncio.create_task(w)
  98.     await asyncio.sleep(0.1)
  99.     t.cancel()
  100.     assert m.mutations == 0
  101.     await s
  102.     assert m.mutations == 1
  103.  
  104.  
  105. @pytest.mark.asyncio
  106. async def test_shielding_propagation():
  107.     """
  108.     t's cancellation will not cause s to encounter a
  109.     CancelledError, and since s is awaiting u, u won't
  110.     be cancelled either. w will handle the cancellation
  111.     from s's callsite.
  112.     """
  113.     m = Mutable()
  114.     u = m.a_mutate(delay=0.2)
  115.     s = asyncio.shield(asyncio.gather(m.a_mutate(delay=0.2), u))
  116.     w = wait_for_cancellation(s)
  117.     t = asyncio.create_task(w)
  118.     await asyncio.sleep(0.1)
  119.     t.cancel()
  120.     assert m.mutations == 0
  121.     await s
  122.     assert m.mutations == 2
RAW Paste Data
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy. OK, I Understand
 
Top