Coverage for tests / asgi / testcase.py: 99%

1import asyncio (empty)

2import math (empty)

3from collections.abc import Callable (empty)

4from typing import Any (empty)

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

6 

7import pytest (empty)

8from dirty_equals import Contains, IsFloat, IsList, IsPartialDict, IsStr (empty)

9from fast_depends import Depends (empty)

10from freezegun import freeze_time (empty)

11from starlette.applications import Starlette (empty)

12from starlette.routing import Mount (empty)

13from starlette.testclient import TestClient (empty)

14from starlette.websockets import WebSocketDisconnect (empty)

15 

16from faststream._internal.context import Context (empty)

17from faststream.annotations import FastStream, Logger (empty)

18from faststream.asgi import ( (empty)

19 AsgiFastStream, 

20 AsgiResponse, 

21 AsyncAPIRoute, 

22 Request, 

23 get, 

24 make_asyncapi_asgi, 

25 make_ping_asgi, 

26 post, 

27) 

28from faststream.asgi.params import Header, Query (empty)

29from faststream.asgi.types import ASGIApp, Scope (empty)

30from faststream.specification import AsyncAPI (empty)

31 

32 

33class AsgiTestcase: (empty)

34 def get_broker(self) -> Any: (empty)

35 raise NotImplementedError 

36 

37 def get_test_broker(self, broker: Any) -> Any: (empty)

38 raise NotImplementedError 

39 

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

41 async def test_not_found(self) -> None: (empty)

42 broker = self.get_broker() (empty)

43 app = AsgiFastStream(broker) (empty)

44 

45 async with self.get_test_broker(broker): (empty)

46 with TestClient(app) as client: (empty)

47 response = client.get("/") (empty)

48 assert response.status_code == 404 (empty)

49 

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

51 async def test_ws_not_found(self) -> None: (empty)

52 broker = self.get_broker() (empty)

53 

54 app = AsgiFastStream(broker) (empty)

55 

56 async with self.get_test_broker(broker): (empty)

57 with TestClient(app) as client: (empty)

58 with pytest.raises(WebSocketDisconnect): (empty)

59 with client.websocket_connect("/ws"): # raises error (empty)

60 pass 

61 

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

63 async def test_asgi_ping_healthy(self) -> None: (empty)

64 broker = self.get_broker() (empty)

65 

66 app = AsgiFastStream( (empty)

67 broker, 

68 asgi_routes=[("/health", make_ping_asgi(broker, timeout=5.0))], 

69 ) 

70 

71 async with self.get_test_broker(broker): (empty)

72 with TestClient(app) as client: (empty)

73 response = client.get("/health") (empty)

74 assert response.status_code == 204 (empty)

75 

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

77 async def test_asgi_ping_unhealthy(self) -> None: (empty)

78 broker = self.get_broker() (empty)

79 

80 app = AsgiFastStream( (empty)

81 broker, 

82 asgi_routes=[ 

83 ("/health", make_ping_asgi(broker, timeout=5.0)), 

84 ], 

85 ) 

86 async with self.get_test_broker(broker) as br: (empty)

87 br.ping = AsyncMock() (empty)

88 br.ping.return_value = False (empty)

89 

90 with TestClient(app) as client: (empty)

91 response = client.get("/health") (empty)

92 assert response.status_code == 500 (empty)

93 

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

95 async def test_asyncapi_asgi(self) -> None: (empty)

96 broker = self.get_broker() (empty)

97 

98 app = AsgiFastStream( (empty)

99 broker, 

100 specification=AsyncAPI(), 

101 asyncapi_path="/docs", 

102 ) 

103 

104 async with self.get_test_broker(broker): (empty)

105 with TestClient(app) as client: (empty)

106 response = client.get("/docs") (empty)

107 assert response.status_code == 200, response (empty)

108 assert response.text (empty)

109 

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

111 async def test_asyncapi_asgi_if_broker_set_by_method(self) -> None: (empty)

112 broker = self.get_broker() (empty)

113 

114 app = AsgiFastStream( (empty)

115 specification=AsyncAPI(), 

116 asyncapi_path="/docs", 

117 ) 

118 

119 app.set_broker(broker) (empty)

120 

121 async with self.get_test_broker(broker): (empty)

122 with TestClient(app) as client: (empty)

123 response = client.get("/docs") (empty)

124 assert response.status_code == 200, response (empty)

125 assert response.text (empty)

126 

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

128 @pytest.mark.parametrize( (empty)

129 ("decorator", "client_method"), (empty)

130 ( (empty)

131 pytest.param(get, "get", id="get"), (empty)

132 pytest.param(post, "post", id="post"), (empty)

133 ), 

134 ) 

