[doc-schema] the "/" route on a domain now returns the OpenAPI-validated Schema (not a list of schemas), the "dummy_domain" test now validates OpenAPI specs

This commit is contained in:
maxime 2023-08-01 20:31:17 +02:00
parent 20563081f5
commit 14e051bd91
8 changed files with 100 additions and 21 deletions

View File

@ -12,6 +12,8 @@ from schema import SchemaError
from starlette.applications import Starlette from starlette.applications import Starlette
from starlette.routing import Router, Route from starlette.routing import Router, Route
from starlette.schemas import SchemaGenerator
from .lib.responses import ORJSONResponse
from .lib.acl import AclRoute from .lib.acl import AclRoute
@ -42,15 +44,15 @@ class HalfDomain(Starlette):
self.app = app self.app = app
self.m_domain = importlib.import_module(domain) if module is None else module self.m_domain = importlib.import_module(domain) if module is None else module
d_domain = getattr(self.m_domain, 'domain', domain) self.d_domain = getattr(self.m_domain, 'domain', domain)
self.name = d_domain['name'] self.name = self.d_domain['name']
self.id = d_domain['id'] self.id = self.d_domain['id']
self.version = d_domain['version'] self.version = self.d_domain['version']
self.halfapi_version = d_domain.get('halfapi_version', __version__) self.halfapi_version = self.d_domain.get('halfapi_version', __version__)
self.deps = d_domain.get('deps', tuple()) self.deps = self.d_domain.get('deps', tuple())
if not router: if not router:
self.router = d_domain.get('routers', '.routers') self.router = self.d_domain.get('routers', '.routers')
else: else:
self.router = router self.router = router
@ -405,7 +407,7 @@ class HalfDomain(Starlette):
Generator(HalfRoute) Generator(HalfRoute)
""" """
yield HalfRoute('/', yield HalfRoute('/',
JSONRoute([ self.schema() ]), self.schema_openapi(),
[{'acl': lib_acl.public}], [{'acl': lib_acl.public}],
'GET' 'GET'
) )
@ -459,3 +461,35 @@ class HalfDomain(Starlette):
} }
schema['paths'] = self.schema_dict() schema['paths'] = self.schema_dict()
return schema return schema
def schema_openapi(self) -> Route:
schema = SchemaGenerator(
{
'openapi': '3.0.0',
'info': {
'title': self.name,
'version': self.version,
'x-acls': tuple(getattr(self.m_acl, 'ACLS', ())),
**({
f'x-{key}': value
for key, value in self.d_domain.items()
}),
}
}
)
async def inner(request, *args, **kwargs):
"""
description: |
Returns the current API routes description (OpenAPI v3)
as a JSON object
responses:
200:
description: API Schema in OpenAPI v3 format
"""
return ORJSONResponse(
schema.get_schema(routes=request.app.routes))
return inner

View File

@ -122,14 +122,11 @@ class TestDomain(TestCase):
def check_routes(self): def check_routes(self):
r = self.client.request('get', '/') r = self.client.request('get', '/')
assert r.status_code == 200 assert r.status_code == 200
schemas = r.json() schema = r.json()
assert isinstance(schemas, list)
for schema in schemas:
assert isinstance(schema, dict) assert isinstance(schema, dict)
assert 'openapi' in schema assert 'openapi' in schema
assert 'info' in schema assert 'info' in schema
assert 'paths' in schema assert 'paths' in schema
assert 'domain' in schema
r = self.client.request('get', '/halfapi/acls') r = self.client.request('get', '/halfapi/acls')
assert r.status_code == 200 assert r.status_code == 200

View File

@ -12,6 +12,9 @@ async def get(test):
""" """
description: description:
returns the path parameter returns the path parameter
responses:
200:
description: test response
""" """
return ORJSONResponse(str(test)) return ORJSONResponse(str(test))
@ -19,6 +22,9 @@ def post(test):
""" """
description: description:
returns the path parameter returns the path parameter
responses:
200:
description: test response
""" """
return str(test) return str(test)
@ -26,6 +32,9 @@ def patch(test):
""" """
description: description:
returns the path parameter returns the path parameter
responses:
200:
description: test response
""" """
return str(test) return str(test)
@ -33,6 +42,9 @@ def put(test):
""" """
description: description:
returns the path parameter returns the path parameter
responses:
200:
description: test response
""" """
return str(test) return str(test)
@ -40,5 +52,8 @@ def delete(test):
""" """
description: description:
returns the path parameter returns the path parameter
responses:
200:
description: test response
""" """
return str(test) return str(test)

View File

@ -6,5 +6,8 @@ def get():
""" """
description: description:
Not implemented Not implemented
responses:
200:
description: test response
""" """
raise NotImplementedError raise NotImplementedError

View File

@ -60,6 +60,9 @@ def get(data):
""" """
description: description:
returns the arguments passed in returns the arguments passed in
responses:
200:
description: test response
""" """
logger.error('%s', data['foo']) logger.error('%s', data['foo'])
return data return data
@ -68,6 +71,9 @@ def post(data):
""" """
description: description:
returns the arguments passed in returns the arguments passed in
responses:
200:
description: test response
""" """
logger.error('%s', data) logger.error('%s', data)
return data return data

View File

@ -25,24 +25,43 @@ ROUTES = {
async def get_abc_alphabet_TEST(request, *args, **kwargs): async def get_abc_alphabet_TEST(request, *args, **kwargs):
""" """
description: Not implemented description: Not implemented
responses:
200:
description: test response
parameters:
- name: test
in: path
description: Test parameter in route with "ROUTES" constant
required: true
schema:
type: string
""" """
return NotImplementedResponse() return NotImplementedResponse()
async def get_abc_pinnochio(request, *args, **kwargs): async def get_abc_pinnochio(request, *args, **kwargs):
""" """
description: Not implemented description: Not implemented
responses:
200:
description: test response
""" """
return NotImplementedResponse() return NotImplementedResponse()
async def get_config(request, *args, **kwargs): async def get_config(request, *args, **kwargs):
""" """
description: Not implemented description: Not implemented
responses:
200:
description: test response
""" """
return NotImplementedResponse() return NotImplementedResponse()
async def get_arguments(request, *args, **kwargs): async def get_arguments(request, *args, **kwargs):
""" """
description: Liste des datatypes. description: Liste des datatypes.
responses:
200:
description: test response
""" """
return ORJSONResponse({ return ORJSONResponse({
'foo': kwargs.get('data').get('foo'), 'foo': kwargs.get('data').get('foo'),

View File

@ -12,6 +12,9 @@ def get(halfapi):
""" """
description: description:
returns the configuration of the domain returns the configuration of the domain
responses:
200:
description: test response
""" """
logger.error('%s', halfapi) logger.error('%s', halfapi)
# TODO: Remove in 0.7.0 # TODO: Remove in 0.7.0

View File

@ -7,6 +7,10 @@ import json
import os import os
import sys import sys
import pprint import pprint
import openapi_spec_validator
import logging
logger = logging.getLogger()
from halfapi.lib.constants import API_SCHEMA from halfapi.lib.constants import API_SCHEMA
@ -58,8 +62,6 @@ def test_schema(application_debug):
c = TestClient(application_debug) c = TestClient(application_debug)
r = c.request('get', '/') r = c.request('get', '/')
schemas = r.json() schema = r.json()
assert isinstance(schemas, list)
for schema in schemas:
assert isinstance(schema, dict) assert isinstance(schema, dict)
assert API_SCHEMA.validate(schema) openapi_spec_validator.validate_spec(schema)