github/openapi-core
1
0
Fork 0
mirror of https://github.com/correl/openapi-core.git synced 2024-12-28 03:00:11 +00:00
Code Issues Projects Releases Packages Wiki Activity

Path pattern finder

Browse source
This commit is contained in:
Artur Maciag 2020-02-21 16:33:45 +00:00
parent 817ff5c746
commit dcb7161af7
20 changed files with 753 additions and 127 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
openapi_core/contrib/flask/handlers.py
View file

@ -3,13 +3,17 @@ from flask.globals import current_app
from flask.json import dumps
from openapi_core.schema.media_types.exceptions import InvalidContentType
from openapi_core.schema.servers.exceptions import InvalidServer
from openapi_core.templating.paths.exceptions import (
ServerNotFound, OperationNotFound, PathNotFound,
)
class FlaskOpenAPIErrorsHandler(object):
OPENAPI_ERROR_STATUS = {
InvalidServer: 500,
ServerNotFound: 400,
OperationNotFound: 405,
PathNotFound: 404,
InvalidContentType: 415,
}

10
openapi_core/schema/servers/models.py
View file

@ -1,5 +1,6 @@
"""OpenAPI core servers models module"""
from six import iteritems
from six.moves.urllib.parse import urljoin
class Server(object):
@ -25,6 +26,15 @@ class Server(object):
variables = self.default_variables
return self.url.format(**variables)
@staticmethod
def is_absolute(url):
return url.startswith('//') or '://' in url
def get_absolute_url(self, base_url=None):
if base_url is not None and not self.is_absolute(self.url):
return urljoin(base_url, self.url)
return self.url
class ServerVariable(object):

13
openapi_core/templating/datatypes.py Normal file
View file

@ -0,0 +1,13 @@
import attr
@attr.s
class TemplateResult(object):
pattern = attr.ib(default=None)
variables = attr.ib(default=None)
@property
def resolved(self):
if not self.variables:
return self.pattern
return self.pattern.format(**self.variables)

36
openapi_core/templating/paths/exceptions.py Normal file
View file

@ -0,0 +1,36 @@
import attr
from openapi_core.exceptions import OpenAPIError
class PathError(OpenAPIError):
"""Path error"""
@attr.s(hash=True)
class PathNotFound(PathError):
"""Find path error"""
url = attr.ib()
def __str__(self):
return "Path not found for {0}".format(self.url)
@attr.s(hash=True)
class OperationNotFound(PathError):
"""Find path operation error"""
url = attr.ib()
method = attr.ib()
def __str__(self):
return "Operation {0} not found for {1}".format(
self.method, self.url)
@attr.s(hash=True)
class ServerNotFound(PathError):
"""Find server error"""
url = attr.ib()
def __str__(self):
return "Server not found for {0}".format(self.url)

88
openapi_core/templating/paths/finders.py
View file

@ -1,27 +1,85 @@
"""OpenAPI core templating paths finders module"""
from openapi_core.templating.paths.util import get_operation_pattern
from more_itertools import peekable
from six import iteritems
from openapi_core.templating.datatypes import TemplateResult
from openapi_core.templating.util import parse, search
from openapi_core.templating.paths.exceptions import (
PathNotFound, OperationNotFound, ServerNotFound,
)
class PathFinder(object):
def __init__(self, spec):
def __init__(self, spec, base_url=None):
self.spec = spec
self.base_url = base_url
def find(self, request):
operation_pattern = self._get_operation_pattern(request)
paths_iter = self._get_paths_iter(request.full_url_pattern)
paths_iter_peek = peekable(paths_iter)
path = self.spec[operation_pattern]
path_variables = {}
operation = self.spec.get_operation(operation_pattern, request.method)
servers = path.servers or operation.servers or self.spec.servers
server = servers[0]
server_variables = {}
if not paths_iter_peek:
raise PathNotFound(request.full_url_pattern)
return path, operation, server, path_variables, server_variables
operations_iter = self._get_operations_iter(
request.method, paths_iter_peek)
operations_iter_peek = peekable(operations_iter)
def _get_operation_pattern(self, request):
server = self.spec.get_server(request.full_url_pattern)
if not operations_iter_peek:
raise OperationNotFound(request.full_url_pattern, request.method)
return get_operation_pattern(
server.default_url, request.full_url_pattern
)
servers_iter = self._get_servers_iter(
request.full_url_pattern, operations_iter_peek)
try:
return next(servers_iter)
except StopIteration:
raise ServerNotFound(request.full_url_pattern)
def _get_paths_iter(self, full_url_pattern):
for path_pattern, path in iteritems(self.spec.paths):
# simple path
if full_url_pattern.endswith(path_pattern):
path_result = TemplateResult(path_pattern, {})
yield (path, path_result)
# template path
else:
result = search(path_pattern, full_url_pattern)
if result:
path_result = TemplateResult(path_pattern, result.named)
yield (path, path_result)
def _get_operations_iter(self, request_method, paths_iter):
for path, path_result in paths_iter:
if request_method not in path.operations:
continue
operation = path.operations[request_method]
yield (path, operation, path_result)
def _get_servers_iter(self, full_url_pattern, ooperations_iter):
for path, operation, path_result in ooperations_iter:
servers = path.servers or operation.servers or self.spec.servers
for server in servers:
server_url_pattern = full_url_pattern.rsplit(
path_result.resolved, 1)[0]
server_url = server.get_absolute_url(self.base_url)
if server_url.endswith('/'):
server_url = server_url[:-1]
# simple path
if server_url_pattern.startswith(server_url):
server_result = TemplateResult(server.url, {})
yield (
path, operation, server,
path_result, server_result,
)
# template path
else:
result = parse(server.url, server_url_pattern)
if result:
server_result = TemplateResult(
server.url, result.named)
yield (
path, operation, server,
path_result, server_result,
)

