Coverage for tests / brokers / redis / test_test_client.py: 96%

1import asyncio (empty)

2 

3import pytest (empty)

4 

5from faststream import BaseMiddleware (empty)

6from faststream.redis import ListSub, StreamSub (empty)

7from faststream.redis.testing import FakeProducer (empty)

8from tests.brokers.base.testclient import BrokerTestclientTestcase (empty)

9 

10from .basic import RedisMemoryTestcaseConfig (empty)

11 

12 

13@pytest.mark.redis() (empty)

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

15class TestTestclient(RedisMemoryTestcaseConfig, BrokerTestclientTestcase): (empty)

16 @pytest.mark.connected() (empty)

17 async def test_with_real_testclient( (empty)

18 self, 

19 queue: str, 

20 ) -> None: 

21 event = asyncio.Event() (empty)

22 

23 broker = self.get_broker() (empty)

24 

25 @broker.subscriber(queue) (empty)

26 def subscriber(m) -> None: (empty)

27 event.set() (empty)

28 

29 async with self.patch_broker(broker, with_real=True) as br: (empty)

30 await asyncio.wait( (empty)

31 ( 

32 asyncio.create_task(br.publish("hello", queue)), 

33 asyncio.create_task(event.wait()), 

34 ), 

35 timeout=3, 

36 ) 

37 

38 assert event.is_set() (empty)

39 

40 async def test_respect_middleware(self, queue: str) -> None: (empty)

41 routes = [] (empty)

42 

43 class Middleware(BaseMiddleware): (empty)

44 async def on_receive(self) -> None: (empty)

45 routes.append(None) (empty)

46 return await super().on_receive() (empty)

47 

48 broker = self.get_broker(middlewares=(Middleware,)) (empty)

49 

50 @broker.subscriber(queue) (empty)

51 async def h1(m) -> None: ... (empty)

52 

53 @broker.subscriber(queue + "1") (empty)

54 async def h2(m) -> None: ... (empty)

55 

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

57 await br.publish("", queue) (empty)

58 await br.publish("", queue + "1") (empty)

59 

60 assert len(routes) == 2 (empty)

61 

62 @pytest.mark.connected() (empty)

63 async def test_real_respect_middleware(self, queue: str) -> None: (empty)

64 routes = [] (empty)

65 

66 class Middleware(BaseMiddleware): (empty)

67 async def on_receive(self) -> None: (empty)

68 routes.append(None) (empty)

69 return await super().on_receive() (empty)

70 

71 broker = self.get_broker(middlewares=(Middleware,)) (empty)

72 

73 @broker.subscriber(queue) (empty)

74 async def h1(m) -> None: ... (empty)

75 

76 @broker.subscriber(queue + "1") (empty)

77 async def h2(m) -> None: ... (empty)

78 

79 async with self.patch_broker(broker, with_real=True) as br: (empty)

80 await br.publish("", queue) (empty)

81 await br.publish("", queue + "1") (empty)

82 await h1.wait_call(3) (empty)

83 await h2.wait_call(3) (empty)

84 

85 assert len(routes) == 2 (empty)

86 

87 async def test_pub_sub_pattern(self) -> None: (empty)

88 broker = self.get_broker() (empty)

89 

90 @broker.subscriber("test.{name}") (empty)

91 async def handler(msg): (empty)

92 return msg (empty)

93 

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

95 assert await (await br.request(1, "test.name.useless")).decode() == 1 (empty)

96 handler.mock.assert_called_once_with(1) (empty)

97 

98 async def test_list( (empty)

99 self, 

100 queue: str, 

101 ) -> None: 

102 broker = self.get_broker() (empty)

103 

104 @broker.subscriber(list=queue) (empty)

105 async def handler(msg): (empty)

106 return msg (empty)

107 

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

109 assert await (await br.request(1, list=queue)).decode() == 1 (empty)

110 handler.mock.assert_called_once_with(1) (empty)

111 

112 async def test_batch_pub_by_default_pub( (empty)

113 self, 

114 queue: str, 

115 ) -> None: 

116 broker = self.get_broker() (empty)

117 

118 @broker.subscriber(list=ListSub(queue, batch=True)) (empty)

119 async def m(msg) -> None: (empty)

120 pass (empty)

121 

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

123 await br.publish("hello", list=queue) (empty)

124 m.mock.assert_called_once_with(["hello"]) (empty)

125 

126 async def test_batch_pub_by_pub_batch( (empty)

127 self, 

128 queue: str, 

129 ) -> None: 

130 broker = self.get_broker() (empty)

131 

