|
4 | 4 | from pydantic import BaseModel |
5 | 5 |
|
6 | 6 | from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response |
7 | | -from aws_lambda_powertools.event_handler.router import Router |
8 | 7 |
|
9 | 8 |
|
10 | 9 | def test_openapi_default_response(): |
@@ -371,325 +370,3 @@ def handler() -> UserResponse: |
371 | 370 | assert "example2" in examples |
372 | 371 | assert examples["example2"].summary == "Example 2" |
373 | 372 | assert examples["example2"].value["id"] == 2 |
374 | | - |
375 | | - |
376 | | -def test_openapi_response_encoding_preserved_with_model(): |
377 | | - """Test that encoding is preserved when using model in response content""" |
378 | | - app = APIGatewayRestResolver(enable_validation=True) |
379 | | - |
380 | | - class FileUploadResponse(BaseModel): |
381 | | - file_id: str |
382 | | - filename: str |
383 | | - content: bytes |
384 | | - |
385 | | - @app.post( |
386 | | - "/upload", |
387 | | - responses={ |
388 | | - 200: { |
389 | | - "description": "File upload response", |
390 | | - "content": { |
391 | | - "multipart/form-data": { |
392 | | - "model": FileUploadResponse, |
393 | | - "encoding": { |
394 | | - "content": { |
395 | | - "contentType": "application/octet-stream", |
396 | | - "headers": { |
397 | | - "X-Custom-Header": { |
398 | | - "description": "Custom encoding header", |
399 | | - "schema": {"type": "string"}, |
400 | | - }, |
401 | | - }, |
402 | | - }, |
403 | | - }, |
404 | | - }, |
405 | | - }, |
406 | | - }, |
407 | | - }, |
408 | | - ) |
409 | | - def upload_file() -> FileUploadResponse: |
410 | | - return FileUploadResponse(file_id="123", filename="test.pdf", content=b"") |
411 | | - |
412 | | - schema = app.get_openapi_schema() |
413 | | - content = schema.paths["/upload"].post.responses[200].content["multipart/form-data"] |
414 | | - |
415 | | - # Verify model schema is present |
416 | | - assert content.schema_.ref == "#/components/schemas/FileUploadResponse" |
417 | | - |
418 | | - # Verify encoding is preserved |
419 | | - encoding = content.encoding |
420 | | - |
421 | | - assert "content" in encoding |
422 | | - assert encoding["content"].contentType == "application/octet-stream" |
423 | | - assert encoding["content"].headers is not None |
424 | | - assert "X-Custom-Header" in encoding["content"].headers |
425 | | - |
426 | | - |
427 | | -def test_openapi_response_all_fields_together(): |
428 | | - """Test response with headers, links, examples, and encoding all together""" |
429 | | - app = APIGatewayRestResolver(enable_validation=True) |
430 | | - |
431 | | - class DataResponse(BaseModel): |
432 | | - data: str |
433 | | - timestamp: int |
434 | | - |
435 | | - @app.get( |
436 | | - "/data", |
437 | | - responses={ |
438 | | - 200: { |
439 | | - "description": "Data response with all fields", |
440 | | - "headers": { |
441 | | - "X-Total-Count": { |
442 | | - "description": "Total count of items", |
443 | | - "schema": {"type": "integer"}, |
444 | | - }, |
445 | | - "X-Page": { |
446 | | - "description": "Current page", |
447 | | - "schema": {"type": "integer"}, |
448 | | - }, |
449 | | - }, |
450 | | - "content": { |
451 | | - "application/json": { |
452 | | - "model": DataResponse, |
453 | | - "examples": { |
454 | | - "success": { |
455 | | - "summary": "Successful response", |
456 | | - "value": {"data": "test", "timestamp": 1234567890}, |
457 | | - }, |
458 | | - }, |
459 | | - "encoding": { |
460 | | - "data": { |
461 | | - "contentType": "text/plain", |
462 | | - }, |
463 | | - }, |
464 | | - }, |
465 | | - }, |
466 | | - "links": { |
467 | | - "next": { |
468 | | - "operationId": "getNextPage", |
469 | | - "parameters": {"page": "$response.headers.X-Page + 1"}, |
470 | | - }, |
471 | | - }, |
472 | | - }, |
473 | | - }, |
474 | | - ) |
475 | | - def get_data() -> DataResponse: |
476 | | - return DataResponse(data="test", timestamp=1234567890) |
477 | | - |
478 | | - schema = app.get_openapi_schema() |
479 | | - response = schema.paths["/data"].get.responses[200] |
480 | | - |
481 | | - # Check headers |
482 | | - assert "X-Total-Count" in response.headers |
483 | | - assert "X-Page" in response.headers |
484 | | - |
485 | | - # Check content with model, examples, and encoding |
486 | | - content = response.content["application/json"] |
487 | | - assert content.schema_.ref == "#/components/schemas/DataResponse" |
488 | | - assert "success" in content.examples |
489 | | - assert "data" in content.encoding |
490 | | - |
491 | | - # Check links |
492 | | - assert "next" in response.links |
493 | | - assert response.links["next"].operationId == "getNextPage" |
494 | | - |
495 | | - |
496 | | -def test_openapi_response_backward_compatibility(): |
497 | | - """Test that existing response definitions still work without new fields""" |
498 | | - app = APIGatewayRestResolver(enable_validation=True) |
499 | | - |
500 | | - class SimpleResponse(BaseModel): |
501 | | - message: str |
502 | | - |
503 | | - # Test 1: Simple response with just description |
504 | | - @app.get("/simple", responses={200: {"description": "Simple response"}}) |
505 | | - def simple_handler(): |
506 | | - return {"message": "hello"} |
507 | | - |
508 | | - # Test 2: Response with model only |
509 | | - @app.get( |
510 | | - "/with-model", |
511 | | - responses={ |
512 | | - 200: { |
513 | | - "description": "With model", |
514 | | - "content": {"application/json": {"model": SimpleResponse}}, |
515 | | - }, |
516 | | - }, |
517 | | - ) |
518 | | - def model_handler() -> SimpleResponse: |
519 | | - return SimpleResponse(message="test") |
520 | | - |
521 | | - # Test 3: Response with schema only |
522 | | - @app.get( |
523 | | - "/with-schema", |
524 | | - responses={ |
525 | | - 200: { |
526 | | - "description": "With schema", |
527 | | - "content": { |
528 | | - "application/json": {"schema": {"type": "object", "properties": {"msg": {"type": "string"}}}}, |
529 | | - }, |
530 | | - }, |
531 | | - }, |
532 | | - ) |
533 | | - def schema_handler(): |
534 | | - return {"msg": "test"} |
535 | | - |
536 | | - schema = app.get_openapi_schema() |
537 | | - |
538 | | - # Verify all endpoints work |
539 | | - assert "/simple" in schema.paths |
540 | | - assert "/with-model" in schema.paths |
541 | | - assert "/with-schema" in schema.paths |
542 | | - |
543 | | - # Check simple response |
544 | | - simple_response = schema.paths["/simple"].get.responses[200] |
545 | | - assert simple_response.description == "Simple response" |
546 | | - |
547 | | - # Check model response |
548 | | - model_response = schema.paths["/with-model"].get.responses[200] |
549 | | - assert model_response.content["application/json"].schema_.ref == "#/components/schemas/SimpleResponse" |
550 | | - |
551 | | - # Check schema response |
552 | | - schema_response = schema.paths["/with-schema"].get.responses[200] |
553 | | - assert schema_response.content["application/json"].schema_.type == "object" |
554 | | - |
555 | | - |
556 | | -def test_openapi_response_empty_optional_fields(): |
557 | | - """Test that empty optional fields are handled correctly""" |
558 | | - app = APIGatewayRestResolver(enable_validation=True) |
559 | | - |
560 | | - @app.get( |
561 | | - "/empty", |
562 | | - responses={ |
563 | | - 200: { |
564 | | - "description": "Response with empty optional fields", |
565 | | - "headers": {}, # Empty headers |
566 | | - "links": {}, # Empty links |
567 | | - "content": { |
568 | | - "application/json": { |
569 | | - "schema": {"type": "object"}, |
570 | | - "examples": {}, # Empty examples |
571 | | - "encoding": {}, # Empty encoding |
572 | | - }, |
573 | | - }, |
574 | | - }, |
575 | | - }, |
576 | | - ) |
577 | | - def empty_handler(): |
578 | | - return {} |
579 | | - |
580 | | - schema = app.get_openapi_schema() |
581 | | - response = schema.paths["/empty"].get.responses[200] |
582 | | - |
583 | | - # Empty dicts should still be present in the schema |
584 | | - assert response.headers == {} |
585 | | - assert response.links == {} |
586 | | - |
587 | | - content = response.content["application/json"] |
588 | | - |
589 | | - # Check if examples and encoding are empty or None (both are valid) |
590 | | - assert content.examples == {} or content.examples is None |
591 | | - assert content.encoding == {} or content.encoding is None |
592 | | - |
593 | | - |
594 | | -def test_openapi_response_multiple_content_types_with_fields(): |
595 | | - """Test response with multiple content types each having their own fields""" |
596 | | - app = APIGatewayRestResolver(enable_validation=True) |
597 | | - |
598 | | - class JsonResponse(BaseModel): |
599 | | - data: str |
600 | | - |
601 | | - @app.get( |
602 | | - "/multi-content", |
603 | | - responses={ |
604 | | - 200: { |
605 | | - "description": "Multiple content types", |
606 | | - "content": { |
607 | | - "application/json": { |
608 | | - "model": JsonResponse, |
609 | | - "examples": { |
610 | | - "json_example": {"value": {"data": "json_data"}}, |
611 | | - }, |
612 | | - }, |
613 | | - "application/xml": { |
614 | | - "schema": {"type": "string"}, |
615 | | - "examples": { |
616 | | - "xml_example": {"value": "<data>xml_data</data>"}, |
617 | | - }, |
618 | | - }, |
619 | | - "text/plain": { |
620 | | - "schema": {"type": "string"}, |
621 | | - "examples": { |
622 | | - "text_example": {"value": "plain text data"}, |
623 | | - }, |
624 | | - }, |
625 | | - }, |
626 | | - }, |
627 | | - }, |
628 | | - ) |
629 | | - def multi_content_handler(): |
630 | | - return {"data": "test"} |
631 | | - |
632 | | - schema = app.get_openapi_schema() |
633 | | - response = schema.paths["/multi-content"].get.responses[200] |
634 | | - |
635 | | - # Check JSON content |
636 | | - json_content = response.content["application/json"] |
637 | | - assert json_content.schema_.ref == "#/components/schemas/JsonResponse" |
638 | | - assert "json_example" in json_content.examples |
639 | | - |
640 | | - # Check XML content |
641 | | - xml_content = response.content["application/xml"] |
642 | | - assert xml_content.schema_.type == "string" |
643 | | - assert "xml_example" in xml_content.examples |
644 | | - |
645 | | - # Check plain text content |
646 | | - text_content = response.content["text/plain"] |
647 | | - assert text_content.schema_.type == "string" |
648 | | - assert "text_example" in text_content.examples |
649 | | - |
650 | | - |
651 | | -def test_openapi_response_with_router(): |
652 | | - """Test that new response fields work with Router""" |
653 | | - app = APIGatewayRestResolver(enable_validation=True) |
654 | | - router = Router() |
655 | | - |
656 | | - class RouterResponse(BaseModel): |
657 | | - result: str |
658 | | - |
659 | | - @router.get( |
660 | | - "/router-test", |
661 | | - responses={ |
662 | | - 200: { |
663 | | - "description": "Router response", |
664 | | - "headers": { |
665 | | - "X-Router-Header": { |
666 | | - "description": "Header from router", |
667 | | - "schema": {"type": "string"}, |
668 | | - }, |
669 | | - }, |
670 | | - "content": { |
671 | | - "application/json": { |
672 | | - "model": RouterResponse, |
673 | | - "examples": { |
674 | | - "router_example": {"value": {"result": "from_router"}}, |
675 | | - }, |
676 | | - }, |
677 | | - }, |
678 | | - }, |
679 | | - }, |
680 | | - ) |
681 | | - def router_handler() -> RouterResponse: |
682 | | - return RouterResponse(result="test") |
683 | | - |
684 | | - app.include_router(router) |
685 | | - schema = app.get_openapi_schema() |
686 | | - |
687 | | - response = schema.paths["/router-test"].get.responses[200] |
688 | | - |
689 | | - # Verify headers |
690 | | - assert "X-Router-Header" in response.headers |
691 | | - |
692 | | - # Verify content with model and examples |
693 | | - content = response.content["application/json"] |
694 | | - assert content.schema_.ref == "#/components/schemas/RouterResponse" |
695 | | - assert "router_example" in content.examples |
0 commit comments