135 async def test_decorators( (empty)

136 self, decorator: Callable[..., ASGIApp], client_method: str (empty)

137 ) -> None: (empty)

138 @decorator (empty)

139 async def some_handler(scope: Scope) -> AsgiResponse: (empty)

140 return AsgiResponse(body=b"test", status_code=200) (empty)

141 

142 broker = self.get_broker() (empty)

143 app = AsgiFastStream(broker, asgi_routes=[("/test", some_handler)]) (empty)

144 

145 async with self.get_test_broker(broker): (empty)

146 with TestClient(app) as client: (empty)

147 response = getattr(client, client_method)("/test") (empty)

148 assert response.status_code == 200 (empty)

149 assert response.text == "test" (empty)

150 

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

152 @pytest.mark.parametrize( (empty)

153 ("decorator", "client_method"), (empty)

154 ( (empty)

155 pytest.param(get, "get", id="get"), (empty)

156 pytest.param(post, "post", id="post"), (empty)

157 ), 

158 ) 

159 async def test_context_injected( (empty)

160 self, decorator: Callable[..., ASGIApp], client_method: str (empty)

161 ) -> None: (empty)

162 @decorator (empty)

163 async def some_handler( (empty)

164 request: Request, logger: Logger, app: FastStream (empty)

165 ) -> AsgiResponse: (empty)

166 return AsgiResponse( (empty)

167 body=f"{request.__class__.__name__} {logger.__class__.__name__} {app.__class__.__name__}".encode(), (empty)

168 status_code=200, (empty)

169 ) 

170 

171 broker = self.get_broker() (empty)

172 app = AsgiFastStream(broker, asgi_routes=[("/test", some_handler)]) (empty)

173 

174 async with self.get_test_broker(broker): (empty)

175 with TestClient(app) as client: (empty)

176 response = getattr(client, client_method)("/test") (empty)

177 assert response.status_code == 200 (empty)

178 assert response.text == "AsgiRequest Logger AsgiFastStream" (empty)

179 

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

181 @pytest.mark.parametrize( (empty)

182 ("decorator", "client_method"), (empty)

183 ( (empty)

184 pytest.param(get, "get", id="get"), (empty)

185 pytest.param(post, "post", id="post"), (empty)

186 ), 

187 ) 

188 async def test_fast_depends_injected( (empty)

189 self, decorator: Callable[..., ASGIApp], client_method: str (empty)

190 ) -> None: (empty)

191 def get_string() -> str: (empty)

192 return "test" (empty)

193 

194 @decorator (empty)

195 async def some_handler(string=Depends(get_string)) -> AsgiResponse: # noqa: B008 (empty)

196 return AsgiResponse(body=string.encode(), status_code=200) (empty)

197 

198 broker = self.get_broker() (empty)

199 app = AsgiFastStream(broker, asgi_routes=[("/test", some_handler)]) (empty)

200 

201 async with self.get_test_broker(broker): (empty)

202 with TestClient(app) as client: (empty)

203 response = getattr(client, client_method)("/test") (empty)

204 assert response.status_code == 200 (empty)

205 assert response.text == "test" (empty)

206 

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

208 @pytest.mark.parametrize( (empty)

209 "dependency", 

210 ( 

211 pytest.param(Query(), id="query"), 

212 pytest.param(Header(), id="header"), 

213 ), 

214 ) 

215 async def test_validation_error_handled(self, dependency: Context) -> None: (empty)

216 @get (empty)

217 async def some_handler(dep=dependency) -> AsgiResponse: (empty)

218 return AsgiResponse(status_code=200) 

219 

220 broker = self.get_broker() (empty)

221 app = AsgiFastStream(broker, asgi_routes=[("/test", some_handler)]) (empty)

222 

223 async with self.get_test_broker(broker): (empty)

224 with TestClient(app) as client: (empty)

225 response = client.get("/test") (empty)

226 assert response.status_code == 422 (empty)

227 assert response.text == "Validation error" (empty)

228 

229 def test_asyncapi_pure_asgi(self) -> None: (empty)

230 broker = self.get_broker() (empty)

231 

232 app = Starlette(routes=[Mount("/", make_asyncapi_asgi(AsyncAPI(broker)))]) (empty)

233 

234 with TestClient(app) as client: (empty)

235 response = client.get("/") (empty)

236 assert response.status_code == 200 (empty)

237 assert response.text == Contains("<!DOCTYPE html>") (empty)

238 

239 # ===== TryItOut tests ===== 

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

