Skip to content

Commit

Permalink
feat: additional Flask responses and errors (#4)
Browse files Browse the repository at this point in the history
  • Loading branch information
markuslevonyak authored Apr 30, 2024
1 parent 3f7d200 commit ba0c253
Show file tree
Hide file tree
Showing 2 changed files with 187 additions and 24 deletions.
72 changes: 48 additions & 24 deletions pantos/common/restapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
"""
import json
import typing

import flask
import flask_restful # type: ignore
Expand All @@ -17,36 +16,48 @@ def get(self):
and running.
"""
return None
return None # pragma: no cover


def ok_response(
data: typing.Union[typing.Dict, typing.List]) -> flask.Response:
def ok_response(data: list | dict) -> flask.Response:
"""Create a Flask response given some data.
Parameters
----------
data : dict or list
data : list or dict
The data that will be wrapped in a Flask response.
Returns
-------
flask.Resposnse
The response object that is used by Flask.
flask.Response
The Flask response object.
"""
return flask.Response(json.dumps(data), status=200,
mimetype='application/json')


def no_content_response() -> flask.Response:
"""Create a Flask response for a successful request without any
content.
Returns
-------
flask.Response
The Flask response object.
"""
return flask.Response(status=204)


def conflict(error_message: str):
"""Raise an HTTPException when a request conflicts
with the current state of the server.
Parameters
----------
error_message : str
The message of the error.
The error message.
Raises
------
Expand All @@ -57,42 +68,55 @@ def conflict(error_message: str):
flask_restful.abort(409, message=error_message)


def not_acceptable(error_message: typing.Union[typing.List[typing.Any],
typing.Dict[typing.Any,
typing.Any]]):
def not_acceptable(error_messages: str | list | dict):
"""Raise an HTTPException for non-acceptable requests.
Parameters
----------
error_message : str
The message of the error.
error_messages : str or list or dict
The error messages.
Raises
------
HTTPException
HTTP exception raised with the code 406.
"""
flask_restful.abort(406, message=error_message)
flask_restful.abort(406, message=error_messages)


def bad_request(error_message: typing.Union[typing.List[typing.Any],
typing.Dict[typing.Any,
typing.Any]]):
def bad_request(error_messages: str | list | dict):
"""Raise an HTTPException for bad requests.
Parameters
----------
error_message : str
The message of the error.
error_messages : str or list or dict
The error messages.
Raises
------
HTTPException
HTTP exception raised with the code 400.
"""
flask_restful.abort(400, message=error_message)
flask_restful.abort(400, message=error_messages)


def forbidden(error_message: str):
"""Raise an HTTPException if a prohibited action is refused.
Parameters
----------
error_message : str
The error message.
Raises
------
HTTPException
HTTP exception raised with the code 403.
"""
flask_restful.abort(403, message=error_message)


def resource_not_found(error_message: str):
Expand All @@ -101,7 +125,7 @@ def resource_not_found(error_message: str):
Parameters
----------
error_message : str
The message of the error.
The error message.
Raises
------
Expand All @@ -112,13 +136,13 @@ def resource_not_found(error_message: str):
flask_restful.abort(404, message=error_message)


def internal_server_error(error_message: typing.Optional[str] = None):
def internal_server_error(error_message: str | None = None):
"""Raise an HTTPException for internal server errors.
Parameters
----------
error_message : str
The message of the error (optional).
error_message : str, optional
The error message.
Raises
------
Expand Down
139 changes: 139 additions & 0 deletions tests/test_restapi.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
import json

import pytest
import werkzeug.exceptions

from pantos.common.restapi import bad_request
from pantos.common.restapi import conflict
from pantos.common.restapi import forbidden
from pantos.common.restapi import internal_server_error
from pantos.common.restapi import no_content_response
from pantos.common.restapi import not_acceptable
from pantos.common.restapi import ok_response
from pantos.common.restapi import resource_not_found


@pytest.fixture
def error_message():
return 'error message'


@pytest.fixture
def error_message_list():
return [
'first error message', 'second error message', 'third error message'
]


@pytest.fixture
def error_message_dict():
return {
'first_property': 'first error message',
'second_property': 'second error message',
'third_property': 'third error message'
}


def test_ok_response():
data = {
'first_property': 1,
'second_propery': 'a',
'third_property': True,
'fourth_property': 1.01
}

response = ok_response(data)

assert response.status_code == 200
assert response.mimetype == 'application/json'
assert json.loads(response.data) == data


def test_no_content_response():
response = no_content_response()

assert response.status_code == 204
assert len(response.data) == 0


def test_bad_request_str(error_message):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
bad_request(error_message)

assert isinstance(exception_info.value, werkzeug.exceptions.BadRequest)
assert exception_info.value.data['message'] == error_message


def test_bad_request_list(error_message_list):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
bad_request(error_message_list)

assert isinstance(exception_info.value, werkzeug.exceptions.BadRequest)
assert exception_info.value.data['message'] == error_message_list


def test_bad_request_dict(error_message_dict):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
bad_request(error_message_dict)

assert isinstance(exception_info.value, werkzeug.exceptions.BadRequest)
assert exception_info.value.data['message'] == error_message_dict


def test_forbidden(error_message):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
forbidden(error_message)

assert isinstance(exception_info.value, werkzeug.exceptions.Forbidden)
assert exception_info.value.data['message'] == error_message


def test_resource_not_found(error_message):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
resource_not_found(error_message)

assert isinstance(exception_info.value, werkzeug.exceptions.NotFound)
assert exception_info.value.data['message'] == error_message


def test_not_acceptable_str(error_message):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
not_acceptable(error_message)

assert isinstance(exception_info.value, werkzeug.exceptions.NotAcceptable)
assert exception_info.value.data['message'] == error_message


def test_not_acceptable_list(error_message_list):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
not_acceptable(error_message_list)

assert isinstance(exception_info.value, werkzeug.exceptions.NotAcceptable)
assert exception_info.value.data['message'] == error_message_list


def test_not_acceptable_dict(error_message_dict):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
not_acceptable(error_message_dict)

assert isinstance(exception_info.value, werkzeug.exceptions.NotAcceptable)
assert exception_info.value.data['message'] == error_message_dict


def test_conflict(error_message):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
conflict(error_message)

assert isinstance(exception_info.value, werkzeug.exceptions.Conflict)
assert exception_info.value.data['message'] == error_message


@pytest.mark.parametrize('with_error_message', [True, False])
def test_internal_server_error(with_error_message, error_message):
with pytest.raises(werkzeug.exceptions.HTTPException) as exception_info:
internal_server_error(error_message if with_error_message else None)

assert isinstance(exception_info.value,
werkzeug.exceptions.InternalServerError)
assert exception_info.value.data['message'] == (
error_message if with_error_message else None)

0 comments on commit ba0c253

Please sign in to comment.