from __future__ import annotations from enum import Enum from typing import Optional, List from pydantic import Field from aiohttp import web from aiohttp_pydantic import PydanticView from aiohttp_pydantic.injectors import Group class ArticleView(PydanticView): async def get( self, with_comments: bool, age: Optional[int] = None, nb_items: int = 7, tags: List[str] = Field(default_factory=list), ): return web.json_response( { "with_comments": with_comments, "age": age, "nb_items": nb_items, "tags": tags, } ) class Pagination(Group): page_num: int page_size: int = 20 @property def num(self) -> int: return self.page_num @property def size(self) -> int: return self.page_size class ArticleViewWithPaginationGroup(PydanticView): async def get( self, with_comments: bool, page: Pagination, ): return web.json_response( { "with_comments": with_comments, "page_num": page.num, "page_size": page.size, } ) class Lang(str, Enum): EN = 'en' FR = 'fr' class ArticleViewWithEnumInQuery(PydanticView): async def get(self, lang: Lang): return web.json_response( { "lang": lang } ) async def test_get_article_without_required_qs_should_return_an_error_message( aiohttp_client, event_loop ): app = web.Application() app.router.add_view("/article", ArticleView) client = await aiohttp_client(app) resp = await client.get("/article") assert resp.status == 400 assert resp.content_type == "application/json" assert await resp.json() == [ { 'input': {}, 'loc': ['with_comments'], 'loc_in': 'query string', 'msg': 'Field required', 'type': 'missing', 'url': 'https://errors.pydantic.dev/2.5/v/missing' } ] async def test_get_article_with_wrong_qs_type_should_return_an_error_message( aiohttp_client, event_loop ): app = web.Application() app.router.add_view("/article", ArticleView) client = await aiohttp_client(app) resp = await client.get("/article", params={"with_comments": "foo"}) assert resp.status == 400 assert resp.content_type == "application/json" assert await resp.json() == [ { 'input': 'foo', 'loc': ['with_comments'], 'loc_in': 'query string', 'msg': 'Input should be a valid boolean, unable to interpret input', 'type': 'bool_parsing', 'url': 'https://errors.pydantic.dev/2.5/v/bool_parsing' } ] async def test_get_article_with_valid_qs_should_return_the_parsed_type( aiohttp_client, event_loop ): app = web.Application() app.router.add_view("/article", ArticleView) client = await aiohttp_client(app) resp = await client.get("/article", params={"with_comments": "yes", "age": 3}) assert resp.status == 200 assert resp.content_type == "application/json" assert await resp.json() == { "with_comments": True, "age": 3, "nb_items": 7, "tags": [], } async def test_get_article_with_valid_qs_and_omitted_optional_should_return_default_value( aiohttp_client, event_loop ): app = web.Application() app.router.add_view("/article", ArticleView) client = await aiohttp_client(app) resp = await client.get("/article", params={"with_comments": "yes"}) assert await resp.json() == { "with_comments": True, "age": None, "nb_items": 7, "tags": [], } assert resp.status == 200 assert resp.content_type == "application/json" async def test_get_article_with_multiple_value_for_qs_age_must_failed( aiohttp_client, event_loop ): app = web.Application() app.router.add_view("/article", ArticleView) client = await aiohttp_client(app) resp = await client.get("/article", params={"age": ["2", "3"], "with_comments": 1}) assert await resp.json() == [ { 'input': ['2', '3'], 'loc': ['age'], 'loc_in': 'query string', 'msg': 'Input should be a valid integer', 'type': 'int_type', 'url': 'https://errors.pydantic.dev/2.5/v/int_type' } ] assert resp.status == 400 assert resp.content_type == "application/json" async def test_get_article_with_multiple_value_of_tags(aiohttp_client, event_loop): app = web.Application() app.router.add_view("/article", ArticleView) client = await aiohttp_client(app) resp = await client.get( "/article", params={"age": 2, "with_comments": 1, "tags": ["aa", "bb"]} ) assert await resp.json() == { "age": 2, "nb_items": 7, "tags": ["aa", "bb"], "with_comments": True, } assert resp.status == 200 assert resp.content_type == "application/json" async def test_get_article_with_one_value_of_tags_must_be_a_list(aiohttp_client, event_loop): app = web.Application() app.router.add_view("/article", ArticleView) client = await aiohttp_client(app) resp = await client.get( "/article", params={"age": 2, "with_comments": 1, "tags": ["aa"]} ) assert await resp.json() == { "age": 2, "nb_items": 7, "tags": ["aa"], "with_comments": True, } assert resp.status == 200 assert resp.content_type == "application/json" async def test_get_article_without_required_field_page(aiohttp_client, event_loop): app = web.Application() app.router.add_view("/article", ArticleViewWithPaginationGroup) client = await aiohttp_client(app) resp = await client.get("/article", params={"with_comments": 1}) assert await resp.json() == [ { 'input': {'with_comments': '1'}, 'loc': ['page_num'], 'loc_in': 'query string', 'msg': 'Field required', 'type': 'missing', 'url': 'https://errors.pydantic.dev/2.5/v/missing' } ] assert resp.status == 400 assert resp.content_type == "application/json" async def test_get_article_with_page(aiohttp_client, event_loop): app = web.Application() app.router.add_view("/article", ArticleViewWithPaginationGroup) client = await aiohttp_client(app) resp = await client.get("/article", params={"with_comments": 1, "page_num": 2}) assert await resp.json() == {"page_num": 2, "page_size": 20, "with_comments": True} assert resp.status == 200 assert resp.content_type == "application/json" async def test_get_article_with_page_and_page_size(aiohttp_client, event_loop): app = web.Application() app.router.add_view("/article", ArticleViewWithPaginationGroup) client = await aiohttp_client(app) resp = await client.get( "/article", params={"with_comments": 1, "page_num": 1, "page_size": 10} ) assert await resp.json() == {"page_num": 1, "page_size": 10, "with_comments": True} assert resp.status == 200 assert resp.content_type == "application/json" async def test_get_article_with_enum_in_query(aiohttp_client, event_loop): app = web.Application() app.router.add_view("/article", ArticleViewWithEnumInQuery) client = await aiohttp_client(app) resp = await client.get( "/article", params={"lang": Lang.EN.value} ) assert await resp.json() == {'lang': Lang.EN} assert resp.status == 200 assert resp.content_type == "application/json" async def test_get_article_with_page_and_wrong_page_size(aiohttp_client, event_loop): app = web.Application() app.router.add_view("/article", ArticleViewWithPaginationGroup) client = await aiohttp_client(app) resp = await client.get( "/article", params={"with_comments": 1, "page_num": 1, "page_size": "large"} ) assert await resp.json() == [ { 'input': 'large', 'loc': ['page_size'], 'loc_in': 'query string', 'msg': 'Input should be a valid integer, unable to parse string as an ' 'integer', 'type': 'int_parsing', 'url': 'https://errors.pydantic.dev/2.5/v/int_parsing' } ] assert resp.status == 400 assert resp.content_type == "application/json"