mirror of
https://github.com/correl/tornado-openapi3.git
synced 2024-11-23 11:09:56 +00:00
144 lines
4.2 KiB
Python
144 lines
4.2 KiB
Python
import json
|
|
import typing
|
|
|
|
from openapi_core.templating.responses.exceptions import (
|
|
ResponseNotFound,
|
|
)
|
|
import tornado.web
|
|
from tornado_openapi3.handler import OpenAPIRequestHandler
|
|
from tornado_openapi3.testing import AsyncOpenAPITestCase
|
|
from tornado_openapi3.types import Deserializer
|
|
|
|
|
|
def spec(responses: dict = dict()) -> dict:
|
|
if not responses:
|
|
responses = {"200": {"description": "Success"}}
|
|
return {
|
|
"openapi": "3.0.0",
|
|
"info": {
|
|
"title": "Test API",
|
|
"version": "1.0.0",
|
|
},
|
|
"components": {
|
|
"schemas": {
|
|
"resource": {
|
|
"type": "object",
|
|
"properties": {"name": {"type": "string"}},
|
|
"required": ["name"],
|
|
},
|
|
},
|
|
},
|
|
"paths": {
|
|
"/resource": {
|
|
"get": {
|
|
"responses": responses,
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
|
|
class TestTestCase(AsyncOpenAPITestCase):
|
|
def setUp(self) -> None: ...
|
|
|
|
def tearDown(self) -> None: ...
|
|
|
|
def test_schema_must_be_implemented(self) -> None:
|
|
with self.assertRaises(NotImplementedError):
|
|
self.spec
|
|
|
|
def test_no_custom_media_type_deserializers(self) -> None:
|
|
self.assertEqual(dict(), self.custom_media_type_deserializers)
|
|
|
|
|
|
class BaseTestCase(AsyncOpenAPITestCase):
|
|
@property
|
|
def spec_dict(self) -> dict:
|
|
return spec()
|
|
|
|
@property
|
|
def custom_media_type_deserializers(self) -> typing.Dict[str, Deserializer]:
|
|
return {
|
|
"application/vnd.example.resource+json": json.loads,
|
|
}
|
|
|
|
def get_app(self) -> tornado.web.Application:
|
|
testcase = self
|
|
|
|
class ResourceHandler(OpenAPIRequestHandler):
|
|
@property
|
|
def spec_dict(self) -> dict:
|
|
return spec()
|
|
|
|
@property
|
|
def custom_media_type_deserializers(self) -> typing.Dict[str, Deserializer]:
|
|
return {
|
|
"application/vnd.example.resource+json": json.loads,
|
|
}
|
|
|
|
async def get(self) -> None:
|
|
await testcase.get(self)
|
|
|
|
return tornado.web.Application([(r"/resource", ResourceHandler)])
|
|
|
|
async def get(self, handler: tornado.web.RequestHandler) -> None: ...
|
|
|
|
|
|
class SuccessTests(BaseTestCase):
|
|
spec_dict = spec(
|
|
responses={
|
|
"200": {
|
|
"description": "Success",
|
|
"content": {
|
|
"application/vnd.example.resource+json": {
|
|
"schema": {"$ref": "#/components/schemas/resource"}
|
|
}
|
|
},
|
|
}
|
|
}
|
|
)
|
|
|
|
async def get(self, handler: tornado.web.RequestHandler) -> None:
|
|
handler.set_header("Content-Type", "application/vnd.example.resource+json")
|
|
handler.finish(json.dumps({"name": "Name"}))
|
|
|
|
def test_success(self) -> None:
|
|
response = self.fetch("/resource")
|
|
self.assertEqual(200, response.code)
|
|
|
|
|
|
class IncorrectResponseTests(BaseTestCase):
|
|
@property
|
|
def spec_dict(self) -> dict:
|
|
return spec(responses={"200": {"description": "Success"}})
|
|
|
|
async def get(self, handler: tornado.web.RequestHandler) -> None:
|
|
handler.set_status(400)
|
|
|
|
def test_unexpected_response_code(self) -> None:
|
|
with self.assertRaises(ResponseNotFound) as context:
|
|
self.fetch("/resource")
|
|
self.assertEqual("400", context.exception.http_status)
|
|
|
|
def test_errors_not_raised_when_using_absolute_url(self) -> None:
|
|
self.fetch(self.get_url("/resource"))
|
|
|
|
|
|
class RaiseErrorTests(BaseTestCase):
|
|
@property
|
|
def spec_dict(self) -> dict:
|
|
return spec(
|
|
responses={
|
|
"500": {
|
|
"description": "An error has occurred.",
|
|
}
|
|
}
|
|
)
|
|
|
|
async def get(self, handler: tornado.web.RequestHandler) -> None:
|
|
handler.set_status(500)
|
|
|
|
def test_fetch_throws_error_on_expected_failure(self) -> None:
|
|
with self.assertRaises(tornado.httpclient.HTTPError) as context:
|
|
self.fetch("/resource", raise_error=True)
|
|
self.assertEqual(500, context.exception.code)
|