Coverage for tests / asyncapi / base / v2_6_0 / arguments.py: 98%

1import sys (empty)

2from dataclasses import dataclass, field (empty)

3from enum import Enum (empty)

4from typing import Annotated, Literal (empty)

5 

6import pydantic (empty)

7import pytest (empty)

8from dirty_equals import IsDict, IsPartialDict, IsStr (empty)

9from fast_depends import Depends (empty)

10 

11from faststream import Context (empty)

12from faststream._internal.broker import BrokerUsecase (empty)

13from tests.marks import PYDANTIC_V2, pydantic_v2 (empty)

14 

15from .basic import AsyncAPI260Factory (empty)

16 

17 

18class FastAPICompatible(AsyncAPI260Factory): (empty)

19 is_fastapi: bool = False (empty)

20 

21 broker_class: type[BrokerUsecase] (empty)

22 dependency_builder = staticmethod(Depends) (empty)

23 

24 def test_custom_naming(self) -> None: (empty)

25 broker = self.broker_class() (empty)

26 

27 @broker.subscriber("test", title="custom_name", description="test description") (empty)

28 async def handle(msg) -> None: ... (empty)

29 

30 schema = self.get_spec(broker).to_jsonable() (empty)

31 key = tuple(schema["channels"].keys())[0] # noqa: RUF015 (empty)

32 

33 assert key == "custom_name" (empty)

34 assert schema["channels"][key]["description"] == "test description" (empty)

35 

36 def test_slash_in_title(self) -> None: (empty)

37 broker = self.broker_class() (empty)

38 

39 @broker.subscriber("test", title="/") (empty)

40 async def handle(msg) -> None: ... (empty)

41 

42 schema = self.get_spec(broker).to_jsonable() (empty)

43 

44 assert next(iter(schema["channels"].keys())) == "/" (empty)

45 

46 assert next(iter(schema["components"]["messages"].keys())) == ".:Message" (empty)

47 assert schema["components"]["messages"][".:Message"]["title"] == "/:Message" (empty)

48 

49 assert next(iter(schema["components"]["schemas"].keys())) == ".:Message:Payload" (empty)

50 assert ( (empty)

51 schema["components"]["schemas"][".:Message:Payload"]["title"] 

52 == "/:Message:Payload" 

53 ) 

54 

55 def test_docstring_description(self) -> None: (empty)

56 broker = self.broker_class() (empty)

57 

58 @broker.subscriber("test", title="custom_name") (empty)

59 async def handle(msg) -> None: (empty)

60 """Test description.""" 

61 

62 schema = self.get_spec(broker).to_jsonable() (empty)

63 key = tuple(schema["channels"].keys())[0] # noqa: RUF015 (empty)

64 

65 assert key == "custom_name" (empty)

66 assert schema["channels"][key]["description"] == "Test description.", schema[ (empty)

67 "channels" 

68 ][key] 

69 

70 def test_empty(self) -> None: (empty)

71 broker = self.broker_class() (empty)

72 

73 @broker.subscriber("test") (empty)

74 async def handle() -> None: ... (empty)

75 

76 schema = self.get_spec(broker).to_jsonable() (empty)

77 

78 payload = schema["components"]["schemas"] (empty)

79 

80 for key, v in payload.items(): (empty)

81 assert key == "EmptyPayload" (empty)

82 assert v == { (empty)

83 "title": key, 

84 "type": "null", 

85 } 

86 

87 def test_no_type(self) -> None: (empty)

88 broker = self.broker_class() (empty)

89 

90 @broker.subscriber("test") (empty)

91 async def handle(msg) -> None: ... (empty)

92 

93 schema = self.get_spec(broker).to_jsonable() (empty)

94 

95 payload = schema["components"]["schemas"] (empty)

96 

97 for key, v in payload.items(): (empty)

98 assert key == "Handle:Message:Payload" (empty)

99 assert v == {"title": key} (empty)

100 

101 def test_simple_type(self) -> None: (empty)

102 broker = self.broker_class() (empty)

103 

104 @broker.subscriber("test") (empty)

105 async def handle(msg: int) -> None: ... (empty)

106 

107 schema = self.get_spec(broker).to_jsonable() (empty)

108 

109 payload = schema["components"]["schemas"] (empty)

110 assert next(iter(schema["channels"].values())).get("description") is None (empty)

111 

112 for key, v in payload.items(): (empty)

113 assert key == "Handle:Message:Payload" (empty)

114 assert v == {"title": key, "type": "integer"} (empty)