241 async def test_try_it_out_message_delivered_to_subscriber( (empty)

242 self, queue: str, mock: MagicMock 

243 ) -> None: 

244 broker = self.get_broker() (empty)

245 

246 @broker.subscriber(queue) (empty)

247 async def handler(msg: Any) -> None: (empty)

248 mock(msg) (empty)

249 

250 app = AsgiFastStream( (empty)

251 broker, 

252 asyncapi_path=AsyncAPIRoute("/asyncapi", try_it_out=True), 

253 ) 

254 

255 async with self.get_test_broker(broker): (empty)

256 with TestClient(app) as client: (empty)

257 client.post( (empty)

258 "/asyncapi/try", 

259 json={ 

260 "channelName": queue, 

261 "message": { 

262 "operation_id": "op", 

263 "operation_type": "subscribe", 

264 "message": {"text": "hello"}, 

265 }, 

266 "options": {"sendToRealBroker": False}, 

267 }, 

268 ) 

269 

270 mock.assert_called_once_with(IsPartialDict(text="hello")) (empty)

271 

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

273 async def test_try_it_out_string_payload_delivered( (empty)

274 self, queue: str, mock: MagicMock 

275 ) -> None: 

276 """Plugin wraps primitive payloads in message.message — ensure they arrive correctly.""" 

277 broker = self.get_broker() (empty)

278 

279 @broker.subscriber(queue) (empty)

280 async def handler(msg: Any) -> None: (empty)

281 mock(msg) (empty)

282 

283 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

284 

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

286 with TestClient(app) as client: (empty)

287 client.post( (empty)

288 "/asyncapi/try", 

289 json={ 

290 "channelName": queue, 

291 "message": { 

292 "operation_id": "op", 

293 "operation_type": "subscribe", 

294 "message": "hello", 

295 }, 

296 "options": {"sendToRealBroker": False}, 

297 }, 

298 ) 

299 

300 mock.assert_called_once_with("hello") (empty)

301 

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

303 async def test_try_it_out_integer_payload_delivered( (empty)

304 self, queue: str, mock: MagicMock 

305 ) -> None: 

306 """Primitive int payload should arrive correctly.""" 

307 broker = self.get_broker() (empty)

308 

309 @broker.subscriber(queue) (empty)

310 async def handler(msg: int) -> None: (empty)

311 mock(msg) (empty)

312 

313 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

314 

315 async with self.get_test_broker(broker): (empty)

316 with TestClient(app) as client: (empty)

317 client.post( (empty)

318 "/asyncapi/try", 

319 json={ 

320 "channelName": queue, 

321 "message": { 

322 "operation_id": "op", 

323 "operation_type": "subscribe", 

324 "message": 42, 

325 }, 

326 "options": {"sendToRealBroker": False}, 

327 }, 

328 ) 

329 

330 mock.assert_called_once_with(42) (empty)

331 

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

333 async def test_try_it_out_float_payload_delivered( (empty)

334 self, queue: str, mock: MagicMock 

335 ) -> None: 

336 """Primitive float payload should arrive correctly.""" 

337 broker = self.get_broker() (empty)

338 

339 @broker.subscriber(queue) (empty)

340 async def handler(msg: float) -> None: (empty)

341 mock(msg) (empty)

342 

343 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

344 

345 async with self.get_test_broker(broker): (empty)

346 with TestClient(app) as client: (empty)

347 client.post( (empty)

348 "/asyncapi/try", 

349 json={ 

350 "channelName": queue, 

351 "message": { 

352 "operation_id": "op", 

353 "operation_type": "subscribe", 

354 "message": math.pi, 

355 }, 

356 "options": {"sendToRealBroker": False}, 

357 }, 

358 ) 

359 

360 mock.assert_called_once_with(IsFloat(approx=math.pi)) (empty)

361 

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

363 async def test_try_it_out_boolean_payload_delivered( (empty)

364 self, queue: str, mock: MagicMock 

365 ) -> None: 

366 """Primitive boolean payload should arrive correctly.""" 

367 broker = self.get_broker() (empty)

368 

369 @broker.subscriber(queue) (empty)

370 async def handler(msg: bool) -> None: (empty)

371 mock(msg) (empty)

372 

373 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

374 

375 async with self.get_test_broker(broker): (empty)

376 with TestClient(app) as client: (empty)

377 client.post( (empty)

378 "/asyncapi/try", 

379 json={ 

380 "channelName": queue, 

381 "message": { 

382 "operation_id": "op", 

383 "operation_type": "subscribe", 

384 "message": True, 

385 }, 

386 "options": {"sendToRealBroker": False}, 

387 }, 

388 ) 

