mirror of
https://github.com/correl/tornado-openapi3.git
synced 2024-11-22 03:00:15 +00:00
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
from dataclasses import dataclass
|
|
from typing import Any, Callable, Dict, Optional, TypeVar
|
|
import unittest
|
|
|
|
import attr
|
|
|
|
from hypothesis import given, settings
|
|
import hypothesis.strategies as s
|
|
|
|
from openapi_core import create_spec # type: ignore
|
|
from openapi_core.validation.response.datatypes import OpenAPIResponse # type: ignore
|
|
from tornado.httpclient import HTTPRequest, HTTPResponse # type: ignore
|
|
from tornado.testing import AsyncHTTPTestCase # type: ignore
|
|
from tornado.web import Application, RequestHandler # type: ignore
|
|
|
|
from tornado_openapi3 import (
|
|
ResponseValidator,
|
|
TornadoResponseFactory,
|
|
)
|
|
|
|
T = TypeVar("T")
|
|
settings(deadline=None)
|
|
|
|
|
|
@dataclass
|
|
class Responses:
|
|
code: int
|
|
headers: Dict[str, str]
|
|
|
|
def as_openapi(self) -> Dict[str, Any]:
|
|
return {
|
|
str(self.code): {
|
|
"description": "Response",
|
|
"headers": {
|
|
name: {"schema": {"type": "string", "enum": [value]}}
|
|
for name, value in self.headers.items()
|
|
},
|
|
}
|
|
}
|
|
|
|
|
|
@s.composite
|
|
def responses(draw: Callable[[Any], Any], min_headers: int = 0) -> Responses:
|
|
field_name = s.text(
|
|
s.characters(
|
|
min_codepoint=33,
|
|
max_codepoint=126,
|
|
blacklist_categories=("Lu",),
|
|
blacklist_characters=":",
|
|
),
|
|
min_size=1,
|
|
)
|
|
field_value = s.text(
|
|
s.characters(
|
|
min_codepoint=0x20, max_codepoint=0x7E, blacklist_characters=" \r\n"
|
|
),
|
|
min_size=1,
|
|
)
|
|
code = s.sampled_from([200, 304, 400, 500])
|
|
headers = s.dictionaries(field_name, field_value, min_size=min_headers)
|
|
return Responses(
|
|
code=draw(code),
|
|
headers=draw(headers),
|
|
)
|
|
|
|
|
|
class TestResponseFactory(unittest.TestCase):
|
|
def test_response(self) -> None:
|
|
tornado_request = HTTPRequest(url="http://example.com")
|
|
tornado_response = HTTPResponse(request=tornado_request, code=200)
|
|
expected = OpenAPIResponse(
|
|
data="",
|
|
status_code=200,
|
|
mimetype="text/html",
|
|
)
|
|
openapi_response = TornadoResponseFactory.create(tornado_response)
|
|
self.assertEqual(attr.asdict(expected), attr.asdict(openapi_response))
|
|
|
|
|
|
class ResponsesHandler(RequestHandler):
|
|
responses: Optional[Responses] = None
|
|
|
|
def get(self) -> None:
|
|
if ResponsesHandler.responses:
|
|
self.set_status(ResponsesHandler.responses.code)
|
|
for name, value in ResponsesHandler.responses.headers.items():
|
|
self.add_header(name, value)
|
|
|
|
|
|
class TestResponse(AsyncHTTPTestCase):
|
|
def get_app(self) -> Application:
|
|
return Application([(r"/.*", ResponsesHandler)])
|
|
|
|
@given(responses())
|
|
def test_simple_request(self, responses: Responses) -> None:
|
|
spec = create_spec(
|
|
{
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Test specification", "version": "0.1"},
|
|
"paths": {
|
|
"/": {
|
|
"get": {
|
|
"responses": responses.as_openapi(),
|
|
}
|
|
}
|
|
},
|
|
}
|
|
)
|
|
ResponsesHandler.responses = responses
|
|
validator = ResponseValidator(spec)
|
|
response = self.fetch("/")
|
|
result = validator.validate(response)
|
|
result.raise_for_errors()
|