132 @broker.subscriber(list=ListSub(queue, batch=True)) (empty)

133 async def m(msg) -> None: (empty)

134 pass (empty)

135 

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

137 await br.publish_batch("hello", list=queue) (empty)

138 m.mock.assert_called_once_with(["hello"]) (empty)

139 

140 async def test_batch_publisher_mock( (empty)

141 self, 

142 queue: str, 

143 ) -> None: 

144 broker = self.get_broker() (empty)

145 

146 batch_list = ListSub(queue + "1", batch=True) (empty)

147 publisher = broker.publisher(list=batch_list) (empty)

148 

149 @publisher (empty)

150 @broker.subscriber(queue) (empty)

151 async def m(msg): (empty)

152 return 1, 2, 3 (empty)

153 

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

155 await br.publish("hello", queue) (empty)

156 m.mock.assert_called_once_with("hello") (empty)

157 publisher.mock.assert_called_once_with([1, 2, 3]) (empty)

158 

159 async def test_stream( (empty)

160 self, 

161 queue: str, 

162 ) -> None: 

163 broker = self.get_broker() (empty)

164 

165 @broker.subscriber(stream=queue) (empty)

166 async def handler(msg): (empty)

167 return msg (empty)

168 

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

170 assert await (await br.request(1, stream=queue)).decode() == 1 (empty)

171 handler.mock.assert_called_once_with(1) (empty)

172 

173 async def test_stream_batch_pub_by_default_pub( (empty)

174 self, 

175 queue: str, 

176 ) -> None: 

177 broker = self.get_broker() (empty)

178 

179 @broker.subscriber(stream=StreamSub(queue, batch=True)) (empty)

180 async def m(msg) -> None: (empty)

181 pass (empty)

182 

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

184 await br.publish("hello", stream=queue) (empty)

185 m.mock.assert_called_once_with(["hello"]) (empty)

186 

187 async def test_stream_publisher( (empty)

188 self, 

189 queue: str, 

190 ) -> None: 

191 broker = self.get_broker() (empty)

192 

193 batch_stream = StreamSub(queue + "1") (empty)

194 publisher = broker.publisher(stream=batch_stream) (empty)

195 

196 @publisher (empty)

197 @broker.subscriber(queue) (empty)

198 async def m(msg): (empty)

199 return 1, 2, 3 (empty)

200 

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

202 await br.publish("hello", queue) (empty)

203 m.mock.assert_called_once_with("hello") (empty)

204 publisher.mock.assert_called_once_with([1, 2, 3]) (empty)

205 

206 async def test_publish_to_none(self) -> None: (empty)

207 broker = self.get_broker() (empty)

208 

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

210 with pytest.raises(ValueError): # noqa: PT011 (empty)

211 await br.publish("hello") (empty)

212 

213 @pytest.mark.connected() (empty)

214 async def test_broker_gets_patched_attrs_within_cm(self) -> None: (empty)

215 await super().test_broker_gets_patched_attrs_within_cm(FakeProducer) (empty)

216 

217 @pytest.mark.connected() (empty)

218 async def test_broker_with_real_doesnt_get_patched(self) -> None: (empty)

219 await super().test_broker_with_real_doesnt_get_patched() (empty)

220 

221 @pytest.mark.connected() (empty)

222 async def test_broker_with_real_patches_publishers_and_subscribers( (empty)

223 self, 

224 queue: str, 

225 ) -> None: 

226 await super().test_broker_with_real_patches_publishers_and_subscribers(queue) (empty)

227 

228 @pytest.mark.xfail(reason="https://github.com/ag2ai/faststream/issues/2513") (empty)

229 async def test_publisher_without_destination(self) -> None: (empty)

230 """Fixes https://github.com/ag2ai/faststream/issues/2513.""" 

231 broker = self.get_broker() (empty)

232 

233 # use two publishers to check that we don't have conflicts 

234 channel_publisher = broker.publisher(channel="") (empty)

235 another_channel_publisher = broker.publisher(channel="") (empty)

236 

237 list_publisher = broker.publisher(list="") (empty)

238 stream_publisher = broker.publisher(stream="") (empty)

239 

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

241 await channel_publisher.publish(None, channel="new-key") (empty)

242 channel_publisher.mock.assert_called_once() (empty)

243 

244 await another_channel_publisher.publish(None, channel="new-key") 

245 another_channel_publisher.mock.assert_called_once() 

246 

247 await list_publisher.publish(None, list="new-key") 

248 list_publisher.mock.assert_called_once() 

249 

250 await stream_publisher.publish(None, stream="new-key") 

251 stream_publisher.mock.assert_called_once()