24
openapi_core/templating/paths/util.py
View file

@ -1,24 +0,0 @@
"""OpenAPI core templating paths util module"""
from six.moves.urllib.parse import urlparse
def is_absolute(url):
return url.startswith('//') or '://' in url
def path_qs(url):
pr = urlparse(url)
result = pr.path
if pr.query:
result += '?' + pr.query
return result
def get_operation_pattern(server_url, request_url_pattern):
"""Return an updated request URL pattern with the server URL removed."""
if server_url[-1] == "/":
# operations have to start with a slash, so do not remove it
server_url = server_url[:-1]
if is_absolute(server_url):
return request_url_pattern.replace(server_url, "", 1)
return path_qs(request_url_pattern).replace(server_url, "", 1)

13
openapi_core/templating/util.py Normal file
View file

@ -0,0 +1,13 @@
from parse import Parser
def search(path_pattern, full_url_pattern):
p = Parser(path_pattern)
p._expression = p._expression + '$'
return p.search(full_url_pattern)
def parse(server_url, server_url_pattern):
p = Parser(server_url)
p._expression = '^' + p._expression
return p.parse(server_url_pattern)

10
openapi_core/validation/request/validators.py
View file

@ -5,14 +5,12 @@ from six import iteritems
from openapi_core.casting.schemas.exceptions import CastError
from openapi_core.deserializing.exceptions import DeserializeError
from openapi_core.schema.media_types.exceptions import InvalidContentType
from openapi_core.schema.operations.exceptions import InvalidOperation
from openapi_core.schema.parameters.exceptions import (
MissingRequiredParameter, MissingParameter,
)
from openapi_core.schema.paths.exceptions import InvalidPath
from openapi_core.schema.request_bodies.exceptions import MissingRequestBody
from openapi_core.schema.servers.exceptions import InvalidServer
from openapi_core.security.exceptions import SecurityError
from openapi_core.templating.paths.exceptions import PathError
from openapi_core.unmarshalling.schemas.enums import UnmarshalContext
from openapi_core.unmarshalling.schemas.exceptions import (
UnmarshalError, ValidateError,
@ -30,7 +28,7 @@ class RequestValidator(BaseValidator):
try:
path, operation, _, _, _ = self._find_path(request)
# don't process if operation errors
except (InvalidServer, InvalidPath, InvalidOperation) as exc:
except PathError as exc:
return RequestValidationResult([exc, ], None, None, None)
try:
@ -53,7 +51,7 @@ class RequestValidator(BaseValidator):
def _validate_parameters(self, request):
try:
path, operation, _, _, _ = self._find_path(request)
except (InvalidServer, InvalidPath, InvalidOperation) as exc:
except PathError as exc:
return RequestValidationResult([exc, ], None, None)
params, params_errors = self._get_parameters(
@ -67,7 +65,7 @@ class RequestValidator(BaseValidator):
def _validate_body(self, request):
try:
_, operation, _, _, _ = self._find_path(request)
except (InvalidServer, InvalidOperation) as exc:
except PathError as exc:
return RequestValidationResult([exc, ], None, None)
body, body_errors = self._get_body(request, operation)

8
openapi_core/validation/response/validators.py
View file

@ -1,13 +1,11 @@
"""OpenAPI core validation response validators module"""
from openapi_core.casting.schemas.exceptions import CastError
from openapi_core.deserializing.exceptions import DeserializeError
from openapi_core.schema.operations.exceptions import InvalidOperation
from openapi_core.schema.media_types.exceptions import InvalidContentType
from openapi_core.schema.paths.exceptions import InvalidPath
from openapi_core.schema.responses.exceptions import (
InvalidResponse, MissingResponseContent,
)
from openapi_core.schema.servers.exceptions import InvalidServer
from openapi_core.templating.paths.exceptions import PathError
from openapi_core.unmarshalling.schemas.enums import UnmarshalContext
from openapi_core.unmarshalling.schemas.exceptions import (
UnmarshalError, ValidateError,
@ -22,7 +20,7 @@ class ResponseValidator(BaseValidator):
try:
_, operation, _, _, _ = self._find_path(request)
# don't process if operation errors
except (InvalidServer, InvalidPath, InvalidOperation) as exc:
except PathError as exc:
return ResponseValidationResult([exc, ], None, None)
try:
@ -47,7 +45,7 @@ class ResponseValidator(BaseValidator):
try:
_, operation, _, _, _ = self._find_path(request)
# don't process if operation errors
except (InvalidServer, InvalidPath, InvalidOperation) as exc:
except PathError as exc:
return ResponseValidationResult([exc, ], None, None)
try:

4
openapi_core/validation/validators.py
View file

@ -5,15 +5,17 @@ class BaseValidator(object):
def __init__(
self, spec,
base_url=None,
custom_formatters=None, custom_media_type_deserializers=None,
):
self.spec = spec
self.base_url = base_url
self.custom_formatters = custom_formatters
self.custom_media_type_deserializers = custom_media_type_deserializers
def _find_path(self, request):
from openapi_core.templating.paths.finders import PathFinder
finder = PathFinder(self.spec)
finder = PathFinder(self.spec, base_url=self.base_url)
return finder.find(request)
def _deserialise_media_type(self, media_type, value):

2
requirements.txt
View file

@ -4,3 +4,5 @@ lazy-object-proxy
strict_rfc3339
isodate
attrs
parse==1.14.0
more-itertools>=5.0.0

1
requirements_2.7.txt
View file

@ -6,3 +6,4 @@ backports.functools-partialmethod
enum34
strict_rfc3339
attrs
more-itertools==5.0.0

2
setup.cfg
View file

@ -30,6 +30,8 @@ install_requires =
isodate
attrs
werkzeug
parse
more-itertools
backports.functools-lru-cache; python_version<"3.0"
backports.functools-partialmethod; python_version<"3.0"
tests_require =

63
tests/integration/contrib/flask/test_flask_decorator.py
View file

@ -39,13 +39,21 @@ class TestFlaskOpenAPIDecorator(object):
return view_response
@pytest.fixture(autouse=True)
def view(self, app, decorator, view_response):
@app.route("/browse/<id>/")
def details_view(self, app, decorator, view_response):
@app.route("/browse/<id>/", methods=['GET', 'POST'])
@decorator
def browse_details(*args, **kwargs):
return view_response(*args, **kwargs)
return browse_details
@pytest.fixture(autouse=True)
def list_view(self, app, decorator, view_response):
@app.route("/browse/")
@decorator
def browse_list(*args, **kwargs):
return view_response(*args, **kwargs)
return browse_list
def test_invalid_content_type(self, client):
def view_response_callable(*args, **kwargs):
from flask.globals import request
@ -80,17 +88,60 @@ class TestFlaskOpenAPIDecorator(object):
'errors': [
{
'class': (
"<class 'openapi_core.schema.servers.exceptions."
"InvalidServer'>"
"<class 'openapi_core.templating.paths.exceptions."
"ServerNotFound'>"
),
'status': 500,
'status': 400,
'title': (
'Invalid request server '
'Server not found for '
'https://localhost/browse/{id}/'
),
}
]
}
assert result.status_code == 400
assert result.json == expected_data
def test_operation_error(self, client):
result = client.post('/browse/12/')
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.templating.paths.exceptions."
"OperationNotFound'>"
),
'status': 405,
'title': (
'Operation post not found for '
'http://localhost/browse/{id}/'
),
}
]
}
assert result.status_code == 405
assert result.json == expected_data
def test_path_error(self, client):
result = client.get('/browse/')
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.templating.paths.exceptions."
"PathNotFound'>"
),
'status': 404,
'title': (
'Path not found for '
'http://localhost/browse/'
),
}
]
}
assert result.status_code == 404
assert result.json == expected_data
def test_endpoint_error(self, client):

