Coverage for tests / conftest.py: 97%

1import asyncio (empty)

2from collections.abc import Generator (empty)

3from typing import Any (empty)

4from unittest.mock import AsyncMock, MagicMock (empty)

5from uuid import uuid4 (empty)

6 

7import pytest (empty)

8from typer.testing import CliRunner (empty)

9 

10from faststream.__about__ import __version__ (empty)

11from faststream._internal.context import ContextRepo (empty)

12 

13 

14@pytest.hookimpl(tryfirst=True) (empty)

15def pytest_keyboard_interrupt( (empty)

16 excinfo: pytest.ExceptionInfo[KeyboardInterrupt], (empty)

17) -> None: # pragma: no cover (empty)

18 pytest.mark.skip("Interrupted Test Session") 

19 

20 

21def pytest_collection_modifyitems(items: list[pytest.Item]) -> None: (empty)

22 for item in items: (empty)

23 item.add_marker("all") (empty)

24 

25 

26@pytest.fixture() (empty)

27def queue() -> str: (empty)

28 return str(uuid4()) (empty)

29 

30 

31@pytest.fixture() (empty)

32def event() -> asyncio.Event: (empty)

33 return asyncio.Event() (empty)

34 

35 

36@pytest.fixture(scope="session") (empty)

37def runner() -> CliRunner: (empty)

38 return CliRunner() (empty)

39 

40 

41@pytest.fixture() (empty)

42def mock() -> Generator[MagicMock, Any, None]: (empty)

43 """Should be generator to share mock between tests.""" 

44 m = MagicMock() (empty)

45 yield m (empty)

46 m.reset_mock() (empty)

47 

48 

49@pytest.fixture() (empty)

50def async_mock() -> Generator[AsyncMock, Any, None]: (empty)

51 """Should be generator to share mock between tests.""" 

52 m = AsyncMock() (empty)

53 yield m (empty)

54 m.reset_mock() (empty)

55 

56 

57@pytest.fixture(scope="session") (empty)

58def version() -> str: (empty)

59 return __version__ (empty)

60 

61 

62@pytest.fixture() (empty)

63def context() -> ContextRepo: (empty)

64 return ContextRepo() (empty)

65 

66 

67@pytest.fixture() (empty)

68def kafka_basic_project() -> str: (empty)

69 return "docs.docs_src.kafka.basic.basic:app" (empty)

70 

71 

72@pytest.fixture() (empty)

73def kafka_ascynapi_project() -> str: (empty)

74 return "docs.docs_src.kafka.basic.basic:asyncapi" 

75 

76 

77@pytest.fixture(autouse=True) (empty)

78def disable_supervisor(monkeypatch): (empty)

79 monkeypatch.setenv("FASTSTREAM_SUPERVISOR_DISABLED", "1") (empty)