Coverage for tests / brokers / rabbit / test_test_client.py: 97%

1import asyncio (empty)

2import datetime as dt (empty)

3from typing import Any (empty)

4from unittest.mock import MagicMock (empty)

5 

6import pytest (empty)

7from freezegun import freeze_time (empty)

8 

9from faststream import BaseMiddleware (empty)

10from faststream.exceptions import SubscriberNotFound (empty)

11from faststream.rabbit import ( (empty)

12 ExchangeType, 

13 RabbitBroker, 

14 RabbitExchange, 

15 RabbitQueue, 

16) 

17from faststream.rabbit.annotations import RabbitMessage (empty)

18from faststream.rabbit.testing import FakeProducer, _is_handler_matches, apply_pattern (empty)

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

20 

21from .basic import RabbitMemoryTestcaseConfig (empty)

22from .test_publish import TestPublishWithExchange as PublishWithExchangeCase (empty)

23 

24_frozen_time = dt.datetime(2026, 2, 10, 12, 0, 0, tzinfo=dt.timezone.utc) (empty)

25 

26 

27@pytest.mark.rabbit() (empty)

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

29class TestTestclient( (empty)

30 PublishWithExchangeCase, RabbitMemoryTestcaseConfig, BrokerTestclientTestcase 

31): 

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

33 async def test_with_real_testclient( (empty)

34 self, 

35 queue: str, 

36 ) -> None: 

37 event = asyncio.Event() (empty)

38 

39 broker = self.get_broker() (empty)

40 

41 @broker.subscriber(queue) (empty)

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

43 event.set() (empty)

44 

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

46 await asyncio.wait( (empty)

47 ( 

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

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

50 ), 

51 timeout=3, 

52 ) 

53 

54 assert event.is_set() (empty)

55 

56 async def test_direct_not_found( (empty)

57 self, 

58 queue: str, 

59 ) -> None: 

60 broker = self.get_broker() (empty)

61 

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

63 with pytest.raises(SubscriberNotFound): (empty)

64 await br.request("", "") (empty)

65 

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

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

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

69 broker = self.get_broker() (empty)

70 

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

72 publisher = broker.publisher(exchange="test_exchange") (empty)

73 another_publisher = broker.publisher(exchange="test_exchange") (empty)

74 

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

76 await publisher.publish(None, routing_key="new-key") (empty)

77 publisher.mock.assert_called_once() 

78 

79 await another_publisher.publish(None, routing_key="new-key") 

80 another_publisher.mock.assert_called_once() 

81 

82 async def test_consume_manual_ack( (empty)

83 self, 

84 queue: str, 

85 exchange: RabbitExchange, 

86 ) -> None: 

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

88 

89 consume = asyncio.Event() (empty)

90 consume2 = asyncio.Event() (empty)

91 consume3 = asyncio.Event() (empty)

92 

93 @broker.subscriber(queue=queue, exchange=exchange) (empty)

94 async def handler(msg: RabbitMessage) -> None: (empty)

95 await msg.raw_message.ack() (empty)

96 consume.set() (empty)

97 

98 @broker.subscriber(queue=queue + "1", exchange=exchange) (empty)

99 async def handler2(msg: RabbitMessage) -> None: (empty)

100 await msg.raw_message.nack() (empty)

101 consume2.set() (empty)

102 raise ValueError (empty)

103 

104 @broker.subscriber(queue=queue + "2", exchange=exchange) (empty)

105 async def handler3(msg: RabbitMessage) -> None: (empty)

106 await msg.raw_message.reject() (empty)

107 consume3.set() (empty)

108 raise ValueError (empty)

109 

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

111 await asyncio.wait( (empty)

112 ( 

113 asyncio.create_task( 

114 br.publish("hello", queue=queue, exchange=exchange), 

115 ), 

116 asyncio.create_task( 

117 br.publish("hello", queue=queue + "1", exchange=exchange), 

118 ), 

119 asyncio.create_task( 

120 br.publish("hello", queue=queue + "2", exchange=exchange), 

121 ), 

122 asyncio.create_task(consume.wait()), 

123 asyncio.create_task(consume2.wait()), 

124 asyncio.create_task(consume3.wait()), 

125 ), 

126 timeout=3, 

127 ) 