389 

390 mock.assert_called_once_with(True) (empty)

391 

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

393 async def test_try_it_out_array_payload_delivered( (empty)

394 self, queue: str, mock: MagicMock 

395 ) -> None: 

396 """Array payload should arrive correctly.""" 

397 broker = self.get_broker() (empty)

398 

399 @broker.subscriber(queue) (empty)

400 async def handler(msg: list[Any]) -> None: (empty)

401 mock(msg) (empty)

402 

403 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

404 

405 async with self.get_test_broker(broker): (empty)

406 with TestClient(app) as client: (empty)

407 client.post( (empty)

408 "/asyncapi/try", 

409 json={ 

410 "channelName": queue, 

411 "message": { 

412 "operation_id": "op", 

413 "operation_type": "subscribe", 

414 "message": ["one", "two", "three"], 

415 }, 

416 "options": {"sendToRealBroker": False}, 

417 }, 

418 ) 

419 

420 mock.assert_called_once_with(IsList("one", "two", "three")) (empty)

421 

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

423 async def test_try_it_out_object_payload_delivered( (empty)

424 self, queue: str, mock: MagicMock 

425 ) -> None: 

426 """Object (dict) payload should arrive correctly.""" 

427 broker = self.get_broker() (empty)

428 

429 @broker.subscriber(queue) (empty)

430 async def handler(msg: dict[str, Any]) -> None: (empty)

431 mock(msg) (empty)

432 

433 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

434 

435 async with self.get_test_broker(broker): (empty)

436 with TestClient(app) as client: (empty)

437 client.post( (empty)

438 "/asyncapi/try", 

439 json={ 

440 "channelName": queue, 

441 "message": { 

442 "operation_id": "op", 

443 "operation_type": "subscribe", 

444 "message": {"field": "value", "count": 42, "mock": True}, 

445 }, 

446 "options": {"sendToRealBroker": False}, 

447 }, 

448 ) 

449 

450 mock.assert_called_once_with(IsPartialDict(field="value", count=42, mock=True)) (empty)

451 

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

453 async def test_try_it_out_memory_subsricber_returns_result(self, queue: str) -> None: (empty)

454 broker = self.get_broker() (empty)

455 

456 @broker.subscriber(queue) (empty)

457 async def handler(msg: dict[str, Any]) -> dict[str, Any]: (empty)

458 return {"result": 1} (empty)

459 

460 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

461 

462 async with self.get_test_broker(broker): (empty)

463 with TestClient(app) as client: (empty)

464 response = client.post( (empty)

465 "/asyncapi/try", 

466 json={ 

467 "channelName": queue, 

468 "message": { 

469 "operation_id": "op", 

470 "operation_type": "subscribe", 

471 "message": {"data": "hello"}, 

472 }, 

473 "options": {"sendToRealBroker": False}, 

474 }, 

475 ) 

476 assert response.status_code == 200, response.json() (empty)

477 assert response.json() == IsPartialDict(result=1) (empty)

478 

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

480 async def test_try_it_out_disabled(self, queue: str) -> None: (empty)

481 broker = self.get_broker() (empty)

482 

483 app = AsgiFastStream( (empty)

484 broker, 

485 asyncapi_path=AsyncAPIRoute("/asyncapi", try_it_out=False), 

486 ) 

487 

488 async with self.get_test_broker(broker): (empty)

489 with TestClient(app) as client: (empty)

490 r = client.post( (empty)

491 "/asyncapi/try", 

492 json={ 

493 "channelName": queue, 

494 "message": { 

495 "operation_id": "op", 

496 "operation_type": "subscribe", 

497 "message": {"text": "hello"}, 

498 }, 

499 "options": {"sendToRealBroker": False}, 

500 }, 

501 ) 

502 assert r.status_code == 404 (empty)

503 

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

505 async def test_try_it_out_missing_channel_returns_400(self) -> None: (empty)

506 broker = self.get_broker() (empty)

507 

508 app = AsgiFastStream(broker, asyncapi_path="/docs") (empty)

509 

510 async with self.get_test_broker(broker): (empty)

511 with TestClient(app) as client: (empty)

512 response = client.post("/docs/try", json={"message": {}}) (empty)

513 assert response.status_code == 400 (empty)

514 assert response.json() == IsPartialDict(details="Missing channelName") (empty)

515 

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

517 async def test_try_it_out_channel_not_found(self, queue: str) -> None: (empty)

518 broker = self.get_broker() (empty)

519 

520 app = AsgiFastStream(broker, asyncapi_path="/docs") (empty)

521 

522 async with self.get_test_broker(broker): (empty)

523 with TestClient(app) as client: (empty)

524 response = client.post( (empty)

525 "/docs/try", 

526 json={ 

527 "channelName": queue, 

528 "message": { 

529 "operation_id": "op", 

530 "operation_type": "subscribe", 

531 "message": {"text": "hello"}, 

532 }, 

533 "options": {"sendToRealBroker": False}, 

534 }, 

535 ) 

536 assert response.status_code == 404, response.status_code (empty)

537 assert response.json() == IsPartialDict( (empty)

538 details=IsStr(regex=r".+ destination not found\.") 

539 ) 

540 

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

542 async def test_try_it_out_path_follows_asyncapi_path(self, queue: str) -> None: (empty)

543 broker = self.get_broker() (empty)

544 

545 @broker.subscriber(queue) (empty)

546 async def handler(msg: dict[str, Any]) -> None: (empty)

547 pass (empty)

548 

549 app = AsgiFastStream(broker, asyncapi_path="/custom/docs") (empty)

550 

551 async with self.get_test_broker(broker): (empty)

552 with TestClient(app) as client: (empty)

553 response = client.post( (empty)

554 "/custom/docs/try", 

555 json={ 

556 "channelName": queue, 

557 "message": { 

558 "operation_id": "op", 

559 "operation_type": "subscribe", 

560 "message": {}, 

561 }, 

562 "options": {"sendToRealBroker": False}, 

563 }, 

564 ) 

565 

566 assert response.status_code == 200 (empty)

567 assert response.json() == "ok" (empty)

568 

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

570 @freeze_time(auto_tick_seconds=5) (empty)

571 async def test_try_it_out_subscriber_completes_within_timeout( (empty)

572 self, queue: str 

573 ) -> None: 

574 """Subscriber that finishes before the configured timeout should return a 200 with its result.""" 

575 broker = self.get_broker() (empty)

576 

577 @broker.subscriber(queue) (empty)

578 async def handler(msg: Any) -> dict[str, Any]: (empty)

579 await asyncio.sleep(5) (empty)

580 return {"result": "done"} (empty)

581 

582 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

583 

584 async with self.get_test_broker(broker): (empty)

585 with TestClient(app) as client: (empty)

586 response = client.post( (empty)

587 "/asyncapi/try", 

588 json={ 

589 "channelName": queue, 

590 "message": { 

591 "operation_id": "op", 

592 "operation_type": "subscribe", 

593 "message": {"data": "hello"}, 

594 }, 

595 "options": {"sendToRealBroker": False}, 

596 }, 

597 ) 

598 

599 assert response.status_code == 200, response.json() (empty)

600 assert response.json() == IsPartialDict(result="done") (empty)

601 

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

603 @freeze_time(auto_tick_seconds=30) (empty)

604 async def test_try_it_out_subscriber_exceeds_timeout_returns_500( (empty)

605 self, queue: str 

606 ) -> None: 

607 """Subscriber that runs longer than the configured timeout should produce a 500 carrying the timeout exception.""" 

608 broker = self.get_broker() (empty)

609 

610 @broker.subscriber(queue) (empty)

611 async def handler(msg: Any) -> None: (empty)

612 await asyncio.sleep(30) (empty)

613 

614 app = AsgiFastStream(broker, asyncapi_path="/asyncapi") (empty)

615 

616 async with self.get_test_broker(broker): (empty)

617 with TestClient(app) as client: (empty)

618 response = client.post( (empty)

619 "/asyncapi/try", 

620 json={ 

621 "channelName": queue, 

622 "message": { 

623 "operation_id": "op", 

624 "operation_type": "subscribe", 

625 "message": {"data": "hello"}, 

626 }, 

627 "options": {"sendToRealBroker": False}, 

628 }, 

629 ) 

630 

631 assert response.status_code == 500 (empty)

632 assert response.json() == IsPartialDict(details=Contains("TimeoutError")) (empty)

633 

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

635 async def test_try_it_out_spec_endpoint_base_overrides_route_default(self) -> None: (empty)

636 broker = self.get_broker() (empty)

637 

638 app = AsgiFastStream( (empty)

639 broker, 

640 asyncapi_path=AsyncAPIRoute( 

641 "/docs", 

642 try_it_out_url="https://api.example.com/try", 

643 ), 

644 ) 

645 

646 async with self.get_test_broker(broker): (empty)

647 with TestClient(app) as client: (empty)

648 response = client.get("/docs") (empty)

649 assert response.status_code == 200 (empty)

650 assert response.text == Contains("https://api.example.com/try") (empty)