115 

116 def test_simple_optional_type(self) -> None: (empty)

117 broker = self.broker_class() (empty)

118 

119 @broker.subscriber("test") (empty)

120 async def handle(msg: int | None) -> None: ... (empty)

121 

122 schema = self.get_spec(broker).to_jsonable() (empty)

123 

124 payload = schema["components"]["schemas"] (empty)

125 

126 for key, v in payload.items(): (empty)

127 assert key == "Handle:Message:Payload" (empty)

128 assert v == IsDict( (empty)

129 { 

130 "anyOf": [{"type": "integer"}, {"type": "null"}], 

131 "title": key, 

132 }, 

133 ) | IsDict( 

134 { # TODO: remove when deprecating PydanticV1 

135 "title": key, 

136 "type": "integer", 

137 }, 

138 ), v 

139 

140 def test_simple_type_with_default(self) -> None: (empty)

141 broker = self.broker_class() (empty)

142 

143 @broker.subscriber("test") (empty)

144 async def handle(msg: int = 1) -> None: ... (empty)

145 

146 schema = self.get_spec(broker).to_jsonable() (empty)

147 

148 payload = schema["components"]["schemas"] (empty)

149 

150 for key, v in payload.items(): (empty)

151 assert key == "Handle:Message:Payload" (empty)

152 assert v == { (empty)

153 "default": 1, 

154 "title": key, 

155 "type": "integer", 

156 } 

157 

158 def test_multi_args_no_type(self) -> None: (empty)

159 broker = self.broker_class() (empty)

160 

161 @broker.subscriber("test") (empty)

162 async def handle(msg, another) -> None: ... (empty)

163 

164 schema = self.get_spec(broker).to_jsonable() (empty)

165 

166 payload = schema["components"]["schemas"] (empty)

167 

168 for key, v in payload.items(): (empty)

169 assert key == "Handle:Message:Payload" (empty)

170 assert v == { (empty)

171 "properties": { 

172 "another": {"title": "Another"}, 

173 "msg": {"title": "Msg"}, 

174 }, 

175 "required": ["msg", "another"], 

176 "title": key, 

177 "type": "object", 

178 } 

179 

180 def test_multi_args_with_type(self) -> None: (empty)

181 broker = self.broker_class() (empty)

182 

183 @broker.subscriber("test") (empty)

184 async def handle(msg: str, another: int) -> None: ... (empty)

185 

186 schema = self.get_spec(broker).to_jsonable() (empty)

187 

188 payload = schema["components"]["schemas"] (empty)

189 

190 for key, v in payload.items(): (empty)

191 assert key == "Handle:Message:Payload" (empty)

192 assert v == { (empty)

193 "properties": { 

194 "another": {"title": "Another", "type": "integer"}, 

195 "msg": {"title": "Msg", "type": "string"}, 

196 }, 

197 "required": ["msg", "another"], 

198 "title": key, 

199 "type": "object", 

200 } 

201 

202 def test_multi_args_with_default(self) -> None: (empty)

203 broker = self.broker_class() (empty)

204 

205 @broker.subscriber("test") (empty)

206 async def handle(msg: str, another: int | None = None) -> None: ... (empty)

207 

208 schema = self.get_spec(broker).to_jsonable() (empty)

209 

210 payload = schema["components"]["schemas"] (empty)

211 

212 for key, v in payload.items(): (empty)

213 assert key == "Handle:Message:Payload" (empty)

214 

215 assert v == { (empty)

216 "properties": { 

217 "another": IsDict( 

218 { 

219 "anyOf": [{"type": "integer"}, {"type": "null"}], 

220 "default": None, 

221 "title": "Another", 

222 }, 

223 ) 

224 | IsDict( 

225 { # TODO: remove when deprecating PydanticV1 

226 "title": "Another", 

227 "type": "integer", 

228 }, 

229 ), 

230 "msg": {"title": "Msg", "type": "string"}, 

231 }, 

232 "required": ["msg"], 

233 "title": key, 

234 "type": "object", 

235 } 

236 

237 def test_dataclass(self) -> None: (empty)

238 @dataclass (empty)

239 class User: (empty)

240 id: int (empty)

241 name: str = "" (empty)

242 

243 broker = self.broker_class() (empty)

244 

245 @broker.subscriber("test") (empty)

246 async def handle(user: User) -> None: ... (empty)

247 

248 schema = self.get_spec(broker).to_jsonable() (empty)

249 

