Coverage for tests/test_param_cancellable.py: 100%

1import warnings 13 ctx1ABCDEFGHIJKLM

2 

3import anyio 13 ctx1ABCDEFGHIJKLM

4import pytest 13 ctx1ABCDEFGHIJKLM

5from asyncer import asyncify 13 ctx1ABCDEFGHIJKLM

6 

7 

8def test_cancellable_warns(): 13 ctx1ABCDEFGHIJKLM

9 def do_async_work(): 13 ctx1abcdefghijklm

10 return "Hello World!" 13 ctx1abcdefghijklm

11 

12 async def main(): 13 ctx1abcdefghijklm

13 result = await asyncify(do_async_work, cancellable=True)() 13 ctx1abcdefghijklm

14 return result 13 ctx1abcdefghijklm

15 

16 with pytest.warns(DeprecationWarning) as record: 13 ctx1abcdefghijklm

17 result = anyio.run(main) 13 ctx1abcdefghijklm

18 assert isinstance(record[0].message, Warning) 13 ctx1abcdefghijklm

19 assert ( 13 ctx1abcdefghijklm

20 "The `cancellable=` keyword argument to `asyncer.asyncify()` is " 

21 "deprecated since Asyncer 0.0.8" in record[0].message.args[0] 

22 ) 

23 assert result == "Hello World!" 13 ctx1abcdefghijklm

24 

25 

26def test_abandon_on_cancel_no(): 13 ctx1ABCDEFGHIJKLM

27 def do_async_work(): 13 ctx1nopqrstuvwxyz

28 return "Hello World!" 13 ctx1nopqrstuvwxyz

29 

30 async def main(): 13 ctx1nopqrstuvwxyz

31 result = await asyncify(do_async_work, abandon_on_cancel=True)() 13 ctx1nopqrstuvwxyz

32 return result 13 ctx1nopqrstuvwxyz

33 

34 with warnings.catch_warnings(): 13 ctx1nopqrstuvwxyz

35 warnings.simplefilter("error") 13 ctx1nopqrstuvwxyz

36 result = anyio.run(main) 13 ctx1nopqrstuvwxyz

37 assert result == "Hello World!" 13 ctx1nopqrstuvwxyz