Skip to content

Instantly share code, notes, and snippets.

@spumer
Created November 21, 2022 12:05
Show Gist options
  • Save spumer/033a28fdc24922a4dee8e49c5ca12c06 to your computer and use it in GitHub Desktop.
Save spumer/033a28fdc24922a4dee8e49c5ca12c06 to your computer and use it in GitHub Desktop.
pytest fixture for fastapi-jsonrpc. Check returned errors described in swagger (app.method(..., errors=[...]))
import collections
import contextlib
import traceback
import jsonrpcapi
import pytest
from testing import AnyDict
@pytest.fixture()
def all_captured_jsonrpc_error_responses(api_app):
errors = collections.defaultdict(list)
@contextlib.asynccontextmanager
async def tracking_middleware(ctx: jsonrpcapi.JsonRpcContext):
try:
yield
except Exception as exc:
if not isinstance(exc, (jsonrpcapi.APIError, jsonrpcapi.BaseError)):
exc_traceback = traceback.format_exception(
None, # python >=3.5: The exc argument is ignored and inferred from the type of value.
value=exc,
tb=exc.__traceback__,
limit=30,
)
pytest.fail(
'Method should raise `jsonrpcapi.APIError` or `jsonrpcapi.BaseError`. '
'Common exception raised instead: \n' + ''.join(exc_traceback)
)
raise
finally:
if ctx.raw_response and 'result' not in ctx.raw_response:
errors[ctx.method_route].append(ctx.raw_response)
entrypoints = {r.entrypoint for r in api_app.routes if isinstance(r, jsonrpcapi.MethodRoute)}
for ep in entrypoints:
ep.middlewares.insert(0, tracking_middleware)
try:
yield errors
finally:
for ep in entrypoints:
ep.middlewares.remove(tracking_middleware)
def _get_first_not_listed_in_method_errors(method_route, captured_error_responses):
expected_errors = method_route.entrypoint.entrypoint_route.errors + method_route.errors
expected_error_data = []
# ValidationError всегда должен содержать подробности ошибки (APIError)
expected_errors.remove(jsonrpcapi.ValidationError)
for err_cls in expected_errors:
if issubclass(err_cls, jsonrpcapi.APIError):
# fmt: off
expected_error_data.append({
'code': jsonrpcapi.ValidationError.CODE,
'message': jsonrpcapi.ValidationError.MESSAGE,
'data': {
'errors': [AnyDict({
# Нам интересны только code и message
# object_name и meta это контекст, игнорируем его
'code': err_cls.code,
'message': err_cls.message,
})],
},
})
# fmt: on
else:
err_resp = err_cls().get_resp()['error']
# Оборачиваем в AnyDict, чтобы игнорировать доп. поля, которые может вернуть сервер
# Обычно в них приходит контекстная информация
expected_error_data.append(AnyDict(err_resp))
for cap_err_resp in captured_error_responses:
error_data = cap_err_resp['error']
if error_data not in expected_error_data:
return error_data
return None
@pytest.fixture()
def _check_all_captured_jsonrpc_error_responses_listed_in_method_errors(all_captured_jsonrpc_error_responses):
yield
for method_route, captured_error_responses in all_captured_jsonrpc_error_responses.items():
not_listed_error = _get_first_not_listed_in_method_errors(method_route, captured_error_responses)
if not_listed_error is not None:
pytest.fail(f'Error {not_listed_error!r} not listed in {method_route.name!r} method errors, but returned')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment