diff --git a/pantos/common/restapi.py b/pantos/common/restapi.py index aaf2920..f7842d3 100644 --- a/pantos/common/restapi.py +++ b/pantos/common/restapi.py @@ -2,7 +2,6 @@ """ import json -import typing import flask import flask_restful # type: ignore @@ -17,28 +16,40 @@ 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. @@ -46,7 +57,7 @@ def conflict(error_message: str): Parameters ---------- error_message : str - The message of the error. + The error message. Raises ------ @@ -57,15 +68,13 @@ 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 ------ @@ -73,18 +82,16 @@ def not_acceptable(error_message: typing.Union[typing.List[typing.Any], 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 ------ @@ -92,7 +99,24 @@ def bad_request(error_message: typing.Union[typing.List[typing.Any], 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): @@ -101,7 +125,7 @@ def resource_not_found(error_message: str): Parameters ---------- error_message : str - The message of the error. + The error message. Raises ------ @@ -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 ------ diff --git a/tests/test_restapi.py b/tests/test_restapi.py new file mode 100644 index 0000000..748c11a --- /dev/null +++ b/tests/test_restapi.py @@ -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)