Significantly improve the error code.

This commit is contained in:
Dorian 2017-03-01 09:18:44 -05:00
parent d362883d72
commit 4aff4a8973
7 changed files with 121 additions and 52 deletions

View File

@ -20,7 +20,7 @@ test: test-api test-webapp
# Runs api tests
test-api: stop build-api
docker-compose up -d db api
docker-compose up -d --build db api
docker-compose run api inv test.style
docker-compose run api \
inv test.server --db-connection=$(DB_CONNECTION) --server-host=api:5000 --verbosity=3

View File

@ -21,7 +21,7 @@ def make_rookeries_app():
"""Creates and setups a Rookeries webapp."""
logger.info("*** Rookeries API Server - version %s ***", __version__)
logger.info("Starting Rookeries API server...")
web_app = app.make_json_app(__name__)
web_app = app.make_api_app(__name__)
web_app.register_blueprint(main.rookeries_app)
logger.info("Configuring Rookeries:")

View File

@ -5,54 +5,14 @@ Creates and configures the Rookeries Flask app
:license: AGPL v3+
"""
import http
import flask
from werkzeug import exceptions
from rookeries import database, security
from rookeries import database, errors, security
def make_json_app(import_name, **kwargs):
"""Creates a JSON-oriented Flask app.
All error responses that you don't specifically manage yourself will have a application/json content type, and
will contain JSON in following format: ::
{ "error": {
"code": 405,
"message": "Method Not Allowed"
}
}
Originally Pavel Repin's `snippet receipe for JSON response apps <http://flask.pocoo.org/snippets/83/>`_.
:param import_name: The name of the app to import.
"""
def make_json_error(error):
"""
Make a JSON error response.
:param error: The error message to turn into JSON.
"""
error_message = error.name if isinstance(error, exceptions.HTTPException) else str(error)
error_code = http.HTTPStatus.INTERNAL_SERVER_ERROR
if isinstance(error, exceptions.HTTPException):
error_code = error.code
error_json = {'error': {'code': error_code, 'message': error_message}}
response = flask.jsonify(error_json)
response.status_code = error_code
return response
def make_api_app(import_name, **kwargs):
api_app = flask.Flask(import_name, **kwargs)
for code in iter(exceptions.default_exceptions):
api_app.register_error_handler(code, make_json_error)
# TODO: Better handling of unexpected errors.
errors.attach_error_handlers_to_api_app(api_app)
database.SQLAlchemy.init_app(api_app)
security.jwt.init_app(api_app)
return api_app

76
api/rookeries/errors.py Normal file
View File

@ -0,0 +1,76 @@
"""
Error handlers for the API.
:copyright: Copyright 2013-2017, Dorian Pula <dorian.pula@amber-penguin-software.ca>
:license: AGPL v3+
"""
import http
import logging
import flask
from werkzeug import exceptions
logger = logging.getLogger(__name__)
HTTP_ERRORS_WITH_PATHS = [http.HTTPStatus.NOT_FOUND, http.HTTPStatus.UNAUTHORIZED]
HTTP_ERRORS_WITHOUT_PATHS = [
http_code
for http_code in exceptions.default_exceptions
if http_code not in HTTP_ERRORS_WITH_PATHS
]
MESSAGE_FOR_STATUS = {
http.HTTPStatus.NOT_FOUND.value: 'Resource not found.',
http.HTTPStatus.UNAUTHORIZED.value: 'Not authorized to access this resource.',
}
def get_message_for_status(http_code):
try:
http_status = http.HTTPStatus(http_code)
except ValueError:
http_status = http.HTTPStatus.INTERNAL_SERVER_ERROR
return MESSAGE_FOR_STATUS.get(http_status, http_status.description)
def error_message_from_http_code_with_resource(error):
error_message = {
'error': {
'status_code': error.code,
'message': get_message_for_status(error.code),
'resource': flask.request.url,
},
}
return flask.make_response(flask.jsonify(error_message), error.code)
def error_message_from_http_code(error):
error_message = {
'error': {
'status_code': error.code,
'message': get_message_for_status(error.code),
},
}
return flask.make_response(flask.jsonify(error_message), error.code)
def error_message_from_exception(error: Exception):
error_message = {
'error': {
'status_code': http.HTTPStatus.INTERNAL_SERVER_ERROR,
'message': f'{error}',
},
}
return flask.make_response(flask.jsonify(error_message), http.HTTPStatus.INTERNAL_SERVER_ERROR)
def attach_error_handlers_to_api_app(app):
for http_code in HTTP_ERRORS_WITH_PATHS:
app.register_error_handler(http_code, error_message_from_http_code_with_resource)
for http_code in HTTP_ERRORS_WITHOUT_PATHS:
app.register_error_handler(http_code, error_message_from_http_code)
app.register_error_handler(Exception, error_message_from_exception)

View File

@ -68,8 +68,9 @@ def test_serve_landing_page_view_returns_about_page(api_base_uri, test_page):
def test_serve_page_view_returns_404_when_no_content_found(api_base_uri, test_page):
expected_json = {
'error': {
'code': 404,
'message': 'Not Found'
'status_code': 404,
'message': 'Resource not found.',
'resource': f'{api_base_uri}/api/pages/missing',
}
}

View File

@ -42,6 +42,13 @@ def editor_user(db_engine):
)
@pytest.fixture(scope='module')
def non_existent_user():
return {
'username': 'does-not-exist',
}
SAMPLE_USERS_REQUEST = {
'admin': {},
'editor': {},
@ -82,6 +89,11 @@ def test_editor_user_fetch_by_admin():
pass
@bdd.scenario('user_management.feature', 'Admin user can not get an non-existent user')
def test_non_existent_user_fetch_by_admin():
pass
@mark.skip(reason="Test scenarios need work")
@bdd.scenario('user_management.feature', 'Any user can not get an existing admin user')
def test_admin_user_fetch_by_anyone():
@ -154,11 +166,12 @@ def test_editor_user_deletion_by_editor():
def jwt_token(user_role, api_base_uri, admin_user, editor_user):
# TODO: Improve selection of fixtures.
user_info = None
if user_role == models.UserRole.admin.name:
user_info = admin_user
elif user_role == models.UserRole.editor.name:
user_info = editor_user
else:
user_info = non_existent_user
jwt_token = requests.post(
url=f'{api_base_uri}/auth',
@ -185,13 +198,14 @@ def create_user_response(user_role, jwt_token, api_base_uri):
@bdd.given(parsers.parse('I get an {user_role} user'))
def get_user_response(user_role, jwt_token, api_base_uri, admin_user, editor_user):
def get_user_response(user_role, jwt_token, api_base_uri, admin_user, editor_user, non_existent_user):
test_user = None
if user_role == models.UserRole.admin.name:
test_user = admin_user
elif user_role == models.UserRole.editor.name:
test_user = editor_user
else:
test_user = non_existent_user
response = requests.get(
url=f'{api_base_uri}/api/users/{test_user["username"]}',
@ -230,8 +244,21 @@ def assert_unauthorized_response(get_user_response: requests.Response):
expected_response_json = {
'status_code': http.HTTPStatus.UNAUTHORIZED,
'error': 'Unauthorized',
'description': 'Not authorized to access this resource.',
'message': 'Not authorized to access this resource.',
'resource': get_user_response.request.url,
}
assert get_user_response.json() == expected_response_json
@bdd.then(parsers.parse('I can get a user can not be found message'))
def assert_unauthorized_response(get_user_response: requests.Response):
assert get_user_response.status_code == http.HTTPStatus.NOT_FOUND
expected_response_json = {
'error': {
'status_code': http.HTTPStatus.NOT_FOUND.value,
'message': 'Resource not found.',
'resource': get_user_response.request.url,
},
}
assert get_user_response.json() == expected_response_json

View File

@ -34,6 +34,11 @@ Scenario: Admin user can get an existing editor user
And I get an editor user
Then I can get an editor user profile
Scenario: Admin user can not get an non-existent user
Given I am an admin user
And I get an non-existent user
Then I can get a user can not be found message
Scenario: Any user can not get an existing admin user
Given I am an unauthenticated user
And I get an admin user