tests for result wrapping
This commit is contained in:
parent
adbc894899
commit
2262cfe45f
@ -119,7 +119,7 @@ class StatusHandler:
|
|||||||
upper_bound = DEFAULT_API_ERR_CODE
|
upper_bound = DEFAULT_API_ERR_CODE
|
||||||
if code < lower_bound or code > upper_bound:
|
if code < lower_bound or code > upper_bound:
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
f"Custom error codes mus be between default "
|
f"Custom error codes must be between default "
|
||||||
f"values of {lower_bound}-{upper_bound}"
|
f"values of {lower_bound}-{upper_bound}"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -144,7 +144,7 @@ class StatusHandler:
|
|||||||
api_server_error=error,
|
api_server_error=error,
|
||||||
)
|
)
|
||||||
|
|
||||||
def unwrap(
|
def raise_for_status(
|
||||||
self,
|
self,
|
||||||
state: Status,
|
state: Status,
|
||||||
) -> None:
|
) -> None:
|
||||||
@ -191,6 +191,10 @@ class ResultWrapper(t.Generic[T]):
|
|||||||
exception: Exception | None,
|
exception: Exception | None,
|
||||||
code_on_error: int,
|
code_on_error: int,
|
||||||
) -> None:
|
) -> None:
|
||||||
|
assert (isinstance(result, NotSet) and exception is not None) or (
|
||||||
|
not isinstance(result, NotSet) and exception is None
|
||||||
|
), "set >NotSet< without exception or result with exception"
|
||||||
|
|
||||||
self._result = result
|
self._result = result
|
||||||
status: Status = STATUS_HANDLER.SUCCESS
|
status: Status = STATUS_HANDLER.SUCCESS
|
||||||
if exception is not None:
|
if exception is not None:
|
||||||
@ -211,6 +215,11 @@ class ResultWrapper(t.Generic[T]):
|
|||||||
def __repr__(self) -> str:
|
def __repr__(self) -> str:
|
||||||
return self.__str__()
|
return self.__str__()
|
||||||
|
|
||||||
|
def unwrap(self) -> T:
|
||||||
|
STATUS_HANDLER.raise_for_status(self.status)
|
||||||
|
|
||||||
|
return self.result
|
||||||
|
|
||||||
|
|
||||||
def wrap_result(
|
def wrap_result(
|
||||||
code_on_error: int = DEFAULT_API_ERR_CODE,
|
code_on_error: int = DEFAULT_API_ERR_CODE,
|
||||||
|
|||||||
@ -145,6 +145,7 @@ def test_login_logout(session, credentials):
|
|||||||
assert status.api_server_error.message == "Nutzer oder Passwort falsch."
|
assert status.api_server_error.message == "Nutzer oder Passwort falsch."
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.api_con_required
|
||||||
def test_assert_login_while_logged_out(session):
|
def test_assert_login_while_logged_out(session):
|
||||||
assert session.session_token is None
|
assert session.session_token is None
|
||||||
assert session._creds is not None
|
assert session._creds is not None
|
||||||
|
|||||||
@ -7,7 +7,7 @@ import pytest
|
|||||||
from pydantic import ValidationError
|
from pydantic import ValidationError
|
||||||
|
|
||||||
from delta_barth import errors
|
from delta_barth import errors
|
||||||
from delta_barth.constants import DEFAULT_API_ERR_CODE
|
from delta_barth.constants import DEFAULT_API_ERR_CODE, DEFAULT_INTERNAL_ERR_CODE
|
||||||
from delta_barth.types import DelBarApiError, Status
|
from delta_barth.types import DelBarApiError, Status
|
||||||
|
|
||||||
|
|
||||||
@ -76,18 +76,18 @@ def test_status_handler_api_error():
|
|||||||
assert new_err.api_server_error == api_err
|
assert new_err.api_server_error == api_err
|
||||||
|
|
||||||
|
|
||||||
def test_status_handler_unwrap():
|
def test_status_handler_raise_for_status():
|
||||||
status_hdlr = errors.StatusHandler()
|
status_hdlr = errors.StatusHandler()
|
||||||
|
|
||||||
# success: should not raise
|
# success: should not raise
|
||||||
err_status = status_hdlr.SUCCESS
|
err_status = status_hdlr.SUCCESS
|
||||||
assert status_hdlr.unwrap(err_status) is None
|
assert status_hdlr.raise_for_status(err_status) is None
|
||||||
# data related errors (predefined)
|
# data related errors (predefined)
|
||||||
err_status = status_hdlr.pipe_states.BAD_QUALITY
|
err_status = status_hdlr.pipe_states.BAD_QUALITY
|
||||||
err_descr = err_status.description
|
err_descr = err_status.description
|
||||||
with pytest.raises(errors.UDataProcessingError):
|
with pytest.raises(errors.UDataProcessingError):
|
||||||
try:
|
try:
|
||||||
status_hdlr.unwrap(err_status)
|
status_hdlr.raise_for_status(err_status)
|
||||||
except errors.UDataProcessingError as err:
|
except errors.UDataProcessingError as err:
|
||||||
descr = str(err)
|
descr = str(err)
|
||||||
assert err_descr in descr
|
assert err_descr in descr
|
||||||
@ -103,7 +103,7 @@ def test_status_handler_unwrap():
|
|||||||
)
|
)
|
||||||
with pytest.raises(errors.UInternalError):
|
with pytest.raises(errors.UInternalError):
|
||||||
try:
|
try:
|
||||||
status_hdlr.unwrap(err_status)
|
status_hdlr.raise_for_status(err_status)
|
||||||
except errors.UInternalError as err:
|
except errors.UInternalError as err:
|
||||||
descr = str(err)
|
descr = str(err)
|
||||||
assert description in descr
|
assert description in descr
|
||||||
@ -115,7 +115,7 @@ def test_status_handler_unwrap():
|
|||||||
err_status = status_hdlr.api_error(error=api_err)
|
err_status = status_hdlr.api_error(error=api_err)
|
||||||
with pytest.raises(errors.UApiError):
|
with pytest.raises(errors.UApiError):
|
||||||
try:
|
try:
|
||||||
status_hdlr.unwrap(err_status)
|
status_hdlr.raise_for_status(err_status)
|
||||||
except errors.UApiError as err:
|
except errors.UApiError as err:
|
||||||
descr = str(err)
|
descr = str(err)
|
||||||
assert description in descr
|
assert description in descr
|
||||||
@ -151,6 +151,7 @@ def test_result_wrapper_class():
|
|||||||
)
|
)
|
||||||
assert twrapper.status == errors.STATUS_HANDLER.SUCCESS
|
assert twrapper.status == errors.STATUS_HANDLER.SUCCESS
|
||||||
assert twrapper.result == test_result
|
assert twrapper.result == test_result
|
||||||
|
assert twrapper.unwrap() == test_result
|
||||||
assert twrapper.status.code != error_code
|
assert twrapper.status.code != error_code
|
||||||
# test for no result
|
# test for no result
|
||||||
test_result = errors.NotSet()
|
test_result = errors.NotSet()
|
||||||
@ -165,3 +166,65 @@ def test_result_wrapper_class():
|
|||||||
assert twrapper.status.code == error_code
|
assert twrapper.status.code == error_code
|
||||||
with pytest.raises(errors.WAccessResultDespiteError):
|
with pytest.raises(errors.WAccessResultDespiteError):
|
||||||
twrapper.result
|
twrapper.result
|
||||||
|
with pytest.raises(errors.UInternalError):
|
||||||
|
twrapper.unwrap()
|
||||||
|
# test other error code spans
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
error_code = DEFAULT_INTERNAL_ERR_CODE - 1
|
||||||
|
twrapper: errors.ResultWrapper[int] = errors.ResultWrapper(
|
||||||
|
result=test_result,
|
||||||
|
exception=exception,
|
||||||
|
code_on_error=52, # not possible because of lower bound
|
||||||
|
)
|
||||||
|
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
error_code = DEFAULT_API_ERR_CODE + 1
|
||||||
|
twrapper: errors.ResultWrapper[int] = errors.ResultWrapper(
|
||||||
|
result=test_result,
|
||||||
|
exception=exception,
|
||||||
|
code_on_error=402, # not possible because of upper bound
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_wrap_result():
|
||||||
|
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()
|
||||||
|
|
||||||
|
@errors.wrap_result(error_code)
|
||||||
|
def test_func_2(x: str, y: str) -> int:
|
||||||
|
return int(int(x) / int(y))
|
||||||
|
|
||||||
|
# success
|
||||||
|
res = test_func_2("2", "1")
|
||||||
|
assert res.result == 2
|
||||||
|
assert res.unwrap() == 2
|
||||||
|
assert res.status == errors.STATUS_HANDLER.SUCCESS
|
||||||
|
# failure 1
|
||||||
|
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"
|
||||||
|
# failure 2
|
||||||
|
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"
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user