278 lines
8.5 KiB
Python
278 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from typing import cast
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from delta_barth import errors
|
|
from delta_barth.constants import DEFAULT_API_ERR_CODE, DEFAULT_INTERNAL_ERR_CODE
|
|
from delta_barth.types import DelBarApiError, Status
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def status_hdlr() -> errors.StatusHandler:
|
|
return errors.StatusHandler()
|
|
|
|
|
|
def test_status_handler_parsing(status_hdlr):
|
|
predef_errs = errors.DATA_PIPELINE_STATUS_DESCR
|
|
|
|
assert status_hdlr.pipe_states is not None
|
|
parsed_pipe_states = status_hdlr.pipe_states
|
|
assert parsed_pipe_states.SUCCESS == status_hdlr.SUCCESS
|
|
parsed_pipe_states = asdict(parsed_pipe_states)
|
|
|
|
for err in predef_errs:
|
|
dopt_err = cast(Status, parsed_pipe_states[err[0]])
|
|
assert isinstance(dopt_err, Status)
|
|
assert dopt_err.code == err[1]
|
|
assert dopt_err.description == err[2]
|
|
assert dopt_err.message == ""
|
|
|
|
status_hdlr._parse_data_pipe_states()
|
|
|
|
|
|
def test_status_handler_InternalErrors_Success(status_hdlr):
|
|
DESCRIPTION = "test case"
|
|
MESSAGE = "an error occurred"
|
|
ERR_CODE = 101
|
|
|
|
new_err = status_hdlr.error(
|
|
description=DESCRIPTION,
|
|
message=MESSAGE,
|
|
code=ERR_CODE,
|
|
)
|
|
assert new_err.code == ERR_CODE
|
|
assert new_err.description == DESCRIPTION
|
|
assert new_err.message == MESSAGE
|
|
# failure cases
|
|
|
|
|
|
def test_status_handler_InternalErrors_CodeOutOfBounds(status_hdlr):
|
|
DESCRIPTION = "test case"
|
|
MESSAGE = "an error occurred"
|
|
|
|
err_code = DEFAULT_INTERNAL_ERR_CODE - 1
|
|
with pytest.raises(ValueError):
|
|
_ = status_hdlr.error(
|
|
description=DESCRIPTION,
|
|
message=MESSAGE,
|
|
code=err_code,
|
|
)
|
|
err_code = DEFAULT_API_ERR_CODE + 1
|
|
with pytest.raises(ValueError):
|
|
_ = status_hdlr.error(
|
|
description=DESCRIPTION,
|
|
message=MESSAGE,
|
|
code=err_code,
|
|
)
|
|
|
|
|
|
def test_status_handler_ApiError(status_hdlr):
|
|
MESSAGE = "an error occurred"
|
|
api_err = DelBarApiError(status_code=401, message="test case")
|
|
assert api_err.status_code == 401
|
|
assert api_err.message == "test case"
|
|
|
|
status_hdlr = errors.StatusHandler()
|
|
new_err = status_hdlr.api_error(error=api_err)
|
|
assert new_err.code == DEFAULT_API_ERR_CODE
|
|
assert "API-Server" in new_err.description
|
|
assert new_err.message != MESSAGE
|
|
assert new_err.api_server_error is not None
|
|
assert new_err.api_server_error == api_err
|
|
|
|
|
|
def test_status_handler_raise_for_status_Success(status_hdlr):
|
|
# success: should not raise
|
|
err_status = status_hdlr.SUCCESS
|
|
assert status_hdlr.raise_for_status(err_status) is None
|
|
|
|
|
|
def test_status_handler_raise_for_status_PredefinedErrors(status_hdlr):
|
|
# data related errors (predefined)
|
|
err_status = status_hdlr.pipe_states.BAD_QUALITY
|
|
err_descr = err_status.description
|
|
with pytest.raises(errors.UDataProcessingError):
|
|
try:
|
|
status_hdlr.raise_for_status(err_status)
|
|
except errors.UDataProcessingError as err:
|
|
descr = str(err)
|
|
assert err_descr in descr
|
|
raise err
|
|
|
|
|
|
def test_status_handler_raise_for_status_InternalErrors(status_hdlr):
|
|
# internal error, not data-related
|
|
description = "test case"
|
|
message = "an error occurred"
|
|
err_code = 101
|
|
err_status = status_hdlr.error(
|
|
description=description,
|
|
message=message,
|
|
code=err_code,
|
|
)
|
|
with pytest.raises(errors.UInternalError):
|
|
try:
|
|
status_hdlr.raise_for_status(err_status)
|
|
except errors.UInternalError as err:
|
|
descr = str(err)
|
|
assert description in descr
|
|
raise err
|
|
|
|
|
|
def test_status_handler_raise_for_status_ExternalApiErrors(status_hdlr):
|
|
# external API error
|
|
api_err = DelBarApiError(status_code=401, message="test case", code="1234")
|
|
description = "Kommunikation mit dem API-Server aufgetreten"
|
|
msg = "Bitte beachten Sie die"
|
|
err_status = status_hdlr.api_error(error=api_err)
|
|
with pytest.raises(errors.UApiError):
|
|
try:
|
|
status_hdlr.raise_for_status(err_status)
|
|
except errors.UApiError as err:
|
|
descr = str(err)
|
|
assert description in descr
|
|
assert msg in descr
|
|
raise err
|
|
|
|
|
|
@pytest.mark.new
|
|
def test_status_handler_exception_parsing():
|
|
status_hdlr = errors.StatusHandler()
|
|
test_message = "Test Exception"
|
|
test_code = 110
|
|
exception = ValueError(test_message)
|
|
status = status_hdlr.exception_to_error(exception, test_code)
|
|
assert status.code == test_code
|
|
assert status.description == exception.__class__.__name__
|
|
assert status.message == test_message
|
|
|
|
|
|
def test_NotSet():
|
|
not_set = errors.NotSet()
|
|
# using slots, dynamic attribute generation should not be possible
|
|
with pytest.raises(AttributeError):
|
|
not_set.test = "try to set value" # type: ignore
|
|
|
|
|
|
def test_result_wrapper_class_Success():
|
|
# successful operation
|
|
test_result = 10
|
|
error_code = 146
|
|
twrapper: errors.ResultWrapper[int] = errors.ResultWrapper(
|
|
result=test_result,
|
|
exception=None,
|
|
code_on_error=error_code,
|
|
)
|
|
assert twrapper.status == errors.STATUS_HANDLER.SUCCESS
|
|
assert twrapper.result == test_result
|
|
assert twrapper.unwrap() == test_result
|
|
assert twrapper.status.code != error_code
|
|
|
|
|
|
def test_result_wrapper_class_Failure():
|
|
# test for no result
|
|
error_code = 146
|
|
test_result = errors.NotSet()
|
|
test_message = "Test of error message"
|
|
exception = ValueError(test_message)
|
|
twrapper: errors.ResultWrapper[int] = errors.ResultWrapper(
|
|
result=test_result,
|
|
exception=exception,
|
|
code_on_error=error_code,
|
|
)
|
|
assert twrapper.status != errors.STATUS_HANDLER.SUCCESS
|
|
assert twrapper.status.code == error_code
|
|
with pytest.raises(errors.WAccessResultDespiteError):
|
|
twrapper.result
|
|
with pytest.raises(errors.UInternalError):
|
|
twrapper.unwrap()
|
|
|
|
|
|
def test_result_wrapper_class_WrongCodeSpans():
|
|
# test other error code spans
|
|
test_result = errors.NotSet()
|
|
test_message = "Test of error message"
|
|
exception = ValueError(test_message)
|
|
with pytest.raises(ValueError):
|
|
error_code = DEFAULT_INTERNAL_ERR_CODE - 1
|
|
_: errors.ResultWrapper[int] = errors.ResultWrapper(
|
|
result=test_result,
|
|
exception=exception,
|
|
code_on_error=error_code, # not possible because of lower bound
|
|
)
|
|
|
|
with pytest.raises(ValueError):
|
|
error_code = DEFAULT_API_ERR_CODE + 1
|
|
_: errors.ResultWrapper[int] = errors.ResultWrapper(
|
|
result=test_result,
|
|
exception=exception,
|
|
code_on_error=error_code, # not possible because of upper bound
|
|
)
|
|
|
|
|
|
def test_wrap_result_None():
|
|
MESSAGE = "Test case wrapped function decorator"
|
|
error_code = 103
|
|
|
|
@errors.wrap_result(error_code)
|
|
def test_func_1() -> None:
|
|
raise ValueError(MESSAGE)
|
|
|
|
res = test_func_1()
|
|
assert isinstance(res, errors.ResultWrapper)
|
|
assert res.status.code == error_code
|
|
assert res.status.description == "ValueError"
|
|
assert res.status.message == MESSAGE
|
|
assert res.status.api_server_error is None
|
|
with pytest.raises(errors.UInternalError):
|
|
res.unwrap()
|
|
|
|
|
|
def test_wrap_result_NotNone_Success():
|
|
error_code = 103
|
|
|
|
@errors.wrap_result(error_code)
|
|
def test_func_2(x: str, y: str) -> int:
|
|
return int(int(x) / int(y))
|
|
|
|
res = test_func_2("2", "1")
|
|
assert res.result == 2
|
|
assert res.unwrap() == 2
|
|
assert res.status == errors.STATUS_HANDLER.SUCCESS
|
|
|
|
|
|
def test_wrap_result_NotNone_FailureZeroDvisionError():
|
|
error_code = 103
|
|
|
|
@errors.wrap_result(error_code)
|
|
def test_func_2(x: str, y: str) -> int:
|
|
return int(int(x) / int(y))
|
|
|
|
res = test_func_2("2", "0")
|
|
with pytest.raises(errors.WAccessResultDespiteError):
|
|
res.result
|
|
with pytest.raises(errors.UInternalError):
|
|
res.unwrap()
|
|
assert res.status.code == error_code
|
|
assert res.status.description == "ZeroDivisionError"
|
|
|
|
|
|
def test_wrap_result_NotNone_FailureValueError():
|
|
error_code = 103
|
|
|
|
@errors.wrap_result(error_code)
|
|
def test_func_2(x: str, y: str) -> int:
|
|
return int(int(x) / int(y))
|
|
|
|
res = test_func_2("2", "test")
|
|
with pytest.raises(errors.WAccessResultDespiteError):
|
|
res.result
|
|
with pytest.raises(errors.UInternalError):
|
|
res.unwrap()
|
|
assert res.status.code == error_code
|
|
assert res.status.description == "ValueError"
|