Coverage for tests / test_param_cancellable.py: 100%

1import warnings 7 ctx1opqrstu

2 

3import anyio 7 ctx1opqrstu

4import pytest 7 ctx1opqrstu

5from asyncer import asyncify 7 ctx1opqrstu

6 

7 

8def test_cancellable_warns(): 7 ctx1opqrstu

9 def do_async_work(): 7 ctx1abcdefg

10 return "Hello World!" 7 ctx1abcdefg

11 

12 async def main(): 7 ctx1abcdefg

13 result = await asyncify(do_async_work, cancellable=True)() 7 ctx1abcdefg

14 return result 7 ctx1abcdefg

15 

16 with pytest.warns(DeprecationWarning) as record: 7 ctx1abcdefg

17 result = anyio.run(main) 7 ctx1abcdefg

18 assert isinstance(record[0].message, Warning) 7 ctx1abcdefg

19 assert ( 7 ctx1abcdefg

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!" 7 ctx1abcdefg

24 

25 

26def test_abandon_on_cancel_no(): 7 ctx1opqrstu

27 def do_async_work(): 7 ctx1hijklmn

28 return "Hello World!" 7 ctx1hijklmn

29 

30 async def main(): 7 ctx1hijklmn

31 result = await asyncify(do_async_work, abandon_on_cancel=True)() 7 ctx1hijklmn

32 return result 7 ctx1hijklmn

33 

34 with warnings.catch_warnings(): 7 ctx1hijklmn

35 warnings.simplefilter("error") 7 ctx1hijklmn

36 result = anyio.run(main) 7 ctx1hijklmn

37 assert result == "Hello World!" 7 ctx1hijklmn