from __future__ import annotations from enum import Enum from typing import List, Optional, Union, Literal from uuid import UUID import pytest from aiohttp import web 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 from aiohttp_pydantic.oas.view import generate_oas class Color(str, Enum): RED = "red" GREEN = "green" PINK = "pink" class Toy(BaseModel): name: str color: Color class Pet(BaseModel): id: int name: str toys: List[Toy] class PetCollectionView(PydanticView): async def get( self, format: str, name: Optional[str] = None, *, promo: Optional[UUID] = None ) -> r200[List[Pet]]: """ Get a list of pets Tags: pet Status Codes: 200: Successful operation """ 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]: 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() 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, 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) 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"] == { "Color": { "description": "An enumeration.", "enum": ["red", "green", "pink"], "title": "Color", "type": "string", }, "Toy": { "properties": { "color": {"$ref": "#/components/schemas/Color"}, "name": {"title": "Name", "type": "string"}, }, "required": ["name", "color"], "title": "Toy", "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", "tags": ["pet"], "parameters": [ { "in": "query", "name": "format", "required": True, "schema": {"title": "format", "type": "string"}, }, { "in": "query", "name": "name", "required": False, "schema": {"title": "name", "type": "string"}, }, { "in": "header", "name": "promo", "required": False, "schema": {"format": "uuid", "title": "promo", "type": "string"}, }, ], "responses": { "200": { "description": "Successful operation", "content": { "application/json": { "schema": { "items": { "properties": { "id": {"title": "Id", "type": "integer"}, "name": {"title": "Name", "type": "string"}, "toys": { "items": {"$ref": "#/components/schemas/Toy"}, "title": "Toys", "type": "array", }, }, "required": ["id", "name", "toys"], "title": "Pet", "type": "object", }, "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": {"title": "Name", "type": "string"}, "toys": { "items": {"$ref": "#/components/schemas/Toy"}, "title": "Toys", "type": "array", }, }, "required": ["id", "name", "toys"], "title": "Pet", "type": "object", } } } }, "responses": { "201": { "description": "", "content": { "application/json": { "schema": { "properties": { "id": {"title": "Id", "type": "integer"}, "name": {"title": "Name", "type": "string"}, "toys": { "items": {"$ref": "#/components/schemas/Toy"}, "title": "Toys", "type": "array", }, }, "required": ["id", "name", "toys"], "title": "Pet", "type": "object", } } }, } }, } 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"}, {"enum": ["now"], "type": "string"}], "default": "now", "title": "day", }, }, ], "responses": { "200": { "description": "", "content": { "application/json": { "schema": { "properties": { "id": {"title": "Id", "type": "integer"}, "name": {"title": "Name", "type": "string"}, "toys": { "items": {"$ref": "#/components/schemas/Toy"}, "title": "Toys", "type": "array", }, }, "required": ["id", "name", "toys"], "title": "Pet", "type": "object", } } }, }, "404": {"description": "", "content": {}}, }, } 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": {"title": "Name", "type": "string"}, "toys": { "items": {"$ref": "#/components/schemas/Toy"}, "title": "Toys", "type": "array", }, }, "required": ["id", "name", "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))