Merge pull request #6 from Maillol/increase_oas
Increase OAS description
This commit is contained in:
MAILLOL Vincent
GitHub
commit
cd8422bde3
49
README.rst
49
README.rst
@ -217,6 +217,9 @@ For example *r200[List[Pet]]* means the server responses with
|
||||
the status code 200 and the response content is a List of Pet where Pet will be
|
||||
defined using a pydantic.BaseModel
|
||||
|
||||
The docstring of methods will be parsed to fill the descriptions in the
|
||||
Open Api Specification.
|
||||
|
||||
|
||||
.. code-block:: python3
|
||||
|
||||
@ -235,20 +238,42 @@ defined using a pydantic.BaseModel
|
||||
|
||||
class PetCollectionView(PydanticView):
|
||||
async def get(self) -> r200[List[Pet]]:
|
||||
"""
|
||||
Find all pets
|
||||
"""
|
||||
pets = self.request.app["model"].list_pets()
|
||||
return web.json_response([pet.dict() for pet in pets])
|
||||
|
||||
async def post(self, pet: Pet) -> r201[Pet]:
|
||||
"""
|
||||
Add a new pet to the store
|
||||
|
||||
Status Codes:
|
||||
201: The pet is created
|
||||
"""
|
||||
self.request.app["model"].add_pet(pet)
|
||||
return web.json_response(pet.dict())
|
||||
|
||||
|
||||
class PetItemView(PydanticView):
|
||||
async def get(self, id: int, /) -> Union[r200[Pet], r404[Error]]:
|
||||
"""
|
||||
Find a pet by ID
|
||||
|
||||
Status Codes:
|
||||
200: Successful operation
|
||||
404: Pet not found
|
||||
"""
|
||||
pet = self.request.app["model"].find_pet(id)
|
||||
return web.json_response(pet.dict())
|
||||
|
||||
async def put(self, id: int, /, pet: Pet) -> r200[Pet]:
|
||||
"""
|
||||
Update an existing pet
|
||||
|
||||
Status Codes:
|
||||
200: successful operation
|
||||
"""
|
||||
self.request.app["model"].update_pet(id, pet)
|
||||
return web.json_response(pet.dict())
|
||||
|
||||
@ -270,12 +295,34 @@ Have a look at `demo`_ for a complete example
|
||||
|
||||
Go to http://127.0.0.1:8080/oas
|
||||
|
||||
You can generate the OAS in a json file using the command:
|
||||
You can generate the OAS in a json or yaml file using the aiohttp_pydantic.oas command:
|
||||
|
||||
.. code-block:: bash
|
||||
|
||||
python -m aiohttp_pydantic.oas demo.main
|
||||
|
||||
.. code-block:: bash
|
||||
$ python3 -m aiohttp_pydantic.oas --help
|
||||
usage: __main__.py [-h] [-b FILE] [-o FILE] [-f FORMAT] [APP [APP ...]]
|
||||
|
||||
Generate Open API Specification
|
||||
|
||||
positional arguments:
|
||||
APP The name of the module containing the asyncio.web.Application. By default the variable named
|
||||
'app' is loaded but you can define an other variable name ending the name of module with :
|
||||
characters and the name of variable. Example: my_package.my_module:my_app If your
|
||||
asyncio.web.Application is returned by a function, you can use the syntax:
|
||||
my_package.my_module:my_app()
|
||||
|
||||
optional arguments:
|
||||
-h, --help show this help message and exit
|
||||
-b FILE, --base-oas-file FILE
|
||||
A file that will be used as base to generate OAS
|
||||
-o FILE, --output FILE
|
||||
File to write the output
|
||||
-f FORMAT, --format FORMAT
|
||||
The output format, can be 'json' or 'yaml' (default is json)
|
||||
|
||||
|
||||
.. _demo: https://github.com/Maillol/aiohttp-pydantic/tree/main/demo
|
||||
.. _aiohttp view: https://docs.aiohttp.org/en/stable/web_quickstart.html#class-based-views
|
||||
|
||||
@ -1,10 +1,28 @@
|
||||
import argparse
|
||||
import importlib
|
||||
import json
|
||||
|
||||
from typing import Dict, Protocol, Optional, Callable
|
||||
import sys
|
||||
from .view import generate_oas
|
||||
|
||||
|
||||
class YamlModule(Protocol):
|
||||
"""
|
||||
Yaml Module type hint
|
||||
"""
|
||||
|
||||
def dump(self, data) -> str:
|
||||
pass
|
||||
|
||||
|
||||
yaml: Optional[YamlModule]
|
||||
|
||||
try:
|
||||
import yaml
|
||||
except ImportError:
|
||||
yaml = None
|
||||
|
||||
|
||||
def application_type(value):
|
||||
"""
|
||||
Return aiohttp application defined in the value.
|
||||
@ -26,6 +44,35 @@ def application_type(value):
|
||||
raise argparse.ArgumentTypeError(error) from error
|
||||
|
||||
|
||||
def base_oas_file_type(value) -> Dict:
|
||||
"""
|
||||
Load base oas file
|
||||
"""
|
||||
try:
|
||||
with open(value) as oas_file:
|
||||
data = oas_file.read()
|
||||
except OSError as error:
|
||||
raise argparse.ArgumentTypeError(error) from error
|
||||
|
||||
return json.loads(data)
|
||||
|
||||
|
||||
def format_type(value) -> Callable:
|
||||
"""
|
||||
Date Dumper one of (json, yaml)
|
||||
"""
|
||||
dumpers = {"json": lambda data: json.dumps(data, sort_keys=True, indent=4)}
|
||||
if yaml is not None:
|
||||
dumpers["yaml"] = yaml.dump
|
||||
|
||||
try:
|
||||
return dumpers[value]
|
||||
except KeyError:
|
||||
raise argparse.ArgumentTypeError(
|
||||
f"Wrong format value. (allowed values: {tuple(dumpers.keys())})"
|
||||
) from None
|
||||
|
||||
|
||||
def setup(parser: argparse.ArgumentParser):
|
||||
parser.add_argument(
|
||||
"apps",
|
||||
@ -35,11 +82,52 @@ def setup(parser: argparse.ArgumentParser):
|
||||
help="The name of the module containing the asyncio.web.Application."
|
||||
" By default the variable named 'app' is loaded but you can define"
|
||||
" an other variable name ending the name of module with : characters"
|
||||
" and the name of variable. Example: my_package.my_module:my_app",
|
||||
" and the name of variable. Example: my_package.my_module:my_app"
|
||||
" If your asyncio.web.Application is returned by a function, you can"
|
||||
" use the syntax: my_package.my_module:my_app()",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-b",
|
||||
"--base-oas-file",
|
||||
metavar="FILE",
|
||||
dest="base",
|
||||
type=base_oas_file_type,
|
||||
help="A file that will be used as base to generate OAS",
|
||||
default={},
|
||||
)
|
||||
parser.add_argument(
|
||||
"-o",
|
||||
"--output",
|
||||
metavar="FILE",
|
||||
type=argparse.FileType("w"),
|
||||
help="File to write the output",
|
||||
default=sys.stdout,
|
||||
)
|
||||
|
||||
if yaml:
|
||||
help_output_format = (
|
||||
"The output format, can be 'json' or 'yaml' (default is json)"
|
||||
)
|
||||
else:
|
||||
help_output_format = "The output format, only 'json' is available install pyyaml to have yaml output format"
|
||||
|
||||
parser.add_argument(
|
||||
"-f",
|
||||
"--format",
|
||||
metavar="FORMAT",
|
||||
dest="formatter",
|
||||
type=format_type,
|
||||
help=help_output_format,
|
||||
default=format_type("json"),
|
||||
)
|
||||
|
||||
parser.set_defaults(func=show_oas)
|
||||
|
||||
|
||||
def show_oas(args: argparse.Namespace):
|
||||
print(json.dumps(generate_oas(args.apps), sort_keys=True, indent=4))
|
||||
"""
|
||||
Display Open API Specification on the stdout.
|
||||
"""
|
||||
spec = args.base
|
||||
spec.update(generate_oas(args.apps))
|
||||
print(args.formatter(spec), file=args.output)
|
||||
|
||||
120
aiohttp_pydantic/oas/docstring_parser.py
Normal file
120
aiohttp_pydantic/oas/docstring_parser.py
Normal file
@ -0,0 +1,120 @@
|
||||
"""
|
||||
Utility to extract extra OAS description from docstring.
|
||||
"""
|
||||
|
||||
import re
|
||||
import textwrap
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class LinesIterator:
|
||||
def __init__(self, lines: str):
|
||||
self._lines = lines.splitlines()
|
||||
self._i = -1
|
||||
|
||||
def next_line(self) -> str:
|
||||
if self._i == len(self._lines) - 1:
|
||||
raise StopIteration from None
|
||||
self._i += 1
|
||||
return self._lines[self._i]
|
||||
|
||||
def rewind(self) -> str:
|
||||
if self._i == -1:
|
||||
raise StopIteration from None
|
||||
self._i -= 1
|
||||
return self._lines[self._i]
|
||||
|
||||
def __iter__(self):
|
||||
return self
|
||||
|
||||
def __next__(self):
|
||||
return self.next_line()
|
||||
|
||||
|
||||
def _i_extract_block(lines: LinesIterator):
|
||||
"""
|
||||
Iter the line within an indented block and dedent them.
|
||||
"""
|
||||
|
||||
# Go to the first not empty or not white space line.
|
||||
try:
|
||||
line = next(lines)
|
||||
except StopIteration:
|
||||
return # No block to extract.
|
||||
while line.strip() == "":
|
||||
try:
|
||||
line = next(lines)
|
||||
except StopIteration:
|
||||
return
|
||||
|
||||
# Get the size of the indentation.
|
||||
if (match := re.search("^ +", line)) is None:
|
||||
return # No block to extract.
|
||||
indent = match.group()
|
||||
yield line[len(indent) :]
|
||||
|
||||
# Yield lines until the indentation is the same or is greater than
|
||||
# the first block line.
|
||||
try:
|
||||
line = next(lines)
|
||||
except StopIteration:
|
||||
return
|
||||
while (is_empty := line.strip() == "") or line.startswith(indent):
|
||||
yield "" if is_empty else line[len(indent) :]
|
||||
try:
|
||||
line = next(lines)
|
||||
except StopIteration:
|
||||
return
|
||||
|
||||
lines.rewind()
|
||||
|
||||
|
||||
def _dedent_under_first_line(text: str) -> str:
|
||||
"""
|
||||
Apply textwrap.dedent ignoring the first line.
|
||||
"""
|
||||
lines = text.splitlines()
|
||||
other_lines = "\n".join(lines[1:])
|
||||
if other_lines:
|
||||
return f"{lines[0]}\n{textwrap.dedent(other_lines)}"
|
||||
return text
|
||||
|
||||
|
||||
def status_code(docstring: str) -> Dict[int, str]:
|
||||
"""
|
||||
Extract the "Status Code:" block of the docstring.
|
||||
"""
|
||||
iterator = LinesIterator(docstring)
|
||||
for line in iterator:
|
||||
if re.fullmatch("status\\s+codes?\\s*:", line, re.IGNORECASE):
|
||||
blocks = []
|
||||
lines = []
|
||||
for line_of_block in _i_extract_block(iterator):
|
||||
if re.search("^\\d{3}\\s*:", line_of_block):
|
||||
if lines:
|
||||
blocks.append("\n".join(lines))
|
||||
lines = []
|
||||
lines.append(line_of_block)
|
||||
if lines:
|
||||
blocks.append("\n".join(lines))
|
||||
|
||||
return {
|
||||
int(status.strip()): _dedent_under_first_line(desc.strip())
|
||||
for status, desc in (block.split(":", 1) for block in blocks)
|
||||
}
|
||||
return {}
|
||||
|
||||
|
||||
def operation(docstring: str) -> str:
|
||||
"""
|
||||
Extract all docstring except the "Status Code:" block.
|
||||
"""
|
||||
lines = LinesIterator(docstring)
|
||||
ret = []
|
||||
for line in lines:
|
||||
if re.fullmatch("status\\s+codes?\\s*:", line, re.IGNORECASE):
|
||||
for _ in _i_extract_block(lines):
|
||||
pass
|
||||
else:
|
||||
ret.append(line)
|
||||
return ("\n".join(ret)).strip()
|
||||
@ -133,6 +133,7 @@ class Parameters:
|
||||
class Response:
|
||||
def __init__(self, spec: dict):
|
||||
self._spec = spec
|
||||
self._spec.setdefault("description", "")
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
|
||||
@ -8,6 +8,7 @@ from aiohttp.web_app import Application
|
||||
from pydantic import BaseModel
|
||||
|
||||
from aiohttp_pydantic.oas.struct import OpenApiSpec3, OperationObject, PathItem
|
||||
from . import docstring_parser
|
||||
|
||||
from ..injectors import _parse_func_signature
|
||||
from ..utils import is_pydantic_base_model
|
||||
@ -18,7 +19,7 @@ from .typing import is_status_code_type
|
||||
def _handle_optional(type_):
|
||||
"""
|
||||
Returns the type wrapped in Optional or None.
|
||||
|
||||
>>> from typing import Optional
|
||||
>>> _handle_optional(int)
|
||||
>>> _handle_optional(Optional[str])
|
||||
<class 'str'>
|
||||
@ -36,14 +37,15 @@ class _OASResponseBuilder:
|
||||
generate the OAS operation response.
|
||||
"""
|
||||
|
||||
def __init__(self, oas: OpenApiSpec3, oas_operation):
|
||||
def __init__(self, oas: OpenApiSpec3, oas_operation, status_code_descriptions):
|
||||
self._oas_operation = oas_operation
|
||||
self._oas = oas
|
||||
self._status_code_descriptions = status_code_descriptions
|
||||
|
||||
def _handle_pydantic_base_model(self, obj):
|
||||
if is_pydantic_base_model(obj):
|
||||
response_schema = obj.schema(ref_template="#/components/schemas/{model}")
|
||||
if def_sub_schemas := response_schema.get("definitions", None):
|
||||
if def_sub_schemas := response_schema.pop("definitions", None):
|
||||
self._oas.components.schemas.update(def_sub_schemas)
|
||||
return response_schema
|
||||
return {}
|
||||
@ -64,10 +66,16 @@ class _OASResponseBuilder:
|
||||
"schema": self._handle_list(typing.get_args(obj)[0])
|
||||
}
|
||||
}
|
||||
desc = self._status_code_descriptions.get(int(status_code))
|
||||
if desc:
|
||||
self._oas_operation.responses[status_code].description = desc
|
||||
|
||||
elif is_status_code_type(obj):
|
||||
status_code = obj.__name__[1:]
|
||||
self._oas_operation.responses[status_code].content = {}
|
||||
desc = self._status_code_descriptions.get(int(status_code))
|
||||
if desc:
|
||||
self._oas_operation.responses[status_code].description = desc
|
||||
|
||||
def _handle_union(self, obj):
|
||||
if typing.get_origin(obj) is typing.Union:
|
||||
@ -90,13 +98,16 @@ def _add_http_method_to_oas(
|
||||
)
|
||||
description = getdoc(handler)
|
||||
if description:
|
||||
oas_operation.description = description
|
||||
oas_operation.description = docstring_parser.operation(description)
|
||||
status_code_descriptions = docstring_parser.status_code(description)
|
||||
else:
|
||||
status_code_descriptions = {}
|
||||
|
||||
if body_args:
|
||||
body_schema = next(iter(body_args.values())).schema(
|
||||
ref_template="#/components/schemas/{model}"
|
||||
)
|
||||
if def_sub_schemas := body_schema.get("definitions", None):
|
||||
if def_sub_schemas := body_schema.pop("definitions", None):
|
||||
oas.components.schemas.update(def_sub_schemas)
|
||||
|
||||
oas_operation.request_body.content = {
|
||||
@ -127,7 +138,9 @@ def _add_http_method_to_oas(
|
||||
|
||||
return_type = handler.__annotations__.get("return")
|
||||
if return_type is not None:
|
||||
_OASResponseBuilder(oas, oas_operation).build(return_type)
|
||||
_OASResponseBuilder(oas, oas_operation, status_code_descriptions).build(
|
||||
return_type
|
||||
)
|
||||
|
||||
|
||||
def generate_oas(apps: List[Application]) -> dict:
|
||||
|
||||
@ -9,8 +9,14 @@ from aiohttp.web_exceptions import HTTPMethodNotAllowed
|
||||
from aiohttp.web_response import StreamResponse
|
||||
from pydantic import ValidationError
|
||||
|
||||
from .injectors import (AbstractInjector, BodyGetter, HeadersGetter,
|
||||
MatchInfoGetter, QueryGetter, _parse_func_signature)
|
||||
from .injectors import (
|
||||
AbstractInjector,
|
||||
BodyGetter,
|
||||
HeadersGetter,
|
||||
MatchInfoGetter,
|
||||
QueryGetter,
|
||||
_parse_func_signature,
|
||||
)
|
||||
|
||||
|
||||
class PydanticView(AbstractView):
|
||||
|
||||
29
demo/view.py
29
demo/view.py
@ -10,25 +10,54 @@ from .model import Error, Pet
|
||||
|
||||
class PetCollectionView(PydanticView):
|
||||
async def get(self, age: Optional[int] = None) -> r200[List[Pet]]:
|
||||
"""
|
||||
List all pets
|
||||
|
||||
Status Codes:
|
||||
200: Successful operation
|
||||
"""
|
||||
pets = self.request.app["model"].list_pets()
|
||||
return web.json_response(
|
||||
[pet.dict() for pet in pets if age is None or age == pet.age]
|
||||
)
|
||||
|
||||
async def post(self, pet: Pet) -> r201[Pet]:
|
||||
"""
|
||||
Add a new pet to the store
|
||||
|
||||
Status Codes:
|
||||
201: Successful operation
|
||||
"""
|
||||
self.request.app["model"].add_pet(pet)
|
||||
return web.json_response(pet.dict())
|
||||
|
||||
|
||||
class PetItemView(PydanticView):
|
||||
async def get(self, id: int, /) -> Union[r200[Pet], r404[Error]]:
|
||||
"""
|
||||
Find a pet by ID
|
||||
|
||||
Status Codes:
|
||||
200: Successful operation
|
||||
404: Pet not found
|
||||
"""
|
||||
pet = self.request.app["model"].find_pet(id)
|
||||
return web.json_response(pet.dict())
|
||||
|
||||
async def put(self, id: int, /, pet: Pet) -> r200[Pet]:
|
||||
"""
|
||||
Update an existing object
|
||||
|
||||
Status Codes:
|
||||
200: Successful operation
|
||||
404: Pet not found
|
||||
"""
|
||||
self.request.app["model"].update_pet(id, pet)
|
||||
return web.json_response(pet.dict())
|
||||
|
||||
async def delete(self, id: int, /) -> r204:
|
||||
"""
|
||||
Deletes a pet
|
||||
"""
|
||||
self.request.app["model"].remove_pet(id)
|
||||
return web.Response(status=204)
|
||||
|
||||
@ -8,6 +8,7 @@ pyparsing==2.4.7
|
||||
pytest==6.1.2
|
||||
pytest-aiohttp==0.3.0
|
||||
pytest-cov==2.10.1
|
||||
pyyaml==5.3.1
|
||||
six==1.15.0
|
||||
toml==0.10.2
|
||||
typing-extensions==3.7.4.3
|
||||
|
||||
1
tests/test_oas/test_cmd/oas_base.json
Normal file
1
tests/test_oas/test_cmd/oas_base.json
Normal file
@ -0,0 +1 @@
|
||||
{"info": {"title": "MyApp", "version": "1.0.0"}}
|
||||
@ -1,10 +1,13 @@
|
||||
import argparse
|
||||
from textwrap import dedent
|
||||
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
|
||||
from aiohttp_pydantic.oas import cmd
|
||||
|
||||
PATH_TO_BASE_JSON_FILE = str(Path(__file__).parent / "oas_base.json")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cmd_line():
|
||||
@ -13,10 +16,11 @@ def cmd_line():
|
||||
return parser
|
||||
|
||||
|
||||
def test_show_oad_of_app(cmd_line, capfd):
|
||||
def test_show_oas_of_app(cmd_line):
|
||||
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample"])
|
||||
args.output = StringIO()
|
||||
args.func(args)
|
||||
captured = capfd.readouterr()
|
||||
|
||||
expected = dedent(
|
||||
"""
|
||||
{
|
||||
@ -57,13 +61,13 @@ def test_show_oad_of_app(cmd_line, capfd):
|
||||
"""
|
||||
)
|
||||
|
||||
assert captured.out.strip() == expected.strip()
|
||||
assert args.output.getvalue().strip() == expected.strip()
|
||||
|
||||
|
||||
def test_show_oad_of_sub_app(cmd_line, capfd):
|
||||
def test_show_oas_of_sub_app(cmd_line):
|
||||
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample:sub_app"])
|
||||
args.output = StringIO()
|
||||
args.func(args)
|
||||
captured = capfd.readouterr()
|
||||
expected = dedent(
|
||||
"""
|
||||
{
|
||||
@ -89,16 +93,26 @@ def test_show_oad_of_sub_app(cmd_line, capfd):
|
||||
"""
|
||||
)
|
||||
|
||||
assert captured.out.strip() == expected.strip()
|
||||
assert args.output.getvalue().strip() == expected.strip()
|
||||
|
||||
|
||||
def test_show_oad_of_a_callable(cmd_line, capfd):
|
||||
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample:make_app()"])
|
||||
def test_show_oas_of_a_callable(cmd_line):
|
||||
args = cmd_line.parse_args(
|
||||
[
|
||||
"tests.test_oas.test_cmd.sample:make_app()",
|
||||
"--base-oas-file",
|
||||
PATH_TO_BASE_JSON_FILE,
|
||||
]
|
||||
)
|
||||
args.output = StringIO()
|
||||
args.func(args)
|
||||
captured = capfd.readouterr()
|
||||
expected = dedent(
|
||||
"""
|
||||
{
|
||||
"info": {
|
||||
"title": "MyApp",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"openapi": "3.0.0",
|
||||
"paths": {
|
||||
"/route-3/{a}": {
|
||||
@ -121,4 +135,4 @@ def test_show_oad_of_a_callable(cmd_line, capfd):
|
||||
"""
|
||||
)
|
||||
|
||||
assert captured.out.strip() == expected.strip()
|
||||
assert args.output.getvalue().strip() == expected.strip()
|
||||
|
||||
88
tests/test_oas/test_docstring_parser.py
Normal file
88
tests/test_oas/test_docstring_parser.py
Normal file
@ -0,0 +1,88 @@
|
||||
from aiohttp_pydantic.oas.docstring_parser import (
|
||||
status_code,
|
||||
operation,
|
||||
_i_extract_block,
|
||||
LinesIterator,
|
||||
)
|
||||
from inspect import getdoc
|
||||
import pytest
|
||||
|
||||
|
||||
def web_handler():
|
||||
"""
|
||||
bla bla bla
|
||||
|
||||
|
||||
Status Codes:
|
||||
200: line 1
|
||||
|
||||
line 2:
|
||||
- line 3
|
||||
- line 4
|
||||
|
||||
line 5
|
||||
|
||||
300: line A 1
|
||||
|
||||
301: line B 1
|
||||
line B 2
|
||||
400: line C 1
|
||||
|
||||
line C 2
|
||||
|
||||
line C 3
|
||||
|
||||
bla bla
|
||||
"""
|
||||
|
||||
|
||||
def test_lines_iterator():
|
||||
lines_iterator = LinesIterator("AAAA\nBBBB")
|
||||
with pytest.raises(StopIteration):
|
||||
lines_iterator.rewind()
|
||||
|
||||
assert lines_iterator.next_line() == "AAAA"
|
||||
assert lines_iterator.rewind()
|
||||
assert lines_iterator.next_line() == "AAAA"
|
||||
assert lines_iterator.next_line() == "BBBB"
|
||||
with pytest.raises(StopIteration):
|
||||
lines_iterator.next_line()
|
||||
|
||||
|
||||
def test_status_code():
|
||||
|
||||
expected = {
|
||||
200: "line 1\n\nline 2:\n - line 3\n - line 4\n\nline 5",
|
||||
300: "line A 1",
|
||||
301: "line B 1\nline B 2",
|
||||
400: "line C 1\n\nline C 2\n\n line C 3",
|
||||
}
|
||||
|
||||
assert status_code(getdoc(web_handler)) == expected
|
||||
|
||||
|
||||
def test_operation():
|
||||
expected = "bla bla bla\n\n\nbla bla"
|
||||
assert operation(getdoc(web_handler)) == expected
|
||||
|
||||
|
||||
def test_i_extract_block():
|
||||
lines = LinesIterator(" aaaa\n\n bbbb\n\n cccc\n dddd")
|
||||
text = "\n".join(_i_extract_block(lines))
|
||||
assert text == """aaaa\n\n bbbb\n\ncccc"""
|
||||
|
||||
lines = LinesIterator("")
|
||||
text = "\n".join(_i_extract_block(lines))
|
||||
assert text == ""
|
||||
|
||||
lines = LinesIterator("aaaa\n bbbb") # the indented block is cut by a new block
|
||||
text = "\n".join(_i_extract_block(lines))
|
||||
assert text == ""
|
||||
|
||||
lines = LinesIterator(" \n ")
|
||||
text = "\n".join(_i_extract_block(lines))
|
||||
assert text == ""
|
||||
|
||||
lines = LinesIterator(" aaaa\n bbbb")
|
||||
text = "\n".join(_i_extract_block(lines))
|
||||
assert text == "aaaa\nbbbb"
|
||||
@ -33,6 +33,9 @@ class PetCollectionView(PydanticView):
|
||||
) -> r200[List[Pet]]:
|
||||
"""
|
||||
Get a list of pets
|
||||
|
||||
Status Codes:
|
||||
200: Successful operation
|
||||
"""
|
||||
return web.json_response()
|
||||
|
||||
@ -49,6 +52,19 @@ class PetItemView(PydanticView):
|
||||
return web.json_response()
|
||||
|
||||
async def delete(self, id: int, /) -> r204:
|
||||
"""
|
||||
Status Code:
|
||||
204: Empty but OK
|
||||
"""
|
||||
return web.json_response()
|
||||
|
||||
|
||||
class TestResponseReturnASimpleType(PydanticView):
|
||||
async def get(self) -> r200[int]:
|
||||
"""
|
||||
Status Codes:
|
||||
200: The new number
|
||||
"""
|
||||
return web.json_response()
|
||||
|
||||
|
||||
@ -57,6 +73,7 @@ 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", TestResponseReturnASimpleType)
|
||||
oas.setup(app)
|
||||
|
||||
client = await aiohttp_client(app)
|
||||
@ -115,29 +132,11 @@ async def test_pets_route_should_have_get_method(generated_oas):
|
||||
],
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful operation",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"items": {
|
||||
"definitions": {
|
||||
"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",
|
||||
},
|
||||
},
|
||||
"properties": {
|
||||
"id": {"title": "Id", "type": "integer"},
|
||||
"name": {"title": "Name", "type": "string"},
|
||||
@ -154,7 +153,7 @@ async def test_pets_route_should_have_get_method(generated_oas):
|
||||
"type": "array",
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
@ -167,23 +166,6 @@ async def test_pets_route_should_have_post_method(generated_oas):
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"definitions": {
|
||||
"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",
|
||||
},
|
||||
},
|
||||
"properties": {
|
||||
"id": {"title": "Id", "type": "integer"},
|
||||
"name": {"title": "Name", "type": "string"},
|
||||
@ -202,26 +184,10 @@ async def test_pets_route_should_have_post_method(generated_oas):
|
||||
},
|
||||
"responses": {
|
||||
"201": {
|
||||
"description": "",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"definitions": {
|
||||
"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",
|
||||
},
|
||||
},
|
||||
"properties": {
|
||||
"id": {"title": "Id", "type": "integer"},
|
||||
"name": {"title": "Name", "type": "string"},
|
||||
@ -236,7 +202,7 @@ async def test_pets_route_should_have_post_method(generated_oas):
|
||||
"type": "object",
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
@ -248,15 +214,16 @@ async def test_generated_oas_should_have_pets_id_paths(generated_oas):
|
||||
|
||||
async def test_pets_id_route_should_have_delete_method(generated_oas):
|
||||
assert generated_oas["paths"]["/pets/{id}"]["delete"] == {
|
||||
"description": "",
|
||||
"parameters": [
|
||||
{
|
||||
"required": True,
|
||||
"in": "path",
|
||||
"name": "id",
|
||||
"required": True,
|
||||
"schema": {"title": "id", "type": "integer"},
|
||||
}
|
||||
],
|
||||
"responses": {"204": {"content": {}}},
|
||||
"responses": {"204": {"content": {}, "description": "Empty but OK"}},
|
||||
}
|
||||
|
||||
|
||||
@ -272,26 +239,10 @@ async def test_pets_id_route_should_have_get_method(generated_oas):
|
||||
],
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"definitions": {
|
||||
"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",
|
||||
},
|
||||
},
|
||||
"properties": {
|
||||
"id": {"title": "Id", "type": "integer"},
|
||||
"name": {"title": "Name", "type": "string"},
|
||||
@ -306,9 +257,9 @@ async def test_pets_id_route_should_have_get_method(generated_oas):
|
||||
"type": "object",
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
"404": {"content": {}},
|
||||
"404": {"description": "", "content": {}},
|
||||
},
|
||||
}
|
||||
|
||||
@ -327,23 +278,6 @@ async def test_pets_id_route_should_have_put_method(generated_oas):
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"definitions": {
|
||||
"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",
|
||||
},
|
||||
},
|
||||
"properties": {
|
||||
"id": {"title": "Id", "type": "integer"},
|
||||
"name": {"title": "Name", "type": "string"},
|
||||
@ -361,3 +295,15 @@ async def test_pets_id_route_should_have_put_method(generated_oas):
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
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",
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user