250 payload = schema["components"]["schemas"] (empty)

251 

252 for key, v in payload.items(): (empty)

253 assert key == "User" (empty)

254 assert v == { (empty)

255 "properties": { 

256 "id": {"title": "Id", "type": "integer"}, 

257 "name": {"default": "", "title": "Name", "type": "string"}, 

258 }, 

259 "required": ["id"], 

260 "title": key, 

261 "type": "object", 

262 } 

263 

264 def test_dataclasses_nested(self): (empty)

265 @dataclass (empty)

266 class Product: (empty)

267 id: int (empty)

268 name: str = "" (empty)

269 

270 @dataclass (empty)

271 class Order: (empty)

272 id: int (empty)

273 products: list[Product] = field(default_factory=list) (empty)

274 

275 broker = self.broker_class() (empty)

276 

277 @broker.subscriber("test") (empty)

278 async def handle(order: Order): ... (empty)

279 

280 schema = self.get_spec(broker).to_jsonable() (empty)

281 

282 payload = schema["components"]["schemas"] (empty)

283 

284 assert payload == { (empty)

285 "Product": { 

286 "properties": { 

287 "id": {"title": "Id", "type": "integer"}, 

288 "name": {"default": "", "title": "Name", "type": "string"}, 

289 }, 

290 "required": ["id"], 

291 "title": "Product", 

292 "type": "object", 

293 }, 

294 "Order": { 

295 "properties": { 

296 "id": {"title": "Id", "type": "integer"}, 

297 "products": { 

298 "items": {"$ref": "#/components/schemas/Product"}, 

299 "title": "Products", 

300 "type": "array", 

301 }, 

302 }, 

303 "required": ["id"], 

304 "title": "Order", 

305 "type": "object", 

306 }, 

307 } 

308 

309 def test_pydantic_model(self): (empty)

310 class User(pydantic.BaseModel): (empty)

311 name: str = "" (empty)

312 id: int (empty)

313 

314 broker = self.broker_class() (empty)

315 

316 @broker.subscriber("test") (empty)

317 async def handle(user: User) -> None: ... (empty)

318 

319 schema = self.get_spec(broker).to_jsonable() (empty)

320 

321 payload = schema["components"]["schemas"] (empty)

322 

323 for key, v in payload.items(): (empty)

324 assert key == "User" (empty)

325 assert v == { (empty)

326 "properties": { 

327 "id": {"title": "Id", "type": "integer"}, 

328 "name": {"default": "", "title": "Name", "type": "string"}, 

329 }, 

330 "required": ["id"], 

331 "title": key, 

332 "type": "object", 

333 } 

334 

335 def test_pydantic_model_with_enum(self) -> None: (empty)

336 class Status(str, Enum): (empty)

337 registered = "registered" (empty)

338 banned = "banned" (empty)

339 

340 class User(pydantic.BaseModel): (empty)

341 name: str = "" (empty)

342 id: int (empty)

343 status: Status (empty)

344 

345 broker = self.broker_class() (empty)

346 

347 @broker.subscriber("test") (empty)

348 async def handle(user: User) -> None: ... (empty)

349 

350 schema = self.get_spec(broker).to_jsonable() (empty)

351 

352 payload = schema["components"]["schemas"] (empty)

353 

354 assert payload == { (empty)

355 "Status": IsPartialDict( 

356 { 

357 "enum": ["registered", "banned"], 

358 "title": "Status", 

359 "type": "string", 

360 }, 

361 ), 

362 "User": { 

363 "properties": { 

364 "id": {"title": "Id", "type": "integer"}, 

365 "name": {"default": "", "title": "Name", "type": "string"}, 

366 "status": {"$ref": "#/components/schemas/Status"}, 

367 }, 

368 "required": ["id", "status"], 

369 "title": "User", 

370 "type": "object", 

371 }, 

372 }, payload 

373 

374 def test_pydantic_model_mixed_regular(self) -> None: (empty)

375 class Email(pydantic.BaseModel): (empty)

376 addr: str (empty)

377 

378 class User(pydantic.BaseModel): (empty)

379 name: str = "" (empty)

380 id: int (empty)

381 email: Email (empty)

382 

383 broker = self.broker_class() (empty)

384 

385 @broker.subscriber("test") (empty)

386 async def handle(user: User, description: str = "") -> None: ... (empty)

387 

388 schema = self.get_spec(broker).to_jsonable() (empty)

389 

390 payload = schema["components"]["schemas"] (empty)

391 

