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

1import sys (empty)

2from dataclasses import dataclass (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)

10from fastapi import Depends as APIDepends (empty)

11 

12from faststream import Context (empty)

13from faststream._internal._compat import PYDANTIC_V2 (empty)

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

15from faststream._internal.fastapi import StreamRouter (empty)

16from tests.marks import pydantic_v2 (empty)

17 

18from .basic import AsyncAPI300Factory (empty)

19 

20 

21class FastAPICompatible(AsyncAPI300Factory): (empty)

22 is_fastapi: bool = False (empty)

23 

24 broker_class: BrokerUsecase | StreamRouter (empty)

25 dependency_builder = staticmethod(APIDepends) (empty)

26 

27 def test_default_naming(self) -> None: (empty)

28 broker = self.broker_class() (empty)

29 

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

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

32 

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

34 

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

36 operation_key = tuple(schema["operations"].keys())[0] # noqa: RUF015 (empty)

37 

38 assert channel_key == IsStr(regex=r"test[\w:]*:Handle"), channel_key (empty)

39 assert operation_key == IsStr(regex=r"test[\w:]*:HandleSubscribe"), operation_key (empty)

40 

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

42 broker = self.broker_class() (empty)

43 

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

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

46 

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

48 

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

50 operation_key = tuple(schema["operations"].keys())[0] # noqa: RUF015 (empty)

51 

52 assert channel_key == "custom_name" (empty)

53 assert operation_key == "custom_name" (empty)

54 

55 assert schema["channels"][channel_key]["description"] == "test description" (empty)

56 

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

58 broker = self.broker_class() (empty)

59 

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

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

62 

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

64 

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

66 operation_key = tuple(schema["operations"].keys())[0] # noqa: RUF015 (empty)

67 

68 assert channel_key == "." (empty)

69 assert schema["channels"][channel_key]["address"] == "/" (empty)

70 

71 assert operation_key == ".Subscribe" (empty)

72 

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

74 assert ( (empty)

75 schema["components"]["messages"][".:SubscribeMessage"]["title"] 

76 == "/:SubscribeMessage" 

77 ) 

78 

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

80 assert ( (empty)

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

82 == "/:Message:Payload" 

83 ) 

84 

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

86 broker = self.broker_class() (empty)

87 

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

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

90 """Test description.""" 

91 

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

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

94 

95 assert key == "custom_name" (empty)

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

97 "channels" 

98 ][key]["description"] 

99 

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

101 broker = self.broker_class() (empty)

102 

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

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

105 

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

107 

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

109 

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

111 assert key == "EmptyPayload" (empty)

112 assert v == { (empty)

113 "title": key, 

114 "type": "null", 

115 } 

116 

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

118 broker = self.broker_class() (empty)

119 

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

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

122 

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

124 

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

126 

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

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

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

130 

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

132 broker = self.broker_class() (empty)

133 

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

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

136 

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

138 

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

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

141 

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

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

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

145 

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

147 broker = self.broker_class() (empty)

148 

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

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

151 

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

153 

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

155 

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

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

158 assert v == IsDict( (empty)

159 { 

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

161 "title": key, 

162 }, 

163 ) | IsDict( 

164 { # TODO: remove when deprecating PydanticV1 

165 "title": key, 

166 "type": "integer", 

167 }, 

168 ), v 

169 

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

171 broker = self.broker_class() (empty)

172 

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

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

175 

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

177 

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

179 

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

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

182 assert v == { (empty)

183 "default": 1, 

184 "title": key, 

185 "type": "integer", 

186 } 

187 

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

189 broker = self.broker_class() (empty)

190 

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

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

193 

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

195 

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

197 

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

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

200 assert v == { (empty)

201 "properties": { 

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

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

204 }, 

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

206 "title": key, 

207 "type": "object", 

208 } 

209 

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

211 broker = self.broker_class() (empty)

212 

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

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

215 

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

217 

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

219 

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

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

222 assert v == { (empty)

223 "properties": { 

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

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

226 }, 

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

228 "title": key, 

229 "type": "object", 

230 } 

231 

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

233 broker = self.broker_class() (empty)

234 

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

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

237 

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

239 

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

241 

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

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

244 