75
tests/integration/contrib/flask/test_flask_views.py
View file

@ -28,17 +28,30 @@ class TestFlaskOpenAPIView(object):
yield client
@pytest.fixture
def view_func(self, spec):
def details_view_func(self, spec):
outer = self
class MyView(FlaskOpenAPIView):
class MyDetailsView(FlaskOpenAPIView):
def get(self, id):
return outer.view_response
return MyView.as_view('browse_details', spec)
def post(self, id):
return outer.view_response
return MyDetailsView.as_view('browse_details', spec)
@pytest.fixture
def list_view_func(self, spec):
outer = self
class MyListView(FlaskOpenAPIView):
def get(self):
return outer.view_response
return MyListView.as_view('browse_list', spec)
@pytest.fixture(autouse=True)
def view(self, app, view_func):
app.add_url_rule("/browse/<id>/", view_func=view_func)
def view(self, app, details_view_func, list_view_func):
app.add_url_rule("/browse/<id>/", view_func=details_view_func)
app.add_url_rule("/browse/", view_func=list_view_func)
def test_invalid_content_type(self, client):
self.view_response = make_response('success', 200)
@ -68,18 +81,60 @@ class TestFlaskOpenAPIView(object):
'errors': [
{
'class': (
"<class 'openapi_core.schema.servers.exceptions."
"InvalidServer'>"
"<class 'openapi_core.templating.paths.exceptions."
"ServerNotFound'>"
),
'status': 500,
'status': 400,
'title': (
'Invalid request server '
'Server not found for '
'https://localhost/browse/{id}/'
),
}
]
}
assert result.status_code == 500
assert result.status_code == 400
assert result.json == expected_data
def test_operation_error(self, client):
result = client.post('/browse/12/')
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.templating.paths.exceptions."
"OperationNotFound'>"
),
'status': 405,
'title': (
'Operation post not found for '
'http://localhost/browse/{id}/'
),
}
]
}
assert result.status_code == 405
assert result.json == expected_data
def test_path_error(self, client):
result = client.get('/browse/')
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.templating.paths.exceptions."
"PathNotFound'>"
),
'status': 404,
'title': (
'Path not found for '
'http://localhost/browse/'
),
}
]
}
assert result.status_code == 404
assert result.json == expected_data
def test_endpoint_error(self, client):