392 assert payload == { (empty)

393 "Email": { 

394 "title": "Email", 

395 "type": "object", 

396 "properties": {"addr": {"title": "Addr", "type": "string"}}, 

397 "required": ["addr"], 

398 }, 

399 "User": { 

400 "title": "User", 

401 "type": "object", 

402 "properties": { 

403 "name": {"title": "Name", "default": "", "type": "string"}, 

404 "id": {"title": "Id", "type": "integer"}, 

405 "email": {"$ref": "#/components/schemas/Email"}, 

406 }, 

407 "required": ["id", "email"], 

408 }, 

409 "Handle:Message:Payload": { 

410 "title": "Handle:Message:Payload", 

411 "type": "object", 

412 "properties": { 

413 "user": {"$ref": "#/components/schemas/User"}, 

414 "description": { 

415 "title": "Description", 

416 "default": "", 

417 "type": "string", 

418 }, 

419 }, 

420 "required": ["user"], 

421 }, 

422 } 

423 

424 def test_nested_models_in_union_should_be_in_schemas(self) -> None: (empty)

425 """Test that nested Pydantic models in union types are promoted to components/schemas. 

426 

427 Fixes issue #2443: Nested Pydantic models are not included in AsyncAPI 

428 components/schemas (inplaced instead). 

429 """ 

430 

431 class Email(pydantic.BaseModel): (empty)

432 addr: str (empty)

433 

434 class User(pydantic.BaseModel): (empty)

435 name: str = "" (empty)

436 id: int (empty)

437 email: Email (empty)

438 

439 class Other(pydantic.BaseModel): (empty)

440 id: int (empty)

441 

442 broker = self.broker_class() (empty)

443 

444 @broker.subscriber("test") (empty)

445 async def handle(body: User | Other) -> None: ... (empty)

446 

447 schema = self.get_spec(broker).to_jsonable() (empty)

448 

449 payload = schema["components"]["schemas"] (empty)

450 

451 # Check that nested Email model is promoted to components/schemas 

452 assert "Email" in payload (empty)

453 assert payload["Email"] == { (empty)

454 "title": "Email", 

455 "type": "object", 

456 "properties": {"addr": {"title": "Addr", "type": "string"}}, 

457 "required": ["addr"], 

458 } 

459 

460 def test_nested_models_in_publisher_union_should_be_in_schemas(self) -> None: (empty)

461 """Test that nested Pydantic models in publisher union types are promoted to components/schemas. 

462 

463 Fixes issue #2443: Nested Pydantic models are not included in AsyncAPI 

464 components/schemas (inplaced instead). 

465 """ 

466 

467 class Email(pydantic.BaseModel): (empty)

468 addr: str (empty)

469 

470 class User(pydantic.BaseModel): (empty)

471 name: str = "" (empty)

472 id: int (empty)

473 email: Email (empty)

474 

475 class Other(pydantic.BaseModel): (empty)

476 id: int (empty)

477 

478 broker = self.broker_class() (empty)

479 

480 publisher = broker.publisher("test") (empty)

481 

482 @publisher (empty)

483 def handle0(msg) -> User: ... (empty)

484 

485 @publisher (empty)

486 def handle1(msg) -> Other: ... (empty)

487 

488 schema = self.get_spec(broker).to_jsonable() (empty)

489 

490 payload = schema["components"]["schemas"] (empty)

491 

492 # Check that nested Email model is promoted to components/schemas 

493 assert "Email" in payload (empty)

494 assert payload["Email"] == { (empty)

495 "title": "Email", 

496 "type": "object", 

497 "properties": {"addr": {"title": "Addr", "type": "string"}}, 

498 "required": ["addr"], 

499 } 

500 

501 def test_pydantic_model_with_example(self) -> None: (empty)

502 class User(pydantic.BaseModel): (empty)

503 name: str = "" (empty)

504 id: int (empty)

505 

506 if PYDANTIC_V2: 506 ↛ 513line 506 didn't jump to line 513 because the condition on line 506 was always true(empty)

507 model_config = { (empty)

508 "json_schema_extra": {"examples": [{"name": "john", "id": 1}]}, 

509 } 

510 

511 else: 

512 

513 class Config: 

514 schema_extra = {"examples": [{"name": "john", "id": 1}]} # noqa: RUF012 

515 

516 broker = self.broker_class() (empty)

517 

518 @broker.subscriber("test") (empty)

519 async def handle(user: User) -> None: ... (empty)

520 

