delta-barth/delta-barth-py
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 3 Releases 3 Wiki Activity

add shortcut to success state

Browse Source
This commit is contained in:
Florian Förster 2025-03-05 15:26:39 +01:00
parent 763d3c1aac
commit 7b82d051e2
2 changed files with 21 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/delta_barth/errors.py
View File

@ -41,10 +41,17 @@ class StateHandler:
@property
def pipe_states(self) -> DataPipeStates:
assert self._pipe_states is not None, (
"tried to access not parsed data pipeline errors"
"tried to access not parsed data pipeline states"
)
return self._pipe_states
@property
def SUCCESS(self) -> Status:
assert self._pipe_states is not None, (
"tried to access not parsed data pipeline states"
)
return self.pipe_states.SUCCESS
def _parse_data_pipe_states(self) -> None:
if self._pipe_states is not None:
return

25
tests/test_errors.py
View File

@ -14,19 +14,20 @@ from delta_barth.types import DelBarApiError, Status
def test_state_handler_parsing():
predef_errs = errors.DATA_PIPELINE_STATUS_DESCR
err_hdlr = delta_barth._management.StateHandler()
assert err_hdlr.pipe_states is not None
parsed_pipe_errs = err_hdlr.pipe_states
parsed_pipe_errs = asdict(parsed_pipe_errs)
state_hdlr = delta_barth._management.StateHandler()
assert state_hdlr.pipe_states is not None
parsed_pipe_states = state_hdlr.pipe_states
assert parsed_pipe_states.SUCCESS == state_hdlr.SUCCESS
parsed_pipe_states = asdict(parsed_pipe_states)
for err in predef_errs:
dopt_err = cast(Status, parsed_pipe_errs[err[0]])
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 == ""
err_hdlr._parse_data_pipe_states()
state_hdlr._parse_data_pipe_states()
def test_state_handler_internal():
@ -34,8 +35,8 @@ def test_state_handler_internal():
MESSAGE = "an error occurred"
ERR_CODE = 101
err_hdlr = delta_barth._management.StateHandler()
new_err = err_hdlr.error(
state_hdlr = delta_barth._management.StateHandler()
new_err = state_hdlr.error(
description=DESCRIPTION,
message=MESSAGE,
code=ERR_CODE,
@ -46,14 +47,14 @@ def test_state_handler_internal():
# failure cases
err_code = 50 # default lower bound: 100
with pytest.raises(ValueError):
new_err = err_hdlr.error(
new_err = state_hdlr.error(
description=DESCRIPTION,
message=MESSAGE,
code=err_code,
)
err_code = 500 # default upper bound: 400
with pytest.raises(ValueError):
new_err = err_hdlr.error(
new_err = state_hdlr.error(
description=DESCRIPTION,
message=MESSAGE,
code=err_code,
@ -66,8 +67,8 @@ def test_state_handler_api_error():
assert api_err.status_code == 401
assert api_err.message == "test case"
err_hdlr = delta_barth._management.StateHandler()
new_err = err_hdlr.api_error(error=api_err)
state_hdlr = delta_barth._management.StateHandler()
new_err = state_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