9
tests/integration/validation/test_minimal.py
View file

@ -1,8 +1,9 @@
import pytest
from openapi_core.schema.operations.exceptions import InvalidOperation
from openapi_core.schema.paths.exceptions import InvalidPath
from openapi_core.shortcuts import create_spec
from openapi_core.templating.paths.exceptions import (
PathNotFound, OperationNotFound,
)
from openapi_core.testing import MockRequest
from openapi_core.validation.request.validators import RequestValidator
@ -45,7 +46,7 @@ class TestMinimal(object):
result = validator.validate(request)
assert len(result.errors) == 1
assert isinstance(result.errors[0], InvalidOperation)
assert isinstance(result.errors[0], OperationNotFound)
assert result.body is None
assert result.parameters is None
@ -60,6 +61,6 @@ class TestMinimal(object):
result = validator.validate(request)
assert len(result.errors) == 1
assert isinstance(result.errors[0], InvalidPath)
assert isinstance(result.errors[0], PathNotFound)
assert result.body is None
assert result.parameters is None

68
tests/integration/validation/test_petstore.py
View file

@ -16,9 +16,11 @@ from openapi_core.schema.parameters.exceptions import (
MissingRequiredParameter,
)
from openapi_core.schema.schemas.enums import SchemaType
from openapi_core.schema.servers.exceptions import InvalidServer
from openapi_core.shortcuts import (
create_spec, validate_parameters, validate_body,
create_spec, validate_parameters, validate_body, validate_data,
)
from openapi_core.templating.paths.exceptions import (
ServerNotFound,
)
from openapi_core.testing import MockRequest, MockResponse
from openapi_core.unmarshalling.schemas.exceptions import InvalidSchemaValue
@ -163,7 +165,7 @@ class TestPetstore(object):
)
assert body is None
data_json = {
response_data_json = {
'data': [
{
'id': 1,
@ -173,8 +175,11 @@ class TestPetstore(object):
}
],
}
data = json.dumps(data_json)
response = MockResponse(data)
response_data = json.dumps(response_data_json)
response = MockResponse(response_data)
with pytest.raises(InvalidSchemaValue):
validate_data(spec, request, response)
response_result = response_validator.validate(request, response)
@ -182,7 +187,7 @@ class TestPetstore(object):
assert response_result.errors == [
InvalidSchemaValue(
type=SchemaType.OBJECT,
value=data_json,
value=response_data_json,
schema_errors=schema_errors,
),
]
@ -363,7 +368,7 @@ class TestPetstore(object):
assert body is None
def test_post_birds(self, spec, spec_dict):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
pet_name = 'Cat'
pet_tag = 'cats'
@ -423,7 +428,7 @@ class TestPetstore(object):
assert body.healthy == pet_healthy
def test_post_cats(self, spec, spec_dict):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
pet_name = 'Cat'
pet_tag = 'cats'
@ -483,7 +488,7 @@ class TestPetstore(object):
assert body.healthy == pet_healthy
def test_post_cats_boolean_string(self, spec, spec_dict):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
pet_name = 'Cat'
pet_tag = 'cats'
@ -543,7 +548,7 @@ class TestPetstore(object):
assert body.healthy is False
def test_post_no_one_of_schema(self, spec, spec_dict):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
pet_name = 'Cat'
alias = 'kitty'
@ -580,7 +585,7 @@ class TestPetstore(object):
validate_body(spec, request)
def test_post_cats_only_required_body(self, spec, spec_dict):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
pet_name = 'Cat'
pet_healthy = True
@ -625,7 +630,7 @@ class TestPetstore(object):
assert not hasattr(body, 'address')
def test_post_pets_raises_invalid_mimetype(self, spec):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
data_json = {
'name': 'Cat',
@ -660,7 +665,7 @@ class TestPetstore(object):
validate_body(spec, request)
def test_post_pets_missing_cookie(self, spec, spec_dict):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
pet_name = 'Cat'
pet_healthy = True
@ -694,7 +699,7 @@ class TestPetstore(object):
assert not hasattr(body, 'address')
def test_post_pets_missing_header(self, spec, spec_dict):
host_url = 'http://petstore.swagger.io/v1'
host_url = 'https://staging.gigantic-server.com/v1'
path_pattern = '/v1/pets'
pet_name = 'Cat'
pet_healthy = True
@ -748,12 +753,29 @@ class TestPetstore(object):
headers=headers, cookies=cookies,
)
with pytest.raises(InvalidServer):
with pytest.raises(ServerNotFound):
validate_parameters(spec, request)
with pytest.raises(InvalidServer):
with pytest.raises(ServerNotFound):
validate_body(spec, request)
data_id = 1
data_name = 'test'
data_json = {
'data': {
'id': data_id,
'name': data_name,
'ears': {
'healthy': True,
},
},
}
data = json.dumps(data_json)
response = MockResponse(data)
with pytest.raises(ServerNotFound):
validate_data(spec, request, response)
def test_get_pet(self, spec, response_validator):
host_url = 'http://petstore.swagger.io/v1'
path_pattern = '/v1/pets/{petId}'
@ -1075,14 +1097,22 @@ class TestPetstore(object):
message = 'Bad request'
rootCause = 'Tag already exist'
additionalinfo = 'Tag Dog already exist'
data_json = {
response_data_json = {
'code': code,
'message': message,
'rootCause': rootCause,
'additionalinfo': additionalinfo,
}
data = json.dumps(data_json)
response = MockResponse(data, status_code=404)
response_data = json.dumps(response_data_json)
response = MockResponse(response_data, status_code=404)
data = validate_data(spec, request, response)
assert isinstance(data, BaseModel)
assert data.code == code
assert data.message == message
assert data.rootCause == rootCause
assert data.additionalinfo == additionalinfo
response_result = response_validator.validate(request, response)

36
tests/integration/validation/test_validators.py
View file

@ -9,15 +9,15 @@ from openapi_core.schema.media_types.exceptions import (
InvalidContentType,
)
from openapi_core.extensions.models.models import BaseModel
from openapi_core.schema.operations.exceptions import InvalidOperation
from openapi_core.schema.parameters.exceptions import MissingRequiredParameter
from openapi_core.schema.paths.exceptions import InvalidPath
from openapi_core.schema.request_bodies.exceptions import MissingRequestBody
from openapi_core.schema.responses.exceptions import (
MissingResponseContent, InvalidResponse,
)
from openapi_core.schema.servers.exceptions import InvalidServer
from openapi_core.shortcuts import create_spec
from openapi_core.templating.paths.exceptions import (
PathNotFound, OperationNotFound,
)
from openapi_core.testing import MockRequest, MockResponse
from openapi_core.unmarshalling.schemas.exceptions import InvalidSchemaValue
from openapi_core.validation.exceptions import InvalidSecurity
@ -48,7 +48,7 @@ class TestRequestValidator(object):
@pytest.fixture(scope='session')
def validator(self, spec):
return RequestValidator(spec)
return RequestValidator(spec, base_url=self.host_url)
def test_request_server_error(self, validator):
request = MockRequest('http://petstore.invalid.net/v1', 'get', '/')
@ -56,7 +56,7 @@ class TestRequestValidator(object):
result = validator.validate(request)
assert len(result.errors) == 1
assert type(result.errors[0]) == InvalidServer
assert type(result.errors[0]) == PathNotFound
assert result.body is None
assert result.parameters is None
@ -66,7 +66,7 @@ class TestRequestValidator(object):
result = validator.validate(request)
assert len(result.errors) == 1
assert type(result.errors[0]) == InvalidPath
assert type(result.errors[0]) == PathNotFound
assert result.body is None
assert result.parameters is None
@ -76,7 +76,7 @@ class TestRequestValidator(object):
result = validator.validate(request)
assert len(result.errors) == 1
assert type(result.errors[0]) == InvalidOperation
assert type(result.errors[0]) == OperationNotFound
assert result.body is None
assert result.parameters is None
@ -149,7 +149,7 @@ class TestRequestValidator(object):
'user': '123',
}
request = MockRequest(
self.host_url, 'post', '/v1/pets',
'https://development.gigantic-server.com', 'post', '/v1/pets',
path_pattern='/v1/pets',
headers=headers, cookies=cookies,
)
@ -176,7 +176,7 @@ class TestRequestValidator(object):
'user': '123',
}
request = MockRequest(
self.host_url, 'post', '/v1/pets',
'https://development.gigantic-server.com', 'post', '/v1/pets',
path_pattern='/v1/pets', mimetype='text/csv',
headers=headers, cookies=cookies,
)
@ -220,7 +220,7 @@ class TestRequestValidator(object):
'user': '123',
}
request = MockRequest(
self.host_url, 'post', '/v1/pets',
'https://development.gigantic-server.com', 'post', '/v1/pets',
path_pattern='/v1/pets', data=data,
headers=headers, cookies=cookies,
)
@ -326,7 +326,7 @@ class TestPathItemParamsValidator(object):
@pytest.fixture(scope='session')
def validator(self, spec):
return RequestValidator(spec)
return RequestValidator(spec, base_url='http://example.com')
def test_request_missing_param(self, validator):
request = MockRequest('http://example.com', 'get', '/resource')
@ -373,7 +373,8 @@ class TestPathItemParamsValidator(object):
},
}
]
validator = RequestValidator(create_spec(spec_dict))
validator = RequestValidator(
create_spec(spec_dict), base_url='http://example.com')
request = MockRequest('http://example.com', 'get', '/resource')
result = validator.validate(request)
@ -395,7 +396,8 @@ class TestPathItemParamsValidator(object):
},
}
]
validator = RequestValidator(create_spec(spec_dict))
validator = RequestValidator(
create_spec(spec_dict), base_url='http://example.com')
request = MockRequest('http://example.com', 'get', '/resource')
result = validator.validate(request)
@ -419,7 +421,7 @@ class TestResponseValidator(object):
@pytest.fixture
def validator(self, spec):
return ResponseValidator(spec)
return ResponseValidator(spec, base_url=self.host_url)
def test_invalid_server(self, validator):
request = MockRequest('http://petstore.invalid.net/v1', 'get', '/')
@ -428,18 +430,18 @@ class TestResponseValidator(object):
result = validator.validate(request, response)
assert len(result.errors) == 1
assert type(result.errors[0]) == InvalidServer
assert type(result.errors[0]) == PathNotFound
assert result.data is None
assert result.headers is None
def test_invalid_operation(self, validator):
request = MockRequest(self.host_url, 'get', '/v1')
request = MockRequest(self.host_url, 'patch', '/v1/pets')
response = MockResponse('Not Found', status_code=404)
result = validator.validate(request, response)
assert len(result.errors) == 1
assert type(result.errors[0]) == InvalidPath
assert type(result.errors[0]) == OperationNotFound
assert result.data is None
assert result.headers is None

