Coverage for tests / brokers / confluent / test_test_client.py: 99%

1import asyncio (empty)

2from unittest.mock import AsyncMock, patch (empty)

3 

4import pytest (empty)

5 

6from faststream import AckPolicy, BaseMiddleware (empty)

7from faststream.confluent.annotations import KafkaMessage (empty)

8from faststream.confluent.message import FAKE_CONSUMER (empty)

9from faststream.confluent.testing import FakeProducer (empty)

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

11from tests.tools import spy_decorator (empty)

12 

13from .basic import ConfluentMemoryTestcaseConfig (empty)

14 

15 

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

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

18class TestTestclient(ConfluentMemoryTestcaseConfig, BrokerTestclientTestcase): (empty)

19 async def test_message_nack_seek(self, queue: str) -> None: (empty)

20 broker = self.get_broker(apply_types=True) (empty)

21 

22 @broker.subscriber( (empty)

23 queue, 

24 group_id=f"{queue}-consume", 

25 auto_offset_reset="earliest", 

26 ack_policy=AckPolicy.REJECT_ON_ERROR, 

27 ) 

28 async def m(msg: KafkaMessage) -> None: (empty)

29 await msg.nack() (empty)

30 

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

32 with patch.object( (empty)

33 FAKE_CONSUMER, 

34 "seek", 

35 spy_decorator(FAKE_CONSUMER.seek), 

36 ) as mocked: 

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

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

39 mocked.mock.assert_called_once() (empty)

40 

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

42 async def test_with_real_testclient(self, queue: str) -> None: (empty)

43 event = asyncio.Event() (empty)

44 

45 broker = self.get_broker() (empty)

46 

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

48 

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

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

51 event.set() (empty)

52 

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

54 await asyncio.wait( (empty)

55 ( 

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

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

58 ), 

59 timeout=10, 

60 ) 

61 

62 assert event.is_set() (empty)

63 

64 async def test_publisher_autoflush_mock(self, queue: str) -> None: (empty)

65 broker = self.get_broker() (empty)

66 

67 publisher = broker.publisher(queue + "1", autoflush=True) (empty)

68 publisher.flush = AsyncMock() (empty)

69 

70 @publisher (empty)

71 @broker.subscriber(queue) (empty)

72 async def m(msg): (empty)

73 pass (empty)

74 

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

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

77 

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

79 publisher.mock.assert_called_once() (empty)

80 

81 publisher.flush.assert_awaited_once() (empty)

82 

83 async def test_batch_publisher_autoflush_mock(self, queue: str) -> None: (empty)

84 broker = self.get_broker() (empty)

85 

86 publisher = broker.publisher(queue + "1", batch=True, autoflush=True) (empty)

87 publisher.flush = AsyncMock() (empty)

88 

89 @publisher (empty)

90 @broker.subscriber(queue) (empty)

91 async def m(msg): (empty)

92 return 1, 2, 3 (empty)

93 

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

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

96 

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

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

99 

100 publisher.flush.assert_awaited_once() (empty)

101 

102 async def test_batch_pub_by_default_pub( (empty)

103 self, 

104 queue: str, 

105 ) -> None: 

106 broker = self.get_broker() (empty)

107 

108 @broker.subscriber(queue, batch=True) (empty)

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

110 pass (empty)

111 

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

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

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

115 

116 async def test_batch_pub_by_pub_batch( (empty)

117 self, 

118 queue: str, 

119 ) -> None: 

120 broker = self.get_broker() (empty)

121 

122 @broker.subscriber(queue, batch=True) (empty)

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

124 pass (empty)

125 

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

127 await br.publish_batch("hello", topic=queue) (empty)

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

129 

130 async def test_batch_publisher_mock(self, queue: str) -> None: (empty)

131 broker = self.get_broker() (empty)

132 

133 publisher = broker.publisher(queue + "1", batch=True) (empty)

134 

135 @publisher (empty)

136 @broker.subscriber(queue) (empty)

137 async def m(msg): (empty)

138 return 1, 2, 3 (empty)

139 

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

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

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

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

144 

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

146 routes = [] (empty)

147 