521 schema = self.get_spec(broker).to_jsonable() (empty)

522 

523 payload = schema["components"]["schemas"] (empty)

524 

525 for key, v in payload.items(): (empty)

526 assert key == "User" (empty)

527 assert v == { (empty)

528 "examples": [{"id": 1, "name": "john"}], 

529 "properties": { 

530 "id": {"title": "Id", "type": "integer"}, 

531 "name": {"default": "", "title": "Name", "type": "string"}, 

532 }, 

533 "required": ["id"], 

534 "title": "User", 

535 "type": "object", 

536 } 

537 

538 def test_pydantic_model_with_keyword_property(self) -> None: (empty)

539 class TestModel(pydantic.BaseModel): (empty)

540 discriminator: int = 0 (empty)

541 

542 broker = self.broker_class() (empty)

543 

544 @broker.subscriber("test") (empty)

545 async def handle(model: TestModel) -> None: ... (empty)

546 

547 schema = self.get_spec(broker).to_jsonable() (empty)

548 

549 payload = schema["components"]["schemas"] (empty)

550 

551 for key, v in payload.items(): (empty)

552 assert key == "TestModel" (empty)

553 assert v == { (empty)

554 "properties": { 

555 "discriminator": { 

556 "default": 0, 

557 "title": "Discriminator", 

558 "type": "integer", 

559 }, 

560 }, 

561 "title": key, 

562 "type": "object", 

563 } 

564 

565 def test_ignores_depends(self) -> None: (empty)

566 broker = self.broker_class() (empty)

567 

568 def dep(name: str = "") -> str: (empty)

569 return name 

570 

571 def dep2(name2: str) -> str: (empty)

572 return name2 

573 

574 dependencies = (self.dependency_builder(dep2),) (empty)

575 message = self.dependency_builder(dep) (empty)

576 

577 @broker.subscriber("test", dependencies=dependencies) (empty)

578 async def handle(id: int, message=message) -> None: ... (empty)

579 

580 schema = self.get_spec(broker).to_jsonable() (empty)

581 

582 payload = schema["components"]["schemas"] (empty)

583 

584 for key, v in payload.items(): (empty)

585 assert key == "Handle:Message:Payload" (empty)

586 assert v == { (empty)

587 "properties": { 

588 "id": {"title": "Id", "type": "integer"}, 

589 "name": {"default": "", "title": "Name", "type": "string"}, 

590 "name2": {"title": "Name2", "type": "string"}, 

591 }, 

592 "required": ["id", "name2"], 

593 "title": key, 

594 "type": "object", 

595 }, v 

596 

597 @pydantic_v2 (empty)

598 def test_discriminator(self) -> None: (empty)

599 class Sub2(pydantic.BaseModel): (empty)

600 type: Literal["sub2"] (empty)

601 

602 class Sub(pydantic.BaseModel): (empty)

603 type: Literal["sub"] (empty)

604 

605 broker = self.broker_class() (empty)

606 

607 @broker.subscriber("test") (empty)

608 async def handle( (empty)

609 user: Annotated[Sub2 | Sub, pydantic.Field(discriminator="type")], (empty)

610 ): ... 

611 

612 schema = self.get_spec(broker).to_jsonable() (empty)

613 

614 key = next(iter(schema["components"]["messages"].keys())) (empty)

615 

616 assert key == IsStr(regex=r"test[\w:]*:Handle:Message"), key (empty)

617 

618 expected_schema = IsPartialDict({ (empty)

619 "discriminator": "type", 

620 "oneOf": [ 

621 {"$ref": "#/components/schemas/Sub2"}, 

622 {"$ref": "#/components/schemas/Sub"}, 

623 ], 

624 "title": "Handle:Message:Payload", 

625 }) 

626 

627 fastapi_payload = schema["components"]["schemas"].get("Handle:Message:Payload") (empty)

628 if self.is_fastapi: (empty)

629 if fastapi_payload: 629 ↛ 637line 629 didn't jump to line 637 because the condition on line 629 was always true(empty)

630 assert fastapi_payload == IsPartialDict({ (empty)

631 "anyOf": [ 

632 {"$ref": "#/components/schemas/Sub2"}, 

633 {"$ref": "#/components/schemas/Sub"}, 

634 ], 

635 }) 

636 

637 expected_schema = ( (empty)

638 IsPartialDict({"$ref": "#/components/schemas/Handle:Message:Payload"}) 

639 | expected_schema 

640 ) 

641 

