Coverage for tests / brokers / base / codec.py: 98%

1from unittest.mock import MagicMock (empty)

2 

3import pytest (empty)

4 

5from faststream._internal.parser import DefaultCodec (empty)

6from tests.brokers.base.basic import BaseTestcaseConfig (empty)

7 

8 

9@pytest.mark.asyncio() (empty)

10class CodecTestcase(BaseTestcaseConfig): (empty)

11 async def test_codec_decode_called( (empty)

12 self, 

13 mock: MagicMock, 

14 queue: str, 

15 ) -> None: 

16 class TrackingCodec(DefaultCodec): (empty)

17 async def decode(self, msg): (empty)

18 mock() (empty)

19 return await super().decode(msg) (empty)

20 

21 codec = TrackingCodec() (empty)

22 broker = self.get_broker() (empty)

23 

24 args, kwargs = self.get_subscriber_params(queue, codec=codec) (empty)

25 

26 @broker.subscriber(*args, **kwargs) (empty)

27 async def handle(m) -> None: (empty)

28 pass (empty)

29 

30 async with self.patch_broker(broker) as br: (empty)

31 await br.publish(b"hello", queue) (empty)

32 

33 mock.assert_called_once() (empty)

34 

35 async def test_codec_not_set_uses_default( (empty)

36 self, 

37 mock: MagicMock, 

38 queue: str, 

39 ) -> None: 

40 broker = self.get_broker() (empty)

41 

42 args, kwargs = self.get_subscriber_params(queue) (empty)

43 

44 @broker.subscriber(*args, **kwargs) (empty)

45 async def handle(m) -> None: (empty)

46 mock(m) (empty)

47 

48 async with self.patch_broker(broker) as br: (empty)

49 await br.publish({"key": "value"}, queue) (empty)

50 

51 mock.assert_called_once_with({"key": "value"}) (empty)

52 

53 async def test_codec_and_decoder_conflict_raises( (empty)

54 self, 

55 queue: str, 

56 ) -> None: 

57 broker = self.get_broker() (empty)

58 codec = DefaultCodec() (empty)

59 

60 async def my_decoder(msg, original): (empty)

61 return await original(msg) 

62 

63 args, kwargs = self.get_subscriber_params(queue, codec=codec, decoder=my_decoder) (empty)

64 

65 @broker.subscriber(*args, **kwargs) (empty)

66 async def handle(m) -> None: (empty)

67 pass # pragma: no cover 

68 

69 # ValueError raised inside _get_parser_and_decoder() during start(), 

70 # which TestBroker.__aenter__ calls before yielding — hence it propagates 

71 # from the "async with" expression rather than from the body. 

72 with pytest.raises(ValueError, match="codec"): (empty)

73 async with self.patch_broker(broker): (empty)

74 pass # pragma: no cover 

75 

76 async def test_broker_level_codec( (empty)

77 self, 

78 mock: MagicMock, 

79 queue: str, 

80 ) -> None: 

81 class TrackingCodec(DefaultCodec): (empty)

82 async def decode(self, msg): (empty)

83 mock() (empty)

84 return await super().decode(msg) (empty)

85 

86 broker = self.get_broker(codec=TrackingCodec()) (empty)

87 

88 args, kwargs = self.get_subscriber_params(queue) (empty)

89 

90 @broker.subscriber(*args, **kwargs) (empty)

91 async def handle(m) -> None: (empty)

92 pass (empty)

93 

94 async with self.patch_broker(broker) as br: (empty)

95 await br.publish(b"hello", queue) (empty)

96 

97 mock.assert_called_once() (empty)