148 class Middleware(BaseMiddleware): (empty)

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

150 routes.append(None) (empty)

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

152 

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

154 

155 @broker.subscriber(queue) (empty)

156 async def h1(msg) -> None: ... (empty)

157 

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

159 async def h2(msg) -> None: ... (empty)

160 

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

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

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

164 

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

166 

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

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

169 routes = [] (empty)

170 

171 class Middleware(BaseMiddleware): (empty)

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

173 routes.append(None) (empty)

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

175 

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

177 

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

179 

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

181 async def h1(msg) -> None: ... (empty)

182 

183 args2, kwargs2 = self.get_subscriber_params(queue + "1") (empty)

184 

185 @broker.subscriber(*args2, **kwargs2) (empty)

186 async def h2(msg) -> None: ... (empty)

187 

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

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

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

191 await h1.wait_call(10) (empty)

192 await h2.wait_call(10) (empty)

193 

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

195 

196 async def test_multiple_subscribers_different_groups(self, queue: str) -> None: (empty)

197 broker = self.get_broker() (empty)

198 

199 @broker.subscriber(queue, group_id="group1") (empty)

200 async def subscriber1(msg) -> None: ... (empty)

201 

202 @broker.subscriber(queue, group_id="group2") (empty)

203 async def subscriber2(msg) -> None: ... (empty)

204 

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

206 await br.start() (empty)

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

208 

209 assert subscriber1.mock.call_count == 1 (empty)

210 assert subscriber2.mock.call_count == 1 (empty)

211 

212 async def test_multiple_subscribers_same_group(self, queue: str) -> None: (empty)

213 broker = self.get_broker() (empty)

214 

215 @broker.subscriber(queue, group_id="group1") (empty)

216 async def subscriber1(msg) -> None: ... (empty)

217 

218 @broker.subscriber(queue, group_id="group1") (empty)

219 async def subscriber2(msg) -> None: ... (empty)

220 

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

222 await br.start() (empty)

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

224 

225 # we can't guarantee the order of calls 

226 assert {subscriber1.mock.call_count, subscriber2.mock.call_count} == {1, 0} (empty)

227 

228 async def test_multiple_batch_subscriber_with_different_group( (empty)

229 self, queue: str 

230 ) -> None: 

231 broker = self.get_broker() (empty)

232 

233 @broker.subscriber(queue, batch=True, group_id="group1") (empty)

234 async def subscriber1(msg) -> None: ... (empty)

235 

236 @broker.subscriber(queue, batch=True, group_id="group2") (empty)

237 async def subscriber2(msg) -> None: ... (empty)

238 

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

240 await br.start() (empty)

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

242 

243 assert subscriber1.mock.call_count == 1 (empty)

244 assert subscriber2.mock.call_count == 1 (empty)

245 

246 async def test_multiple_batch_subscriber_with_same_group(self, queue: str) -> None: (empty)

247 broker = self.get_broker() (empty)

248 

249 @broker.subscriber(queue, batch=True, group_id="group1") (empty)

250 async def subscriber1(msg) -> None: ... (empty)

251 

252 @broker.subscriber(queue, batch=True, group_id="group1") (empty)

253 async def subscriber2(msg) -> None: ... (empty)

254 

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

256 await br.start() (empty)

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

258 

259 # we can't guarantee the order of calls 

260 assert {subscriber1.mock.call_count, subscriber2.mock.call_count} == {1, 0} (empty)

261 

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

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

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

265 

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

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

268 await super().test_broker_with_real_doesnt_get_patched() (empty)

269 

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

271 async def test_broker_with_real_patches_publishers_and_subscribers( (empty)

272 self, queue: str 

273 ) -> None: 

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

275 

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

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

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

279 broker = self.get_broker() (empty)

280 

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

282 publisher = broker.publisher(topic="") (empty)

283 another_publisher = broker.publisher(topic="") (empty)

284 

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

286 await publisher.publish(None, topic="new-key") (empty)

287 publisher.mock.assert_called_once() (empty)

288 

289 await another_publisher.publish(None, topic="new-key") 

290 another_publisher.mock.assert_called_once()