392
tests/unit/templating/test_paths_finders.py Normal file
View file

@ -0,0 +1,392 @@
import pytest
from openapi_core.schema.infos.models import Info
from openapi_core.schema.operations.models import Operation
from openapi_core.schema.parameters.models import Parameter
from openapi_core.schema.paths.models import Path
from openapi_core.schema.servers.models import Server, ServerVariable
from openapi_core.schema.specs.models import Spec
from openapi_core.templating.datatypes import TemplateResult
from openapi_core.templating.paths.exceptions import (
PathNotFound, OperationNotFound, ServerNotFound,
)
from openapi_core.templating.paths.finders import PathFinder
from openapi_core.testing import MockRequest
class BaseTestSimpleServer(object):
server_url = 'http://petstore.swagger.io'
@pytest.fixture
def server(self):
return Server(self.server_url, {})
@pytest.fixture
def servers(self, server):
return [server, ]
class BaseTestVariableServer(BaseTestSimpleServer):
server_url = 'http://petstore.swagger.io/{version}'
server_variable_name = 'version'
server_variable_default = 'v1'
server_variable_enum = ['v1', 'v2']
@pytest.fixture
def server_variable(self):
return ServerVariable(
self.server_variable_name,
default=self.server_variable_default,
enum=self.server_variable_enum,
)
@pytest.fixture
def server_variables(self, server_variable):
return {
self.server_variable_name: server_variable,
}
@pytest.fixture
def server(self, server_variables):
return Server(self.server_url, server_variables)
class BaseTestSimplePath(object):
path_name = '/resource'
@pytest.fixture
def path(self, operations):
return Path(self.path_name, operations)
@pytest.fixture
def paths(self, path):
return {
self.path_name: path,
}
class BaseTestVariablePath(BaseTestSimplePath):
path_name = '/resource/{resource_id}'
path_parameter_name = 'resource_id'
@pytest.fixture
def parameter(self):
return Parameter(self.path_parameter_name, 'path')
@pytest.fixture
def parameters(self, parameter):
return {
self.path_parameter_name: parameter
}
@pytest.fixture
def path(self, operations, parameters):
return Path(self.path_name, operations, parameters=parameters)
class BaseTestSpecServer(object):
@pytest.fixture
def info(self):
return Info('Test schema', '1.0')
@pytest.fixture
def operation(self):
return Operation('get', self.path_name, {}, {})
@pytest.fixture
def operations(self, operation):
return {
'get': operation,
}
@pytest.fixture
def spec(self, info, paths, servers):
return Spec(info, paths, servers)
@pytest.fixture
def finder(self, spec):
return PathFinder(spec)
class BaseTestPathServer(BaseTestSpecServer):
@pytest.fixture
def path(self, operations, servers):
return Path(self.path_name, operations, servers=servers)
@pytest.fixture
def spec(self, info, paths):
return Spec(info, paths)
class BaseTestOperationServer(BaseTestSpecServer):
@pytest.fixture
def operation(self, servers):
return Operation('get', self.path_name, {}, {}, servers=servers)
@pytest.fixture
def spec(self, info, paths):
return Spec(info, paths)
class BaseTestServerNotFound(object):
@pytest.fixture
def servers(self):
return []
def test_raises(self, finder):
request_uri = '/resource'
request = MockRequest(
'http://petstore.swagger.io', 'get', request_uri)
with pytest.raises(ServerNotFound):
finder.find(request)
class BaseTestOperationNotFound(object):
@pytest.fixture
def operations(self):
return {}
def test_raises(self, finder):
request_uri = '/resource'
request = MockRequest(
'http://petstore.swagger.io', 'get', request_uri)
with pytest.raises(OperationNotFound):
finder.find(request)
class BaseTestValid(object):
def test_simple(self, finder, path, operation, server):
request_uri = '/resource'
request = MockRequest(
'http://petstore.swagger.io', 'get', request_uri)
result = finder.find(request)
path_result = TemplateResult(self.path_name, {})
server_result = TemplateResult(self.server_url, {})
assert result == (
path, operation, server, path_result, server_result,
)
class BaseTestVariableValid(object):
@pytest.mark.parametrize('version', ['v1', 'v2'])
def test_variable(self, finder, path, operation, server, version):
request_uri = '/{0}/resource'.format(version)
request = MockRequest(
'http://petstore.swagger.io', 'get', request_uri)
result = finder.find(request)
path_result = TemplateResult(self.path_name, {})
server_result = TemplateResult(self.server_url, {'version': version})
assert result == (
path, operation, server, path_result, server_result,
)
class BaseTestPathVariableValid(object):
@pytest.mark.parametrize('res_id', ['111', '222'])
def test_path_variable(self, finder, path, operation, server, res_id):
request_uri = '/resource/{0}'.format(res_id)
request = MockRequest(
'http://petstore.swagger.io', 'get', request_uri)
result = finder.find(request)
path_result = TemplateResult(self.path_name, {'resource_id': res_id})
server_result = TemplateResult(self.server_url, {})
assert result == (
path, operation, server, path_result, server_result,
)
class BaseTestPathNotFound(object):
@pytest.fixture
def paths(self):
return {}
def test_raises(self, finder):
request_uri = '/resource'
request = MockRequest(
'http://petstore.swagger.io', 'get', request_uri)
with pytest.raises(PathNotFound):
finder.find(request)
class TestSpecSimpleServerServerNotFound(
BaseTestServerNotFound, BaseTestSpecServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestSpecSimpleServerOperationNotFound(
BaseTestOperationNotFound, BaseTestSpecServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestSpecSimpleServerPathNotFound(
BaseTestPathNotFound, BaseTestSpecServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestOperationSimpleServerServerNotFound(
BaseTestServerNotFound, BaseTestOperationServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestOperationSimpleServerOperationNotFound(
BaseTestOperationNotFound, BaseTestOperationServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestOperationSimpleServerPathNotFound(
BaseTestPathNotFound, BaseTestOperationServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestPathSimpleServerServerNotFound(
BaseTestServerNotFound, BaseTestPathServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestPathSimpleServerOperationNotFound(
BaseTestOperationNotFound, BaseTestPathServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestPathSimpleServerPathNotFound(
BaseTestPathNotFound, BaseTestPathServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestSpecSimpleServerValid(
BaseTestValid, BaseTestSpecServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestOperationSimpleServerValid(
BaseTestValid, BaseTestOperationServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestPathSimpleServerValid(
BaseTestValid, BaseTestPathServer,
BaseTestSimplePath, BaseTestSimpleServer):
pass
class TestSpecSimpleServerVariablePathValid(
BaseTestPathVariableValid, BaseTestSpecServer,
BaseTestVariablePath, BaseTestSimpleServer):
pass
class TestOperationSimpleServerVariablePathValid(
BaseTestPathVariableValid, BaseTestOperationServer,
BaseTestVariablePath, BaseTestSimpleServer):
pass
class TestPathSimpleServerVariablePathValid(
BaseTestPathVariableValid, BaseTestPathServer,
BaseTestVariablePath, BaseTestSimpleServer):
pass
class TestSpecVariableServerServerNotFound(
BaseTestServerNotFound, BaseTestSpecServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestSpecVariableServerOperationNotFound(
BaseTestOperationNotFound, BaseTestSpecServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestSpecVariableServerPathNotFound(
BaseTestPathNotFound, BaseTestSpecServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestOperationVariableServerServerNotFound(
BaseTestServerNotFound, BaseTestOperationServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestOperationVariableServerOperationNotFound(
BaseTestOperationNotFound, BaseTestOperationServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestOperationVariableServerPathNotFound(
BaseTestPathNotFound, BaseTestOperationServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestPathVariableServerServerNotFound(
BaseTestServerNotFound, BaseTestPathServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestPathVariableServerOperationNotFound(
BaseTestOperationNotFound, BaseTestPathServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestPathVariableServerPathNotFound(
BaseTestPathNotFound, BaseTestPathServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestSpecVariableServerValid(
BaseTestVariableValid, BaseTestSpecServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestOperationVariableServerValid(
BaseTestVariableValid, BaseTestOperationServer,
BaseTestSimplePath, BaseTestVariableServer):
pass
class TestPathVariableServerValid(
BaseTestVariableValid, BaseTestPathServer,
BaseTestSimplePath, BaseTestVariableServer):
pass

18
tests/unit/templating/test_paths_util.py
View file

@ -1,18 +0,0 @@
from openapi_core.templating.paths.util import path_qs
class TestPathQs(object):
def test_path(self):
url = 'https://test.com:1234/path'
result = path_qs(url)
assert result == '/path'
def test_query(self):
url = 'https://test.com:1234/path?query=1'
result = path_qs(url)
assert result == '/path?query=1'