128 

129 assert consume.is_set() (empty)

130 assert consume2.is_set() (empty)

131 assert consume3.is_set() (empty)

132 

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

134 routes = [] (empty)

135 

136 class Middleware(BaseMiddleware): (empty)

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

138 routes.append(None) (empty)

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

140 

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

142 

143 @broker.subscriber(queue) (empty)

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

145 

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

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

148 

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

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

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

152 

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

154 

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

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

157 routes = [] (empty)

158 

159 class Middleware(BaseMiddleware): (empty)

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

161 routes.append(None) (empty)

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

163 

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

165 

166 @broker.subscriber(queue) (empty)

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

168 

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

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

171 

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

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

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

175 await h1.wait_call(3) (empty)

176 await h2.wait_call(3) (empty)

177 

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

179 

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

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

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

183 

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

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

186 await super().test_broker_with_real_doesnt_get_patched() (empty)

187 

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

189 async def test_broker_with_real_patches_publishers_and_subscribers( (empty)

190 self, 

191 queue: str, 

192 ) -> None: 

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

194 

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

196 @pytest.mark.parametrize( (empty)

197 ("expiration", "expected"), 

198 ( 

199 pytest.param(None, None, id="none"), 

200 pytest.param(1, 1, id="int"), 

201 pytest.param(1.5, 1.5, id="float"), 

202 pytest.param(dt.timedelta(seconds=1.1), 1.1, id="timedelta"), 

203 pytest.param(_frozen_time, 0, id="datetime"), 

204 ), 

205 ) 

206 @freeze_time(_frozen_time) (empty)

207 async def test_publish_expiration_propagated( (empty)

208 self, expiration: Any, expected: Any, queue: str, mock: MagicMock 

209 ) -> None: 

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

211 

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

213 

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

215 async def m(msg: RabbitMessage) -> None: (empty)

216 mock(msg) (empty)

217 

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

219 await br.start() (empty)

220 await br.publish("hello", queue, expiration=expiration) (empty)

221 msg = mock.call_args[0][0] (empty)

222 assert msg.raw_message.expiration == expected (empty)

223 

224 

225@pytest.mark.parametrize( (empty)

226 ("pattern", "current", "result"), 

227 ( 

228 pytest.param("#", "1.2.3", True, id="#"), 

229 pytest.param("*", "1", True, id="*"), 

230 pytest.param("*", "1.2", False, id="* - broken"), 

231 pytest.param("test.*", "test.1", True, id="test.*"), 

232 pytest.param("test.#", "test.1", True, id="test.#"), 

233 pytest.param("#.test.#", "1.2.test.1.2", True, id="#.test.#"), 

234 pytest.param("#.test.*", "1.2.test.1", True, id="#.test.*"), 

235 pytest.param("#.test.*.*", "1.2.test.1.2", True, id="#.test.*."), 

236 pytest.param("#.test.*.*.*", "1.2.test.1.2", False, id="#.test.*.*.* - broken"), 

237 pytest.param( 

238 "#.test.*.test.#", 

239 "1.2.test.1.test.1.2", 

240 True, 

241 id="#.test.*.test.#", 

242 ), 

243 pytest.param("#.*.test", "1.2.2.test", True, id="#.*.test"), 

244 pytest.param("#.2.*.test", "1.2.2.test", True, id="#.2.*.test"), 

245 pytest.param("#.*.*.test", "1.2.2.test", True, id="#.*.*.test"), 

246 pytest.param("*.*.*.test", "1.2.test", False, id="*.*.*.test - broken"), 

247 pytest.param("#.*.*.test", "1.2.test", False, id="#.*.*.test - broken"), 

248 ), 

249) 

250def test(pattern: str, current: str, result: bool) -> None: (empty)

251 assert apply_pattern(pattern, current) == result (empty)

252 

253 

254exch_direct = RabbitExchange("exchange", auto_delete=True, type=ExchangeType.DIRECT) (empty)

