from __future__ import annotations from enum import Enum from typing import List, Optional, Union, Literal, Annotated from uuid import UUID import pytest from aiohttp import web from pydantic import Field, RootModel from pydantic.main import BaseModel from aiohttp_pydantic import PydanticView, oas from aiohttp_pydantic.injectors import Group from aiohttp_pydantic.oas.typing import r200, r201, r204, r404, r400 from aiohttp_pydantic.oas.view import generate_oas class Color(str, Enum): """ Pet color """ RED = "red" GREEN = "green" PINK = "pink" class Lang(str, Enum): EN = 'en' FR = 'fr' class Toy(BaseModel): name: str color: Color class Pet(BaseModel): id: int name: Optional[str] = Field(None) toys: List[Toy] class Error(BaseModel): code: int text: str class Cat(BaseModel): pet_type: Literal['cat'] meows: int class Dog(BaseModel): pet_type: Literal['dog'] barks: float Animal = RootModel[Annotated[Union[Cat, Dog], Field(discriminator='pet_type')]] class PetCollectionView(PydanticView): async def get( self, format: str, lang: Lang = Lang.EN, name: Optional[str] = None, *, promo: Optional[UUID] = None ) -> r200[List[Pet]]: """ Get a list of pets Tags: pet Status Codes: 200: Successful operation OperationId: createPet """ return web.json_response() async def post(self, pet: Pet) -> r201[Pet]: """Create a Pet""" return web.json_response() class PetItemView(PydanticView): async def get( self, id: int, /, size: Union[int, Literal["x", "l", "s"]], day: Union[int, Literal["now"]] = "now", ) -> Union[r200[Pet], r404[Error], r400[Error]]: return web.json_response() async def put(self, id: int, /, pet: Pet): return web.json_response() async def delete(self, id: int, /) -> r204: """ Status Code: 204: Empty but OK """ return web.json_response() class ViewResponseReturnASimpleType(PydanticView): async def get(self) -> r200[int]: """ Status Codes: 200: The new number """ return web.json_response() class DiscriminatedView(PydanticView): async def post(self, /, request: Animal) -> r200[int]: return web.json_response() async def ensure_content_durability(client): """ Reload the page 2 times to ensure that content is always the same note: pydantic can return a cached dict, if a view updates the dict the output will be incoherent """ response_1 = await client.get("/oas/spec") assert response_1.status == 200 assert response_1.content_type == "application/json" content_1 = await response_1.json() response_2 = await client.get("/oas/spec") content_2 = await response_2.json() assert content_1 == content_2 return content_2 @pytest.fixture async def generated_oas(aiohttp_client, event_loop) -> web.Application: app = web.Application() app.router.add_view("/pets", PetCollectionView) app.router.add_view("/pets/{id}", PetItemView) app.router.add_view("/simple-type", ViewResponseReturnASimpleType) app.router.add_view("/animals", DiscriminatedView) oas.setup(app) return await ensure_content_durability(await aiohttp_client(app)) async def test_generated_oas_should_have_components_schemas(generated_oas): assert generated_oas["components"]["schemas"] == { 'Cat': {'properties': {'meows': {'title': 'Meows', 'type': 'integer'}, 'pet_type': {'const': 'cat', 'title': 'Pet Type'}}, 'required': ['pet_type', 'meows'], 'title': 'Cat', 'type': 'object'}, "Color": { "description": "Pet color", "enum": ["red", "green", "pink"], "title": "Color", "type": "string", }, 'Dog': {'properties': {'barks': {'title': 'Barks', 'type': 'number'}, 'pet_type': {'const': 'dog', 'title': 'Pet Type'}}, 'required': ['pet_type', 'barks'], 'title': 'Dog', 'type': 'object'}, 'Error': { 'properties': { 'code': {'title': 'Code', 'type': 'integer'}, 'text': {'title': 'Text', 'type': 'string'}}, 'required': ['code', 'text'], 'title': 'Error', 'type': 'object' }, 'Lang': { 'enum': ['en', 'fr'], 'title': 'Lang', 'type': 'string' }, "Toy": { "properties": { "color": {"$ref": "#/components/schemas/Color"}, "name": {"title": "Name", "type": "string"}, }, "required": ["name", "color"], "title": "Toy", "type": "object", }, 'Pet': { 'properties': { 'id': {'title': 'Id', 'type': 'integer'}, 'name': { 'anyOf': [ {'type': 'string'}, {'type': 'null'} ], 'default': None, 'title': 'Name'}, 'toys': { 'items': {'$ref': '#/components/schemas/Toy'}, 'title': 'Toys', 'type': 'array' } }, 'required': ['id', 'toys'], 'title': 'Pet', 'type': 'object' } } async def test_generated_oas_should_have_pets_paths(generated_oas): assert "/pets" in generated_oas["paths"] async def test_pets_route_should_have_get_method(generated_oas): assert generated_oas["paths"]["/pets"]["get"] == { "description": "Get a list of pets", "operationId": "createPet", "tags": ["pet"], "parameters": [ { "in": "query", "name": "format", "required": True, "schema": {"title": "format", "type": "string"}, }, { 'in': 'query', 'name': 'lang', 'required': False, 'schema': { 'allOf': [{'$ref': '#/components/schemas/Lang'}], 'default': 'en', 'title': 'lang' } }, { "in": "query", "name": "name", "required": False, "schema": { 'anyOf': [{'type': 'string'}, {'type': 'null'}], 'default': None, 'title': 'name' }, }, { "in": "header", "name": "promo", "required": False, "schema": { 'anyOf': [ {'format': 'uuid', 'type': 'string'}, {'type': 'null'} ], 'default': None, 'title': 'promo' }, }, ], "responses": { "200": { "description": "Successful operation", "content": { "application/json": { "schema": { "items": {'$ref': '#/components/schemas/Pet'}, "type": "array", } } }, } }, } async def test_pets_route_should_have_post_method(generated_oas): assert generated_oas["paths"]["/pets"]["post"] == { "description": "Create a Pet", "requestBody": { "content": { "application/json": { "schema": { "properties": { "id": {"title": "Id", "type": "integer"}, "name": { 'anyOf': [ {'type': 'string'}, {'type': 'null'} ], 'default': None, 'title': 'Name' }, "toys": { "items": {"$ref": "#/components/schemas/Toy"}, "title": "Toys", "type": "array", }, }, "required": ["id", "toys"], "title": "Pet", "type": "object", } } } }, "responses": { "201": { "description": "", "content": { "application/json": { "schema": {'$ref': '#/components/schemas/Pet'} } }, } }, } async def test_generated_oas_should_have_pets_id_paths(generated_oas): assert "/pets/{id}" in generated_oas["paths"] async def test_pets_id_route_should_have_delete_method(generated_oas): assert generated_oas["paths"]["/pets/{id}"]["delete"] == { "description": "", "parameters": [ { "in": "path", "name": "id", "required": True, "schema": {"title": "id", "type": "integer"}, } ], "responses": {"204": {"content": {}, "description": "Empty but OK"}}, } async def test_pets_id_route_should_have_get_method(generated_oas): assert generated_oas["paths"]["/pets/{id}"]["get"] == { "parameters": [ { "in": "path", "name": "id", "required": True, "schema": {"title": "id", "type": "integer"}, }, { "in": "query", "name": "size", "required": True, "schema": { "anyOf": [ {"type": "integer"}, {"enum": ["x", "l", "s"], "type": "string"}, ], "title": "size", }, }, { "in": "query", "name": "day", "required": False, "schema": { 'anyOf': [{'type': 'integer'}, {'const': 'now'}], 'default': 'now', 'title': 'day' }, }, ], 'responses': { '200': {'content': {'application/json': {'schema': {'$ref': '#/components/schemas/Pet'}}}, 'description': ''}, '400': {'content': {'application/json': {'schema': {'$ref': '#/components/schemas/Error'}}}, 'description': ''}, '404': {'content': {'application/json': {'schema': {'$ref': '#/components/schemas/Error'}}}, 'description': ''} } } async def test_pets_id_route_should_have_put_method(generated_oas): assert generated_oas["paths"]["/pets/{id}"]["put"] == { "parameters": [ { "in": "path", "name": "id", "required": True, "schema": {"title": "id", "type": "integer"}, } ], "requestBody": { "content": { "application/json": { "schema": { "properties": { "id": {"title": "Id", "type": "integer"}, "name": { 'anyOf': [ {'type': 'string'}, {'type': 'null'} ], 'default': None, 'title': 'Name'}, "toys": { "items": {"$ref": "#/components/schemas/Toy"}, "title": "Toys", "type": "array", }, }, "required": ["id", "toys"], "title": "Pet", "type": "object", } } } }, } async def test_simple_type_route_should_have_get_method(generated_oas): assert generated_oas["paths"]["/simple-type"]["get"] == { "description": "", "responses": { "200": { "content": {"application/json": {"schema": {}}}, "description": "The new number", } }, } async def test_generated_view_info_default(): apps = (web.Application(),) spec = generate_oas(apps) assert spec == { "info": {"title": "Aiohttp pydantic application", "version": "1.0.0"}, "openapi": "3.0.0", } async def test_generated_view_info_as_version(): apps = (web.Application(),) spec = generate_oas(apps, version_spec="test version") assert spec == { "info": {"title": "Aiohttp pydantic application", "version": "test version"}, "openapi": "3.0.0", } async def test_generated_view_info_as_title(): apps = (web.Application(),) spec = generate_oas(apps, title_spec="test title") assert spec == { "info": {"title": "test title", "version": "1.0.0"}, "openapi": "3.0.0", } class Pagination(Group): page: int = 1 page_size: int = 20 async def test_use_parameters_group_should_not_impact_the_oas(aiohttp_client): class PetCollectionView1(PydanticView): async def get(self, page: int = 1, page_size: int = 20) -> r200[List[Pet]]: return web.json_response() class PetCollectionView2(PydanticView): async def get(self, pagination: Pagination) -> r200[List[Pet]]: return web.json_response() app1 = web.Application() app1.router.add_view("/pets", PetCollectionView1) oas.setup(app1) app2 = web.Application() app2.router.add_view("/pets", PetCollectionView2) oas.setup(app2) assert await ensure_content_durability( await aiohttp_client(app1) ) == await ensure_content_durability(await aiohttp_client(app2))