diff --git a/falcon/media/validators/__init__.py b/falcon/media/validators/__init__.py index f5172b066..463827d7d 100644 --- a/falcon/media/validators/__init__.py +++ b/falcon/media/validators/__init__.py @@ -1 +1,2 @@ from . import jsonschema +from . import jsonschema_rs diff --git a/falcon/media/validators/base.py b/falcon/media/validators/base.py new file mode 100644 index 000000000..e2fc515b3 --- /dev/null +++ b/falcon/media/validators/base.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +from abc import abstractmethod +from functools import wraps +from inspect import iscoroutinefunction +from typing import Any, Callable, Optional, Tuple, Type, TypeVar, Union + +import falcon + + +class Validator: + """Base validator class.""" + + exceptions: Union[Tuple[Type[Exception], ...], Type[Exception]] + """The exceptions raised by the validation library""" + + @classmethod + @abstractmethod + def from_schema(cls, schema: Any) -> Validator: + """Construct the class from a schema object.""" + + @abstractmethod + def validate(self, media: Any) -> None: + """Validate the input media.""" + + @abstractmethod + def get_exception_message(self, exception: Exception) -> Optional[str]: + """Return a message from an exception.""" + + +_T = TypeVar('_T') + + +def validator_factory( + validator: Type[Validator], req_schema: Any, resp_schema: Any, is_async: bool +) -> Callable[[_T], _T]: + """Create a validator decorator for that uses the specified ``Validator`` class. + + Args: + validator (Type[Validator]): The validator class. + req_schema (Any): The schema used in the request body. Type will depend on + what is accepted by ``Validator.from_schema``. + When ``None`` validation will be skipped. + resp_schema (Any): The schema used in the response body. Type will depend + on what is accepted by ``Validator.from_schema``. + When ``None`` validation will be skipped. + is_async (bool): Set to ``True`` to force use of the async validator. + """ + + def decorator(func: _T) -> _T: + if iscoroutinefunction(func) or is_async: + return _validate_async(validator, func, req_schema, resp_schema) + + return _validate(validator, func, req_schema, resp_schema) + + return decorator + + +def _validate( + validator: Type[Validator], func, req_schema: Any, resp_schema: Any +) -> Any: + req_validator = None if req_schema is None else validator.from_schema(req_schema) + resp_validator = None if resp_schema is None else validator.from_schema(resp_schema) + + @wraps(func) + def wrapper(self, req, resp, *args, **kwargs): + if req_validator is not None: + try: + req_validator.validate(req.media) + except req_validator.exceptions as ex: + raise falcon.MediaValidationError( + title='Request data failed validation', + description=req_validator.get_exception_message(ex), + ) from ex + + result = func(self, req, resp, *args, **kwargs) + + if resp_validator is not None: + try: + resp_validator.validate(resp.media) + except resp_validator.exceptions as ex: + raise falcon.HTTPInternalServerError( + title='Response data failed validation' + # Do not return 'e.message' in the response to + # prevent info about possible internal response + # formatting bugs from leaking out to users. + ) from ex + + return result + + return wrapper + + +def _validate_async( + validator: Type[Validator], func, req_schema: Any, resp_schema: Any +) -> Any: + req_validator = None if req_schema is None else validator.from_schema(req_schema) + resp_validator = None if resp_schema is None else validator.from_schema(resp_schema) + + @wraps(func) + async def wrapper(self, req, resp, *args, **kwargs): + if req_validator is not None: + m = await req.get_media() + + try: + req_validator.validate(m) + except req_validator.exceptions as ex: + raise falcon.MediaValidationError( + title='Request data failed validation', + description=req_validator.get_exception_message(ex), + ) from ex + + result = await func(self, req, resp, *args, **kwargs) + + if resp_validator is not None: + try: + resp_validator.validate(resp.media) + except resp_validator.exceptions as ex: + raise falcon.HTTPInternalServerError( + title='Response data failed validation' + # Do not return 'e.message' in the response to + # prevent info about possible internal response + # formatting bugs from leaking out to users. + ) from ex + + return result + + return wrapper diff --git a/falcon/media/validators/jsonschema.py b/falcon/media/validators/jsonschema.py index 8fc53ade9..168ffcfb6 100644 --- a/falcon/media/validators/jsonschema.py +++ b/falcon/media/validators/jsonschema.py @@ -1,7 +1,8 @@ -from functools import wraps -from inspect import iscoroutinefunction +from __future__ import annotations -import falcon +from typing import Any, Optional, TYPE_CHECKING + +from . import base as _base try: import jsonschema @@ -9,7 +10,7 @@ pass -def validate(req_schema=None, resp_schema=None, is_async=False): +def validate(req_schema: Any = None, resp_schema: Any = None, is_async: bool = False): """Validate ``req.media`` using JSON Schema. This decorator provides standard JSON Schema validation via the @@ -99,78 +100,26 @@ async def on_post(self, req, resp): """ - def decorator(func): - if iscoroutinefunction(func) or is_async: - return _validate_async(func, req_schema, resp_schema) - - return _validate(func, req_schema, resp_schema) - - return decorator - - -def _validate(func, req_schema=None, resp_schema=None): - @wraps(func) - def wrapper(self, req, resp, *args, **kwargs): - if req_schema is not None: - try: - jsonschema.validate( - req.media, req_schema, format_checker=jsonschema.FormatChecker() - ) - except jsonschema.ValidationError as ex: - raise falcon.MediaValidationError( - title='Request data failed validation', description=ex.message - ) from ex - - result = func(self, req, resp, *args, **kwargs) - - if resp_schema is not None: - try: - jsonschema.validate( - resp.media, resp_schema, format_checker=jsonschema.FormatChecker() - ) - except jsonschema.ValidationError as ex: - raise falcon.HTTPInternalServerError( - title='Response data failed validation' - # Do not return 'e.message' in the response to - # prevent info about possible internal response - # formatting bugs from leaking out to users. - ) from ex - - return result - - return wrapper - - -def _validate_async(func, req_schema=None, resp_schema=None): - @wraps(func) - async def wrapper(self, req, resp, *args, **kwargs): - if req_schema is not None: - m = await req.get_media() - - try: - jsonschema.validate( - m, req_schema, format_checker=jsonschema.FormatChecker() - ) - except jsonschema.ValidationError as ex: - raise falcon.MediaValidationError( - title='Request data failed validation', description=ex.message - ) from ex - - result = await func(self, req, resp, *args, **kwargs) - - if resp_schema is not None: - try: - jsonschema.validate( - resp.media, resp_schema, format_checker=jsonschema.FormatChecker() - ) - except jsonschema.ValidationError as ex: - raise falcon.HTTPInternalServerError( - title='Response data failed validation' - # Do not return 'e.message' in the response to - # prevent info about possible internal response - # formatting bugs from leaking out to users. - ) from ex - - return result - - return wrapper + return _base.validator_factory( + JsonSchemaValidator, req_schema, resp_schema, is_async + ) + + +class JsonSchemaValidator(_base.Validator): + def __init__(self, schema: Any) -> None: + self.schema = schema + self.exceptions = jsonschema.ValidationError + + @classmethod + def from_schema(cls, schema: Any) -> JsonSchemaValidator: + return cls(schema) + + def validate(self, media: Any) -> None: + jsonschema.validate( + media, self.schema, format_checker=jsonschema.FormatChecker() + ) + + def get_exception_message(self, exception: Exception) -> Optional[str]: + if TYPE_CHECKING: + assert isinstance(exception, jsonschema.ValidationError) + return exception.message diff --git a/falcon/media/validators/jsonschema_rs.py b/falcon/media/validators/jsonschema_rs.py new file mode 100644 index 000000000..cde9f6653 --- /dev/null +++ b/falcon/media/validators/jsonschema_rs.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +from typing import Any, Optional, TYPE_CHECKING + +from . import base as _base + +try: + import jsonschema_rs +except ImportError: # pragma: nocover + pass + + +def validate(req_schema: Any = None, resp_schema: Any = None, is_async: bool = False): + """Validate ``req.media`` using JSON Schema. + + This decorator provides standard JSON Schema validation via the + ``jsonschema_rs`` package available from PyPI. + + In the case of failed request media validation, an instance of + :class:`~falcon.MediaValidationError` is raised by the decorator. By + default, this error is rendered as a 400 (:class:`~falcon.HTTPBadRequest`) + response with the ``title`` and ``description`` attributes explaining the + validation failure, but this behavior can be modified by adding a + custom error :func:`handler ` for + :class:`~falcon.MediaValidationError`. + + Note: + The ``jsonschema_rs`` package must be installed separately in order to use + this decorator, as Falcon does not install it by default. + + See `jsonschema_rs PyPi `_ for more + information on defining a compatible dictionary. + + Keyword Args: + req_schema (dict or str): A dictionary that follows the JSON + Schema specification. The request will be validated against this + schema. + Can be also a json string that will be loaded by the jsonschema_rs library + resp_schema (dict or str): A dictionary that follows the JSON + Schema specification. The response will be validated against this + schema. + Can be also a json string that will be loaded by the jsonschema_rs library + is_async (bool): Set to ``True`` for ASGI apps to provide a hint that + the decorated responder is a coroutine function (i.e., that it + is defined with ``async def``) or that it returns an awaitable + coroutine object. + + Normally, when the function source is declared using ``async def``, + the resulting function object is flagged to indicate it returns a + coroutine when invoked, and this can be automatically detected. + However, it is possible to use a regular function to return an + awaitable coroutine object, in which case a hint is required to let + the framework know what to expect. Also, a hint is always required + when using a cythonized coroutine function, since Cython does not + flag them in a way that can be detected in advance, even when the + function is declared using ``async def``. + + Example: + + .. tabs:: + + .. tab:: WSGI + + .. code:: python + + from falcon.media.validators import jsonschema_rs + + # -- snip -- + + @jsonschema_rs.validate(my_post_schema) + def on_post(self, req, resp): + + # -- snip -- + + .. tab:: ASGI + + .. code:: python + + from falcon.media.validators import jsonschema_rs + + # -- snip -- + + @jsonschema_rs.validate(my_post_schema) + async def on_post(self, req, resp): + + # -- snip -- + + .. tab:: ASGI (Cythonized App) + + .. code:: python + + from falcon.media.validators import jsonschema_rs + + # -- snip -- + + @jsonschema_rs.validate(my_post_schema, is_async=True) + async def on_post(self, req, resp): + + # -- snip -- + + """ + + return _base.validator_factory( + JsonSchemaRsValidator, req_schema, resp_schema, is_async + ) + + +class JsonSchemaRsValidator(_base.Validator): + def __init__(self, schema: Any) -> None: + self.schema = schema + if isinstance(schema, str): + self.validator = jsonschema_rs.JSONSchema.from_str(schema) + else: + self.validator = jsonschema_rs.JSONSchema(schema) + self.exceptions = jsonschema_rs.ValidationError + + @classmethod + def from_schema(cls, schema: Any) -> JsonSchemaRsValidator: + return cls(schema) + + def validate(self, media: Any) -> None: + self.validator.validate(media) + + def get_exception_message(self, exception: Exception) -> Optional[str]: + if TYPE_CHECKING: + assert isinstance(exception, jsonschema_rs.ValidationError) + return exception.message diff --git a/tests/asgi/_cythonized.pyx b/tests/asgi/_cythonized.pyx index 091e07358..1ea41befc 100644 --- a/tests/asgi/_cythonized.pyx +++ b/tests/asgi/_cythonized.pyx @@ -4,26 +4,33 @@ from collections import Counter import time import falcon -from falcon.media.validators.jsonschema import validate - +from falcon.media import validators +try: + import jsonschema +except ImportError: + jsonschema = None +try: + import jsonschema_rs +except ImportError: + jsonschema_rs = None _MESSAGE_SCHEMA = { - 'definitions': {}, - '$schema': 'http://json-schema.org/draft-07/schema#', - '$id': 'http://example.com/root.json', - 'type': 'object', - 'title': 'The Root Schema', - 'required': ['message'], - 'properties': { - 'message': { - '$id': '#/properties/message', - 'type': 'string', - 'title': 'The Message Schema', - 'default': '', - 'examples': ['hello world'], - 'pattern': '^(.*)$' - } - } + 'definitions': {}, + '$schema': 'http://json-schema.org/draft-07/schema#', + '$id': 'http://example.com/root.json', + 'type': 'object', + 'title': 'The Root Schema', + 'required': ['message'], + 'properties': { + 'message': { + '$id': '#/properties/message', + 'type': 'string', + 'title': 'The Message Schema', + 'default': '', + 'examples': ['hello world'], + 'pattern': '^(.*)$' + } + } } @@ -43,20 +50,37 @@ class NOPClass: pass -class TestResourceWithValidation: - @validate(resp_schema=_MESSAGE_SCHEMA, is_async=True) - async def on_get(self, req, resp): - resp.media = { - 'message': 'hello world' - } +if jsonschema: + class TestResourceWithValidation: + @validators.jsonschema.validate(resp_schema=_MESSAGE_SCHEMA, is_async=True) + async def on_get(self, req, resp): + resp.media = { + 'message': 'hello world' + } -class TestResourceWithValidationNoHint: - @validate(resp_schema=_MESSAGE_SCHEMA) - async def on_get(self, req, resp): - resp.media = { - 'message': 'hello world' - } + class TestResourceWithValidationNoHint: + @validators.jsonschema.validate(resp_schema=_MESSAGE_SCHEMA) + async def on_get(self, req, resp): + resp.media = { + 'message': 'hello world' + } + +if jsonschema_rs: + class TestResourceWithValidationRs: + @validators.jsonschema_rs.validate(resp_schema=_MESSAGE_SCHEMA, is_async=True) + async def on_get(self, req, resp): + resp.media = { + 'message': 'hello world' + } + + + class TestResourceWithValidationNoHintRs: + @validators.jsonschema_rs.validate(resp_schema=_MESSAGE_SCHEMA) + async def on_get(self, req, resp): + resp.media = { + 'message': 'hello world' + } class TestResourceWithScheduledJobs: @@ -85,7 +109,7 @@ class TestResourceWithScheduledJobsAsyncRequired: pass # NOTE(kgriffs): This will fail later since we can't detect - # up front that it isn't a coroutine function. + # up front that it isn't a coroutine function. resp.schedule(background_job_sync) diff --git a/tests/asgi/test_cythonized_asgi.py b/tests/asgi/test_cythonized_asgi.py index 744dcd952..e6bf157da 100644 --- a/tests/asgi/test_cythonized_asgi.py +++ b/tests/asgi/test_cythonized_asgi.py @@ -20,6 +20,9 @@ if pyximport: from . import _cythonized # type: ignore + jsonschema = _cythonized.jsonschema + jsonschema_rs = _cythonized.jsonschema_rs + _CYTHON_FUNC_TEST_TYPES = [ _cythonized.nop_method, _cythonized.nop_method_async, @@ -29,6 +32,7 @@ _cythonized.NOPClass().nop_method_async, ] else: + jsonschema = jsonschema_rs = None _CYTHON_FUNC_TEST_TYPES = [] from _util import disable_asgi_non_coroutine_wrapping # NOQA @@ -83,6 +87,7 @@ def test_not_cython_func(func): @pytest.mark.skipif(not pyximport, reason='Cython not installed') +@pytest.mark.skipif(jsonschema is None, reason='jsonschema not installed') def test_jsonchema_validator(client): with disable_asgi_non_coroutine_wrapping(): if CYTHON_COROUTINE_HINT: @@ -98,6 +103,23 @@ def test_jsonchema_validator(client): client.simulate_get() +@pytest.mark.skipif(not pyximport, reason='Cython not installed') +@pytest.mark.skipif(jsonschema_rs is None, reason='jsonschema_rs not installed') +def test_jsonchema_rs_validator(client): + with disable_asgi_non_coroutine_wrapping(): + if CYTHON_COROUTINE_HINT: + client.app.add_route('/', _cythonized.TestResourceWithValidationNoHintRs()) + else: + with pytest.raises(TypeError): + client.app.add_route( + '/wowsuchfail', _cythonized.TestResourceWithValidationNoHintRs() + ) + + client.app.add_route('/', _cythonized.TestResourceWithValidationRs()) + + client.simulate_get() + + @pytest.mark.skipif(not pyximport, reason='Cython not installed') def test_scheduled_jobs(client): resource = _cythonized.TestResourceWithScheduledJobs() diff --git a/tests/test_validators.py b/tests/test_validators.py index a7f5ed273..41b6a7136 100644 --- a/tests/test_validators.py +++ b/tests/test_validators.py @@ -1,9 +1,14 @@ -import typing # NOQA: F401 +import json +import typing try: - import jsonschema as _jsonschema # NOQA + import jsonschema except ImportError: - pass + jsonschema = None # type: ignore +try: + import jsonschema_rs +except ImportError: + jsonschema_rs = None # type: ignore import pytest import falcon @@ -13,14 +18,14 @@ from _util import create_app, disable_asgi_non_coroutine_wrapping # NOQA -# NOTE(kgriffs): Default to None if missing. We do it like this, here, instead -# of in the body of the except statement, above, to avoid flake8 import -# ordering errors. -jsonschema = globals().get('_jsonschema') +# # NOTE(kgriffs): Default to None if missing. We do it like this, here, instead +# # of in the body of the except statement, above, to avoid flake8 import +# # ordering errors. +# jsonschema = globals().get('_jsonschema') _VALID_MEDIA = {'message': 'something'} -_INVALID_MEDIA = {} # type: typing.Dict[str, str] +_INVALID_MEDIA: typing.Dict[str, str] = {} _TEST_SCHEMA = { @@ -34,64 +39,94 @@ } -skip_missing_dep = pytest.mark.skipif( - jsonschema is None, reason='jsonschema dependency not found' +@pytest.fixture( + params=[ + pytest.param( + 'jsonschema', + marks=pytest.mark.skipif( + jsonschema is None, reason='jsonschema dependency not found' + ), + ), + pytest.param( + 'jsonschema_rs', + marks=pytest.mark.skipif( + jsonschema_rs is None, reason='jsonschema_rs dependency not found' + ), + ), + pytest.param( + 'jsonschema_rs_string', + marks=pytest.mark.skipif( + jsonschema_rs is None, reason='jsonschema_rs dependency not found' + ), + ), + ] ) - - -class Resource: - @validators.jsonschema.validate(req_schema=_TEST_SCHEMA) - def request_validated(self, req, resp): - assert req.media is not None - return resp - - @validators.jsonschema.validate(resp_schema=_TEST_SCHEMA) - def response_validated(self, req, resp): - assert resp.media is not None - return resp - - @validators.jsonschema.validate(req_schema=_TEST_SCHEMA, resp_schema=_TEST_SCHEMA) - def both_validated(self, req, resp): - assert req.media is not None - assert resp.media is not None - return req, resp - - @validators.jsonschema.validate(req_schema=_TEST_SCHEMA, resp_schema=_TEST_SCHEMA) - def on_put(self, req, resp): - assert req.media is not None - resp.media = _VALID_MEDIA - - -class ResourceAsync: - @validators.jsonschema.validate(req_schema=_TEST_SCHEMA) - async def request_validated(self, req, resp): - # NOTE(kgriffs): Verify that we can await req.get_media() multiple times - for i in range(3): +def resources(request): + the_schema = _TEST_SCHEMA + if request.param == 'jsonschema': + validate = validators.jsonschema.validate + elif request.param == 'jsonschema_rs': + validate = validators.jsonschema_rs.validate + elif request.param == 'jsonschema_rs_string': + validate = validators.jsonschema_rs.validate + the_schema = json.dumps(_TEST_SCHEMA) + else: + pytest.fail(request.param) + + class Resource: + @validate(req_schema=the_schema) + def request_validated(self, req, resp): + assert req.media is not None + return resp + + @validate(resp_schema=the_schema) + def response_validated(self, req, resp): + assert resp.media is not None + return resp + + @validate(req_schema=the_schema, resp_schema=the_schema) + def both_validated(self, req, resp): + assert req.media is not None + assert resp.media is not None + return req, resp + + @validate(req_schema=the_schema, resp_schema=the_schema) + def on_put(self, req, resp): + assert req.media is not None + resp.media = _VALID_MEDIA + + class ResourceAsync: + @validate(req_schema=the_schema) + async def request_validated(self, req, resp): + # NOTE(kgriffs): Verify that we can await req.get_media() multiple times + for i in range(3): + m = await req.get_media() + assert m == _VALID_MEDIA + + assert m is not None + return resp + + @validate(resp_schema=the_schema) + async def response_validated(self, req, resp): + assert resp.media is not None + return resp + + @validate(req_schema=the_schema, resp_schema=the_schema) + async def both_validated(self, req, resp): m = await req.get_media() - assert m == _VALID_MEDIA + assert m is not None - assert m is not None - return resp + assert resp.media is not None - @validators.jsonschema.validate(resp_schema=_TEST_SCHEMA) - async def response_validated(self, req, resp): - assert resp.media is not None - return resp + return req, resp - @validators.jsonschema.validate(req_schema=_TEST_SCHEMA, resp_schema=_TEST_SCHEMA) - async def both_validated(self, req, resp): - m = await req.get_media() - assert m is not None - - assert resp.media is not None - - return req, resp + @validate(req_schema=the_schema, resp_schema=the_schema) + async def on_put(self, req, resp): + m = await req.get_media() + assert m is not None + resp.media = _VALID_MEDIA - @validators.jsonschema.validate(req_schema=_TEST_SCHEMA, resp_schema=_TEST_SCHEMA) - async def on_put(self, req, resp): - m = await req.get_media() - assert m is not None - resp.media = _VALID_MEDIA + return Resource, ResourceAsync class _MockReq: @@ -116,8 +151,8 @@ def __init__(self, valid=True): self.media = _VALID_MEDIA if valid else {} -def call_method(asgi, method_name, *args): - resource = ResourceAsync() if asgi else Resource() +def call_method(resources, asgi, method_name, *args): + resource = resources[1]() if asgi else resources[0]() if asgi: return falcon.async_to_sync(getattr(resource, method_name), *args) @@ -125,71 +160,72 @@ def call_method(asgi, method_name, *args): return getattr(resource, method_name)(*args) -@skip_missing_dep -def test_req_schema_validation_success(asgi): +def test_req_schema_validation_success(asgi, resources): data = MockResp() - assert call_method(asgi, 'request_validated', MockReq(asgi), data) is data + assert ( + call_method(resources, asgi, 'request_validated', MockReq(asgi), data) is data + ) -@skip_missing_dep @pytest.mark.parametrize( 'exception_cls', [falcon.HTTPBadRequest, falcon.MediaValidationError] ) -def test_req_schema_validation_failure(asgi, exception_cls): +def test_req_schema_validation_failure(asgi, exception_cls, resources): with pytest.raises(exception_cls) as excinfo: - call_method(asgi, 'request_validated', MockReq(asgi, False), None) + call_method(resources, asgi, 'request_validated', MockReq(asgi, False), None) - assert excinfo.value.description == "'message' is a required property" + desc = excinfo.value.description.replace('"', "'") + assert desc == "'message' is a required property" -@skip_missing_dep -def test_resp_schema_validation_success(asgi): +def test_resp_schema_validation_success(asgi, resources): data = MockResp() - assert call_method(asgi, 'response_validated', MockReq(asgi), data) is data + assert ( + call_method(resources, asgi, 'response_validated', MockReq(asgi), data) is data + ) -@skip_missing_dep -def test_resp_schema_validation_failure(asgi): +def test_resp_schema_validation_failure(asgi, resources): with pytest.raises(falcon.HTTPInternalServerError) as excinfo: - call_method(asgi, 'response_validated', MockReq(asgi), MockResp(False)) + call_method( + resources, asgi, 'response_validated', MockReq(asgi), MockResp(False) + ) assert excinfo.value.title == 'Response data failed validation' -@skip_missing_dep -def test_both_schemas_validation_success(asgi): +def test_both_schemas_validation_success(asgi, resources): req = MockReq(asgi) resp = MockResp() - result = call_method(asgi, 'both_validated', req, resp) + result = call_method(resources, asgi, 'both_validated', req, resp) assert result[0] is req assert result[1] is resp client = testing.TestClient(create_app(asgi)) - resource = ResourceAsync() if asgi else Resource() + resource = resources[1]() if asgi else resources[0]() client.app.add_route('/test', resource) result = client.simulate_put('/test', json=_VALID_MEDIA) assert result.json == resp.media -@skip_missing_dep -def test_both_schemas_validation_failure(asgi): +def test_both_schemas_validation_failure(asgi, resources): bad_resp = MockResp(False) with pytest.raises(falcon.HTTPInternalServerError) as excinfo: - call_method(asgi, 'both_validated', MockReq(asgi), bad_resp) + call_method(resources, asgi, 'both_validated', MockReq(asgi), bad_resp) assert excinfo.value.title == 'Response data failed validation' with pytest.raises(falcon.HTTPBadRequest) as excinfo: - call_method(asgi, 'both_validated', MockReq(asgi, False), MockResp()) + call_method(resources, asgi, 'both_validated', MockReq(asgi, False), MockResp()) assert excinfo.value.title == 'Request data failed validation' client = testing.TestClient(create_app(asgi)) - resource = ResourceAsync() if asgi else Resource() + resource = resources[1]() if asgi else resources[0]() with disable_asgi_non_coroutine_wrapping(): client.app.add_route('/test', resource) diff --git a/tox.ini b/tox.ini index e6ca31342..b1559ec23 100644 --- a/tox.ini +++ b/tox.ini @@ -92,6 +92,7 @@ commands = coverage run -m pytest tests --ignore=tests/asgi [] deps = {[testenv]deps} pytest-randomly jsonschema + jsonschema_rs commands = coverage run -m pytest tests [] [testenv:pytest_sans_msgpack] @@ -104,6 +105,7 @@ basepython = python3.12 deps = {[testenv]deps} pytest-randomly jsonschema + jsonschema_rs commands = {[with-coverage]commands} [testenv:py312_sans_msgpack] @@ -117,6 +119,7 @@ basepython = python3.12 deps = {[testenv]deps} pytest-randomly jsonschema + jsonschema_rs commands = pytest tests [] # -------------------------------------------------------------------- @@ -139,6 +142,7 @@ basepython = python3.10 deps = {[with-debug-tools]deps} uvicorn jsonschema + jsonschema_rs # -------------------------------------------------------------------- # mypy @@ -148,6 +152,7 @@ deps = {[with-debug-tools]deps} skipsdist = True skip_install = True deps = mypy + jsonschema_rs types-jsonschema commands = python {toxinidir}/tools/clean.py "{toxinidir}/falcon" mypy falcon @@ -155,6 +160,7 @@ commands = python {toxinidir}/tools/clean.py "{toxinidir}/falcon" [testenv:mypy_tests] deps = {[testenv]deps} mypy + jsonschema_rs types-requests types-PyYAML types-ujson