642 assert schema["components"]["messages"][key]["payload"] == expected_schema, ( (empty)

643 schema["components"] 

644 ) 

645 

646 assert schema["components"]["schemas"] == IsPartialDict({ (empty)

647 "Sub": { 

648 "properties": { 

649 "type": IsPartialDict({"const": "sub", "title": "Type"}), 

650 }, 

651 "required": ["type"], 

652 "title": "Sub", 

653 "type": "object", 

654 }, 

655 "Sub2": { 

656 "properties": { 

657 "type": IsPartialDict({"const": "sub2", "title": "Type"}), 

658 }, 

659 "required": ["type"], 

660 "title": "Sub2", 

661 "type": "object", 

662 }, 

663 }), schema["components"]["schemas"] 

664 

665 @pydantic_v2 (empty)

666 def test_nested_discriminator(self) -> None: (empty)

667 class Sub2(pydantic.BaseModel): (empty)

668 type: Literal["sub2"] (empty)

669 

670 class Sub(pydantic.BaseModel): (empty)

671 type: Literal["sub"] (empty)

672 

673 class Model(pydantic.BaseModel): (empty)

674 msg: Sub2 | Sub = pydantic.Field(..., discriminator="type") (empty)

675 

676 broker = self.broker_class() (empty)

677 

678 @broker.subscriber("test") (empty)

679 async def handle(user: Model) -> None: ... (empty)

680 

681 schema = self.get_spec(broker).to_jsonable() (empty)

682 

683 key = next(iter(schema["components"]["messages"].keys())) (empty)

684 assert key == IsStr(regex=r"test[\w:]*:Handle:Message") (empty)

685 assert schema["components"] == { (empty)

686 "messages": { 

687 key: IsPartialDict({ 

688 "payload": {"$ref": "#/components/schemas/Model"}, 

689 }), 

690 }, 

691 "schemas": { 

692 "Sub": { 

693 "properties": { 

694 "type": IsPartialDict({"const": "sub", "title": "Type"}), 

695 }, 

696 "required": ["type"], 

697 "title": "Sub", 

698 "type": "object", 

699 }, 

700 "Sub2": { 

701 "properties": { 

702 "type": IsPartialDict({"const": "sub2", "title": "Type"}), 

703 }, 

704 "required": ["type"], 

705 "title": "Sub2", 

706 "type": "object", 

707 }, 

708 "Model": { 

709 "properties": { 

710 "msg": { 

711 "discriminator": "type", 

712 "oneOf": [ 

713 {"$ref": "#/components/schemas/Sub2"}, 

714 {"$ref": "#/components/schemas/Sub"}, 

715 ], 

716 "title": "Msg", 

717 }, 

718 }, 

719 "required": ["msg"], 

720 "title": "Model", 

721 "type": "object", 

722 }, 

723 }, 

724 }, schema["components"] 

725 

726 def test_with_filter(self) -> None: (empty)

727 class User(pydantic.BaseModel): (empty)

728 name: str = "" (empty)

729 id: int (empty)

730 

731 broker = self.broker_class() (empty)

732 

733 sub = broker.subscriber("test") (empty)

734 

735 @sub( (empty)

736 filter=lambda m: m.content_type == "application/json", (empty)

737 ) 

738 async def handle(id: int) -> None: ... (empty)

739 

740 @sub (empty)

741 async def handle_default(msg) -> None: ... (empty)

742 

743 schema = self.get_spec(broker).to_jsonable() (empty)

744 

745 name, message = next(iter(schema["components"]["messages"].items())) (empty)

746 

747 assert name == IsStr(regex=r"test[\w:]*:\[Handle,HandleDefault\]:Message"), name (empty)

748 

749 assert len(message["payload"]["oneOf"]) == 2 (empty)

750 

751 payload = schema["components"]["schemas"] (empty)

752 

753 assert "Handle:Message:Payload" in list(payload.keys()) (empty)

754 assert "HandleDefault:Message:Payload" in list(payload.keys()) (empty)

755 

756 

757class ArgumentsTestcase(FastAPICompatible): (empty)

758 dependency_builder = staticmethod(Depends) (empty)

759 

760 def test_pydantic_field(self) -> None: (empty)

761 broker = self.broker_class() (empty)

762 

763 @broker.subscriber("msg") (empty)

764 async def msg( (empty)

765 msg: pydantic.PositiveInt = pydantic.Field( (empty)

766 1, (empty)

767 description="some field", (empty)

768 title="Perfect", (empty)

769 examples=[1], (empty)