255exch_fanout = RabbitExchange("exchange", auto_delete=True, type=ExchangeType.FANOUT) (empty)

256exch_topic = RabbitExchange("exchange", auto_delete=True, type=ExchangeType.TOPIC) (empty)

257exch_headers = RabbitExchange("exchange", auto_delete=True, type=ExchangeType.HEADERS) (empty)

258reqular_queue = RabbitQueue("test-reqular-queue", auto_delete=True) (empty)

259 

260routing_key_queue = RabbitQueue( (empty)

261 "test-routing-key-queue", 

262 auto_delete=True, 

263 routing_key="*.info", 

264) 

265one_key_queue = RabbitQueue( (empty)

266 "test-one-key-queue", 

267 auto_delete=True, 

268 bind_arguments={"key": 1}, 

269) 

270any_keys_queue = RabbitQueue( (empty)

271 "test-any-keys-queue", 

272 auto_delete=True, 

273 bind_arguments={"key": 2, "key2": 2, "x-match": "any"}, 

274) 

275all_keys_queue = RabbitQueue( (empty)

276 "test-all-keys-queue", 

277 auto_delete=True, 

278 bind_arguments={"key": 2, "key2": 2, "x-match": "all"}, 

279) 

280 

281broker = RabbitBroker() (empty)

282 

283 

284@pytest.mark.rabbit() (empty)

285@pytest.mark.parametrize( (empty)

286 ( 

287 "queue", 

288 "exchange", 

289 "routing_key", 

290 "headers", 

291 "expected_result", 

292 ), 

293 ( 

294 pytest.param( 

295 reqular_queue, 

296 exch_direct, 

297 reqular_queue.routing(), 

298 {}, 

299 True, 

300 id="direct match", 

301 ), 

302 pytest.param( 

303 reqular_queue, 

304 exch_direct, 

305 "wrong key", 

306 {}, 

307 False, 

308 id="direct mismatch", 

309 ), 

310 pytest.param( 

311 reqular_queue, 

312 exch_fanout, 

313 "", 

314 {}, 

315 True, 

316 id="fanout match", 

317 ), 

318 pytest.param( 

319 routing_key_queue, 

320 exch_topic, 

321 "log.info", 

322 {}, 

323 True, 

324 id="topic match", 

325 ), 

326 pytest.param( 

327 routing_key_queue, 

328 exch_topic, 

329 "log.wrong", 

330 {}, 

331 False, 

332 id="topic mismatch", 

333 ), 

334 pytest.param( 

335 one_key_queue, 

336 exch_headers, 

337 "", 

338 {"key": 1}, 

339 True, 

340 id="one header match", 

341 ), 

342 pytest.param( 

343 one_key_queue, 

344 exch_headers, 

345 "", 

346 {"key": "wrong"}, 

347 False, 

348 id="one header mismatch", 

349 ), 

350 pytest.param( 

351 any_keys_queue, 

352 exch_headers, 

353 "", 

354 {"key2": 2}, 

355 True, 

356 id="any headers match", 

357 ), 

358 pytest.param( 

359 any_keys_queue, 

360 exch_headers, 

361 "", 

362 {"key2": "wrong"}, 

363 False, 

364 id="any headers mismatch", 

365 ), 

366 pytest.param( 

367 all_keys_queue, 

368 exch_headers, 

369 "", 

370 {"key": 2, "key2": 2}, 

371 True, 

372 id="all headers match", 

373 ), 

374 pytest.param( 

375 all_keys_queue, 

376 exch_headers, 

377 "", 

378 {"key": "wrong", "key2": 2}, 

379 False, 

380 id="all headers mismatch", 

381 ), 

382 ), 

383) 

384def test_in_memory_routing( (empty)

385 queue: str, 

386 exchange: RabbitExchange, 

387 routing_key: str, 

388 headers: dict[str, Any], 

389 expected_result: bool, 

390) -> None: 

391 subscriber = broker.subscriber(queue, exchange) (empty)

392 assert ( (empty)

393 _is_handler_matches(subscriber, routing_key, headers, exchange) is expected_result 

394 )