245 assert v == { (empty)

246 "properties": { 

247 "another": IsDict( 

248 { 

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

250 "default": None, 

251 "title": "Another", 

252 }, 

253 ) 

254 | IsDict( 

255 { # TODO: remove when deprecating PydanticV1 

256 "title": "Another", 

257 "type": "integer", 

258 }, 

259 ), 

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

261 }, 

262 "required": ["msg"], 

263 "title": key, 

264 "type": "object", 

265 } 

266 

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

268 @dataclass (empty)

269 class User: (empty)

270 id: int (empty)

271 name: str = "" (empty)

272 

273 broker = self.broker_class() (empty)

274 

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

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

277 

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

279 

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

281 

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

283 assert key == "User" (empty)

284 assert v == { (empty)

285 "properties": { 

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

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

288 }, 

289 "required": ["id"], 

290 "title": key, 

291 "type": "object", 

292 } 

293 

294 def test_pydantic_model(self) -> None: (empty)

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

296 name: str = "" (empty)

297 id: int (empty)

298 

299 broker = self.broker_class() (empty)

300 

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

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

303 

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

305 

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

307 

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

309 assert key == "User" (empty)

310 assert v == { (empty)

311 "properties": { 

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

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

314 }, 

315 "required": ["id"], 

316 "title": key, 

317 "type": "object", 

318 } 

319 

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

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

322 registered = "registered" (empty)

323 banned = "banned" (empty)

324 

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

326 name: str = "" (empty)

327 id: int (empty)

328 status: Status (empty)

329 

330 broker = self.broker_class() (empty)

331 

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

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

334 

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

336 

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

338 

339 assert payload == { (empty)

340 "Status": IsPartialDict( 

341 { 

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

343 "title": "Status", 

344 "type": "string", 

345 }, 

346 ), 

347 "User": { 

348 "properties": { 

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

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

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

352 }, 

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

354 "title": "User", 

355 "type": "object", 

356 }, 

357 }, payload 

358 

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

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

361 addr: str (empty)

362 

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

364 name: str = "" (empty)

365 id: int (empty)

366 email: Email (empty)

367 

368 broker = self.broker_class() (empty)

369 

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

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

372 

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

374 

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

376 

377 assert payload == { (empty)

378 "Email": { 

379 "title": "Email", 

380 "type": "object", 

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

382 "required": ["addr"], 

383 }, 

384 "User": { 

385 "title": "User", 

386 "type": "object", 

387 "properties": { 

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

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

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

391 }, 

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

393 }, 

394 "Handle:Message:Payload": { 

395 "title": "Handle:Message:Payload", 

396 "type": "object", 

397 "properties": { 

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

399 "description": { 

400 "title": "Description", 

401 "default": "", 

402 "type": "string", 

403 }, 

404 }, 

405 "required": ["user"], 

406 }, 

407 } 

408 

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

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

411 

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

413 components/schemas (inplaced instead). 

414 """ 

415 

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

417 addr: str (empty)

418 

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

420 name: str = "" (empty)

421 id: int (empty)

422 email: Email (empty)

423 

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

425 id: int (empty)

426 

427 broker = self.broker_class() (empty)

428 

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

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

431 

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

433 

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

435 

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

437 assert "Email" in payload (empty)

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

439 "title": "Email", 

440 "type": "object", 

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

442 "required": ["addr"], 

443 } 

444 

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

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

447 

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

449 components/schemas (inplaced instead). 

450 """ 

451 

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

453 addr: str (empty)

454 

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

456 name: str = "" (empty)

457 id: int (empty)

458 email: Email (empty)

459 

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

461 id: int (empty)

462 

463 broker = self.broker_class() (empty)

464 

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

466 

467 @publisher (empty)

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

469 

470 @publisher (empty)

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

472 

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

474 

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

476 

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

478 assert "Email" in payload (empty)

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

480 "title": "Email", 

481 "type": "object", 

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

483 "required": ["addr"], 

484 } 

485 

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

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

488 name: str = "" (empty)

489 id: int (empty)

490 

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

492 model_config = { (empty)

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

494 } 

495 

496 else: 

497 

498 class Config: 

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

500 

501 broker = self.broker_class() (empty)

502 

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

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

505 

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

507 

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

509 

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

511 assert key == "User" (empty)

512 assert v == { (empty)

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

514 "properties": { 

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

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

517 }, 

518 "required": ["id"], 

519 "title": "User", 

520 "type": "object", 

521 } 

522 

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

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

525 name: str = "" (empty)

526 id: int (empty)

527 

