Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/anyio/abc/_testing.py: 85%

1from __future__ import annotations 

2 

3import types 

4from abc import ABCMeta, abstractmethod 

5from collections.abc import AsyncGenerator, Iterable 

6from typing import Any, Callable, Coroutine, TypeVar 

7 

8_T = TypeVar("_T") 

9 

10 

11class TestRunner(metaclass=ABCMeta): 

12 """ 

13 Encapsulates a running event loop. Every call made through this object will use the same event 

14 loop. 

15 """ 

16 

17 def __enter__(self) -> TestRunner: 

18 return self 

19 

20 def __exit__( 

21 self, 

22 exc_type: type[BaseException] | None, 

23 exc_val: BaseException | None, 

24 exc_tb: types.TracebackType | None, 

25 ) -> bool | None: 

26 self.close() 

27 return None 

28 

29 @abstractmethod 

30 def close(self) -> None: 

31 """Close the event loop.""" 

32 

33 @abstractmethod 

34 def run_asyncgen_fixture( 

35 self, 

36 fixture_func: Callable[..., AsyncGenerator[_T, Any]], 

37 kwargs: dict[str, Any], 

38 ) -> Iterable[_T]: 

39 """ 

40 Run an async generator fixture. 

41 

42 :param fixture_func: the fixture function 

43 :param kwargs: keyword arguments to call the fixture function with 

44 :return: an iterator yielding the value yielded from the async generator 

45 """ 

46 

47 @abstractmethod 

48 def run_fixture( 

49 self, 

50 fixture_func: Callable[..., Coroutine[Any, Any, _T]], 

51 kwargs: dict[str, Any], 

52 ) -> _T: 

53 """ 

54 Run an async fixture. 

55 

56 :param fixture_func: the fixture function 

57 :param kwargs: keyword arguments to call the fixture function with 

58 :return: the return value of the fixture function 

59 """ 

60 

61 @abstractmethod 

62 def run_test( 

63 self, test_func: Callable[..., Coroutine[Any, Any, Any]], kwargs: dict[str, Any] 

64 ) -> None: 

65 """ 

66 Run an async test function. 

67 

68 :param test_func: the test function 

69 :param kwargs: keyword arguments to call the test function with 

70 """