528 broker = self.broker_class() (empty)

529 

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

531 

532 @sub( # pragma: no branch (empty)

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

534 ) 

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

536 

537 @sub (empty)

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

539 

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

541 

542 assert ( (empty)

543 len( 

544 next(iter(schema["components"]["messages"].values()))["payload"]["oneOf"], 

545 ) 

546 == 2 

547 ) 

548 

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

550 

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

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

553 

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

555 broker = self.broker_class() (empty)

556 

557 def dep(name: str = ""): (empty)

558 return name 

559 

560 def dep2(name2: str): (empty)

561 return name2 

562 

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

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

565 

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

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

568 

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

570 

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

572 

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

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

575 assert v == { (empty)

576 "properties": { 

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

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

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

580 }, 

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

582 "title": key, 

583 "type": "object", 

584 }, v 

585 

586 @pydantic_v2 (empty)

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

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

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

590 

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

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

593 

594 broker = self.broker_class() (empty)

595 

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

597 async def handle( (empty)

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

599 ): ... 

600 

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

602 

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

604 

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

606 

607 p = schema["components"]["messages"][key]["payload"] (empty)

608 assert p == IsPartialDict({ (empty)

609 "$ref": "#/components/schemas/Handle:Message:Payload", 

610 }), p 

611 

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

613 "Sub": { 

614 "properties": { 

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

616 }, 

617 "required": ["type"], 

618 "title": "Sub", 

619 "type": "object", 

620 }, 

621 "Sub2": { 

622 "properties": { 

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

624 }, 

625 "required": ["type"], 

626 "title": "Sub2", 

627 "type": "object", 

628 }, 

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

630 

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

632 

633 discriminator_payload = IsPartialDict({ (empty)

634 "discriminator": "type", 

635 "oneOf": [ 

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

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

638 ], 

639 "title": "Handle:Message:Payload", 

640 }) 

641 

642 if self.is_fastapi: (empty)

643 assert ( (empty)

644 payload 

645 == IsPartialDict({ 

646 "anyOf": [ 

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

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

649 ], 

650 }) 

651 | discriminator_payload 

652 ), payload 

653 

654 else: 

655 assert payload == discriminator_payload (empty)

656 

657 @pydantic_v2 (empty)

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

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

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

661 

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

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

664 

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

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

667 

668 broker = self.broker_class() (empty)

669 

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

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

672 

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

674 

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

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

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

678 "messages": { 

679 key: { 

680 "title": key, 

681 "correlationId": {"location": "$message.header#/correlation_id"}, 

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

683 }, 

684 }, 

685 "schemas": { 

686 "Sub": { 

687 "properties": { 

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

689 }, 

690 "required": ["type"], 

691 "title": "Sub", 

692 "type": "object", 

693 }, 

694 "Sub2": { 

695 "properties": { 

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

697 }, 

698 "required": ["type"], 

699 "title": "Sub2", 

700 "type": "object", 

701 }, 

702 "Model": { 

703 "properties": { 

704 "msg": { 

705 "discriminator": "type", 

706 "oneOf": [ 

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

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

709 ], 

710 "title": "Msg", 

711 }, 

712 }, 

713 "required": ["msg"], 

714 "title": "Model", 

715 "type": "object", 

716 }, 

717 }, 

718 }, schema["components"] 

719 

720 

721class ArgumentsTestcase(FastAPICompatible): (empty)

722 dependency_builder = staticmethod(Depends) (empty)

723 

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

725 broker = self.broker_class() (empty)

726 

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

728 async def msg( (empty)

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

730 1, (empty)

731 description="some field", (empty)

732 title="Perfect", (empty)

733 examples=[1], (empty)

734 ), 

735 ) -> None: ... (empty)

736 

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

738 

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

740 

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

742 assert key == "Perfect" (empty)

743 

744 assert v == { (empty)

745 "default": 1, 

746 "description": "some field", 

747 "examples": [1], 

748 "exclusiveMinimum": 0, 

749 "title": "Perfect", 

750 "type": "integer", 

751 } 

752 

753 def test_ignores_custom_field(self) -> None: (empty)

754 broker = self.broker_class() (empty)

755 

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

757 async def handle( (empty)

758 id: int, (empty)

759 user: str | None = None, (empty)

760 message=Context(), (empty)

761 ) -> None: ... (empty)

762 

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

764 

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

766 

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