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

major refactoring

Browse Source
This commit is contained in:
Florian Förster 2025-03-07 12:23:59 +01:00
parent a94fd4a936
commit 76f58ecba0
13 changed files with 526 additions and 286 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/delta_barth/_management.py
View File

@ -1,8 +1,3 @@
from typing import Final
from delta_barth.errors import STATE_HANDLER
from delta_barth.api._session import Session
from delta_barth.constants import HTTP_BASE_CONTENT_HEADERS
from delta_barth.errors import StateHandler
STATE_HANDLER: Final[StateHandler] = StateHandler()
CURRENT_SESSION: Final[Session] = Session(HTTP_BASE_CONTENT_HEADERS)
__all__ = ["STATE_HANDLER"]

4
src/delta_barth/analysis/forecast.py
View File

@ -8,13 +8,13 @@ import pandas as pd
from sklearn.metrics import mean_squared_error
from xgboost import XGBRegressor
from delta_barth._management import STATE_HANDLER
from delta_barth.analysis import parse
from delta_barth.constants import COL_MAP_SALES_PROGNOSIS, FEATURES_SALES_PROGNOSIS
from delta_barth.errors import STATE_HANDLER
from delta_barth.types import CustomerDataSalesForecast, DataPipeStates, PipeResult
if TYPE_CHECKING:
from delta_barth.api.common import SalesPrognosisResponse
from delta_barth.api.requests import SalesPrognosisResponse
# TODO check pandera for DataFrame validation

48
src/delta_barth/api/_session.py
View File

@ -1,48 +0,0 @@
from __future__ import annotations
import warnings
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from delta_barth.types import HttpContentHeaders
class Session:
def __init__(
self,
base_headers: HttpContentHeaders,
) -> None:
self._headers = base_headers
self._session_token: str | None = None
self._logged_in: bool = False
@property
def headers(self) -> HttpContentHeaders:
return self._headers
@property
def session_token(self) -> str | None:
return self._session_token
@property
def logged_in(self) -> bool:
return self._logged_in
def add_session_token(
self,
token: str,
) -> None:
if self.session_token is not None:
warnings.warn(
"Setting new API session token despite it was already set. "
"Overwriting existing token."
)
self._session_token = token
self._headers.update(DelecoToken=token)
self._logged_in = True
def remove_session_token(self) -> None:
if "DelecoToken" in self.headers:
del self._headers["DelecoToken"]
self._session_token = None
self._logged_in = False

354
src/delta_barth/api/common.py
View File

@ -1,29 +1,208 @@
from __future__ import annotations
import re
from datetime import datetime as Datetime
from typing import TYPE_CHECKING, Final
import requests
from pydantic import BaseModel, PositiveInt, SkipValidation
from pydantic import BaseModel
from requests import Response
from delta_barth._management import CURRENT_SESSION, STATE_HANDLER
from delta_barth.constants import HTTP_BASE_CONTENT_HEADERS
from delta_barth.errors import (
ApiConnectionError,
STATE_HANDLER,
UnspecifiedRequestType,
)
from delta_barth.types import DelBarApiError, HttpRequestTypes
from delta_barth.types import (
ApiCredentials,
DelBarApiError,
HttpRequestTypes,
)
if TYPE_CHECKING:
from delta_barth.types import Status
LOGIN_ERROR_CODES_KNOWN: Final[frozenset[int]] = frozenset((400, 401, 409, 500))
from delta_barth.types import HttpContentHeaders, Status
def _assert_login_status() -> None:
if not CURRENT_SESSION.logged_in:
raise ApiConnectionError("Curent session is not logged in")
class Session:
def __init__(
self,
base_headers: HttpContentHeaders,
) -> None:
self._creds: ApiCredentials | None = None
self._base_url: str | None = None
self._headers = base_headers
self._session_token: str | None = None
self._logged_in: bool = False
@property
def creds(self) -> ApiCredentials:
assert self._creds is not None, "accessed credentials not set"
return self._creds
def set_credentials(
self,
user_name: str,
password: str,
database: str,
mandant: str,
) -> None:
self._creds = validate_credentials(
user_name=user_name,
password=password,
database=database,
mandant=mandant,
)
@property
def base_url(self) -> str:
assert self._base_url is not None, "accessed base URL not set"
return self._base_url
def set_base_url(
self,
base_url: str,
) -> None:
self._base_url = base_url
@property
def headers(self) -> HttpContentHeaders:
return self._headers
@property
def session_token(self) -> str | None:
return self._session_token
@property
def logged_in(self) -> bool:
return self._logged_in
def _add_session_token(
self,
token: str,
) -> None:
assert self.session_token is None, "tried overwriting existing API session token"
self._session_token = token
self._headers.update(DelecoToken=token)
self._logged_in = True
def _remove_session_token(self) -> None:
assert self.session_token is not None, (
"tried to delete non-existing API session token"
)
if "DelecoToken" in self.headers:
del self._headers["DelecoToken"]
self._session_token = None
self._logged_in = False
def login(
self,
) -> tuple[LoginResponse, Status]:
ROUTE: Final[str] = "user/login"
URL: Final = combine_route(self.base_url, ROUTE)
login_req = LoginRequest(
userName=self.creds.user_name,
password=self.creds.password,
databaseName=self.creds.database,
mandantName=self.creds.mandant,
)
resp = requests.put(
URL,
login_req.model_dump_json(),
headers=self.headers, # type: ignore
)
response: LoginResponse
status: Status
if resp.status_code == 200:
response = LoginResponse(**resp.json())
status = STATE_HANDLER.pipe_states.SUCCESS
self._add_session_token(response.token)
else:
response = LoginResponse(token="")
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATE_HANDLER.api_error(err)
return response, status
def logout(
self,
) -> tuple[None, Status]:
ROUTE: Final[str] = "user/logout"
URL: Final = combine_route(self.base_url, ROUTE)
resp = requests.put(
URL,
headers=self.headers, # type: ignore
)
response = None
status: Status
if resp.status_code == 200:
status = STATE_HANDLER.SUCCESS
self._remove_session_token()
else:
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATE_HANDLER.api_error(err)
return response, status
def assert_login(
self,
) -> tuple[LoginResponse, Status]:
# check if login token is still valid
# re-login if necessary
if self.session_token is None:
return self.login()
# use known endpoint which requires a valid token in its header
# evaluate the response to decide if:
# current token is still valid, token is not valid, other errors occurred
ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
URL: Final = combine_route(self.base_url, ROUTE)
params: dict[str, int] = {"FirmaId": 999999}
resp = requests.get(
URL,
params=params,
headers=self.headers, # type: ignore
)
response: LoginResponse
status: Status
if resp.status_code == 200:
# TODO use to check for catching unknown exceptions
# response = LoginResponse(**resp.json())
response = LoginResponse(token=self.session_token)
status = STATE_HANDLER.SUCCESS
elif resp.status_code == 401:
self._remove_session_token()
response, status = self.login()
else: # pragma: no cover
response = LoginResponse(token="")
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATE_HANDLER.api_error(err)
return response, status
def validate_credentials(
user_name: str,
password: str,
database: str,
mandant: str,
) -> ApiCredentials:
return ApiCredentials(
user_name=user_name,
password=password,
database=database,
mandant=mandant,
)
# def _assert_login_status() -> None:
# if not CURRENT_SESSION.logged_in:
# raise ApiConnectionError("Curent session is not logged in")
# assert login:
# doing request to defined end point which
def _strip_url_components(string: str) -> str:
@ -68,118 +247,61 @@ class LoginResponse(BaseModel):
token: str
def login(
base_url: str,
user_name: str,
password: str,
database: str,
mandant: str,
) -> tuple[LoginResponse, Status]:
ROUTE: Final[str] = "user/login"
URL: Final = combine_route(base_url, ROUTE)
# def login(
# base_url: str,
# user_name: str,
# password: str,
# database: str,
# mandant: str,
# ) -> tuple[LoginResponse, Status]:
# ROUTE: Final[str] = "user/login"
# URL: Final = combine_route(base_url, ROUTE)
login_req = LoginRequest(
userName=user_name,
password=password,
databaseName=database,
mandantName=mandant,
)
resp = requests.put(
URL,
login_req.model_dump_json(),
headers=CURRENT_SESSION.headers, # type: ignore
)
# login_req = LoginRequest(
# userName=user_name,
# password=password,
# databaseName=database,
# mandantName=mandant,
# )
# resp = requests.put(
# URL,
# login_req.model_dump_json(),
# headers=CURRENT_SESSION.headers, # type: ignore
# )
response: LoginResponse
status: Status
if resp.status_code == 200:
response = LoginResponse(**resp.json())
status = STATE_HANDLER.pipe_states.SUCCESS
CURRENT_SESSION.add_session_token(response.token)
else:
response = LoginResponse(token="")
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATE_HANDLER.api_error(err)
# response: LoginResponse
# status: Status
# if resp.status_code == 200:
# response = LoginResponse(**resp.json())
# status = STATE_HANDLER.pipe_states.SUCCESS
# CURRENT_SESSION.add_session_token(response.token)
# else:
# response = LoginResponse(token="")
# err = DelBarApiError(status_code=resp.status_code, **resp.json())
# status = STATE_HANDLER.api_error(err)
return response, status
# return response, status
# ** logout
def logout(
base_url: str,
) -> tuple[None, Status]:
ROUTE: Final[str] = "user/logout"
URL: Final = combine_route(base_url, ROUTE)
# def logout(
# base_url: str,
# ) -> tuple[None, Status]:
# ROUTE: Final[str] = "user/logout"
# URL: Final = combine_route(base_url, ROUTE)
resp = requests.put(
URL,
headers=CURRENT_SESSION.headers, # type: ignore
)
# resp = requests.put(
# URL,
# headers=CURRENT_SESSION.headers, # type: ignore
# )
response = None
status: Status
if resp.status_code == 200:
status = STATE_HANDLER.SUCCESS
CURRENT_SESSION.remove_session_token()
else:
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATE_HANDLER.api_error(err)
return response, status
# ** sales data
class SalesPrognosisRequestP(BaseModel):
FirmaId: SkipValidation[int | None]
BuchungsDatum: SkipValidation[Datetime | None]
class SalesPrognosisResponseEntry(BaseModel):
artikelId: PositiveInt
firmaId: PositiveInt
betrag: float # negative values are filtered out later
menge: float # reasons for negative values unknown
buchungsDatum: Datetime
class SalesPrognosisResponse(BaseModel):
daten: tuple[SalesPrognosisResponseEntry, ...]
# error: DelBarApiError | None = None
def get_sales_prognosis_data(
base_url: str,
company_id: int | None = None,
start_date: Datetime | None = None,
) -> tuple[SalesPrognosisResponse, Status]:
_assert_login_status()
ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
URL: Final = combine_route(base_url, ROUTE)
sales_prog_req = SalesPrognosisRequestP(
FirmaId=company_id,
BuchungsDatum=start_date,
)
resp = requests.get(
URL,
params=sales_prog_req.model_dump(mode="json", exclude_none=True),
headers=CURRENT_SESSION.headers, # type: ignore[argumentType]
)
response: SalesPrognosisResponse
status: Status
if resp.status_code == 200:
response = SalesPrognosisResponse(**resp.json())
status = STATE_HANDLER.SUCCESS
else:
response = SalesPrognosisResponse(daten=tuple())
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATE_HANDLER.api_error(err)
# elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value: # pragma: no cover
# response = None
# status: Status
# if resp.status_code == 200:
# status = STATE_HANDLER.SUCCESS
# CURRENT_SESSION.remove_session_token()
# else:
# err = DelBarApiError(status_code=resp.status_code, **resp.json())
# response = SalesPrognosisResponse(daten=tuple(), error=err)
# else: # pragma: no cover
# _raise_for_unknown_error(resp)
# status = STATE_HANDLER.api_error(err)
return response, status
# return response, status

64
src/delta_barth/api/requests.py Normal file
View File

@ -0,0 +1,64 @@
from __future__ import annotations
from datetime import datetime as Datetime
from typing import TYPE_CHECKING, Final
import requests
from pydantic import BaseModel, PositiveInt, SkipValidation
from delta_barth.api.common import combine_route
from delta_barth.errors import STATE_HANDLER
from delta_barth.types import DelBarApiError, Status
if TYPE_CHECKING:
from delta_barth.api.common import Session
# ** sales data
class SalesPrognosisRequestP(BaseModel):
FirmaId: SkipValidation[int | None]
BuchungsDatum: SkipValidation[Datetime | None]
class SalesPrognosisResponseEntry(BaseModel):
artikelId: PositiveInt
firmaId: PositiveInt
betrag: float # negative values are filtered out later
menge: float # reasons for negative values unknown
buchungsDatum: Datetime
class SalesPrognosisResponse(BaseModel):
daten: tuple[SalesPrognosisResponseEntry, ...]
def get_sales_prognosis_data(
session: Session,
company_id: int | None = None,
start_date: Datetime | None = None,
) -> tuple[SalesPrognosisResponse, Status]:
session.assert_login()
ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
URL: Final = combine_route(session.base_url, ROUTE)
sales_prog_req = SalesPrognosisRequestP(
FirmaId=company_id,
BuchungsDatum=start_date,
)
resp = requests.get(
URL,
params=sales_prog_req.model_dump(mode="json", exclude_none=True),
headers=session.headers, # type: ignore[argumentType]
)
response: SalesPrognosisResponse
status: Status
if resp.status_code == 200:
response = SalesPrognosisResponse(**resp.json())
status = STATE_HANDLER.SUCCESS
else:
response = SalesPrognosisResponse(daten=tuple())
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATE_HANDLER.api_error(err)
return response, status

1
src/delta_barth/constants.py
View File

@ -7,6 +7,7 @@ from delta_barth.types import HttpContentHeaders
DEFAULT_INTERNAL_ERR_CODE: Final[int] = 100
DEFAULT_API_ERR_CODE: Final[int] = 400
HTTP_BASE_CONTENT_HEADERS: Final[HttpContentHeaders] = {
"Content-type": "application/json",
"Accept": "application/json",

5
src/delta_barth/errors.py
View File

@ -60,7 +60,7 @@ def _construct_exception(
return exception(err_message)
class StateHandler:
class StatusHandler:
def __init__(self) -> None:
self._pipe_states: DataPipeStates | None = None
self._parse_data_pipe_states()
@ -145,3 +145,6 @@ class StateHandler:
)
add_info = api_err.model_dump(exclude_none=True)
raise _construct_exception(UApiError, descr, msg, add_info)
STATE_HANDLER: Final[StatusHandler] = StatusHandler()

11
src/delta_barth/types.py
View File

@ -5,7 +5,7 @@ from dataclasses import dataclass, field
from typing import NotRequired, TypeAlias, TypedDict
import pandas as pd
from pydantic import BaseModel, SkipValidation
from pydantic import BaseModel, ConfigDict, SkipValidation
# ** Pipeline state management
StatusDescription: TypeAlias = tuple[str, int, str]
@ -32,6 +32,15 @@ class PipeResult:
# ** API
class ApiCredentials(BaseModel):
model_config: ConfigDict = ConfigDict(str_strip_whitespace=True)
user_name: str
password: str
database: str
mandant: str
class DelBarApiError(BaseModel):
status_code: int
message: str = ""

18
tests/api/conftest.py Normal file
View File

@ -0,0 +1,18 @@
import pytest
from delta_barth.api import common
from delta_barth.constants import HTTP_BASE_CONTENT_HEADERS
@pytest.fixture(scope="function")
def session(credentials, api_base_url) -> common.Session:
session = common.Session(HTTP_BASE_CONTENT_HEADERS)
session.set_base_url(api_base_url)
session.set_credentials(
user_name=credentials["user"],
password=credentials["pwd"],
database=credentials["db"],
mandant=credentials["mandant"],
)
return session

190
tests/api/test_common.py
View File

@ -1,18 +1,13 @@
from datetime import datetime as Datetime
import pytest
from pydantic import ValidationError
from delta_barth._management import CURRENT_SESSION
from delta_barth.api import common
from delta_barth.constants import DEFAULT_API_ERR_CODE
from delta_barth.constants import DEFAULT_API_ERR_CODE, HTTP_BASE_CONTENT_HEADERS
from delta_barth.errors import (
ApiConnectionError,
UnspecifiedRequestType,
)
from delta_barth.types import HttpRequestTypes
"http://test.com/ "
@pytest.mark.parametrize(
["case", "expect"],
@ -50,9 +45,42 @@ def test_combine_route(base, route, expect):
assert res == expect
def test_assert_login():
with pytest.raises(ApiConnectionError):
common._assert_login_status()
def test_validate_creds(credentials):
creds = common.validate_credentials(
user_name=credentials["user"],
password=credentials["pwd"],
database=credentials["db"],
mandant=credentials["mandant"],
)
assert creds.user_name == credentials["user"]
assert creds.password == credentials["pwd"]
assert creds.database == credentials["db"]
assert creds.mandant == credentials["mandant"]
# with whitespaces
user = " " + credentials["user"] + " "
creds = common.validate_credentials(
user_name=user,
password=credentials["pwd"],
database=credentials["db"],
mandant=credentials["mandant"],
)
assert user != credentials["user"]
assert creds.user_name == credentials["user"]
# invalid type
user = 123
with pytest.raises(ValidationError):
creds = common.validate_credentials(
user_name=user, # type: ignore
password=credentials["pwd"],
database=credentials["db"],
mandant=credentials["mandant"],
)
# TODO removal
# def test_assert_login():
# with pytest.raises(ApiConnectionError):
# common._assert_login_status()
@pytest.mark.api_con_required
@ -68,89 +96,87 @@ def test_ping(api_base_url):
resp = common.ping(api_base_url, HttpRequestTypes.POST)
@pytest.mark.api_con_required
def test_login_logout(credentials, api_base_url):
assert CURRENT_SESSION.session_token is None
resp, state = common.login(
base_url=api_base_url,
def test_session(credentials, api_base_url):
session = common.Session(HTTP_BASE_CONTENT_HEADERS)
assert session.session_token is None
assert session._creds is None
assert session._base_url is None
session.set_base_url(api_base_url)
assert session._base_url is not None
session.set_credentials(
user_name=credentials["user"],
password=credentials["pwd"],
database=credentials["db"],
mandant=credentials["mandant"],
)
assert state.code == 0
assert CURRENT_SESSION.session_token is not None
resp, state = common.logout(
base_url=api_base_url,
)
assert resp is None
assert state.code == 0
assert CURRENT_SESSION.session_token is None
assert "DelecoToken" not in CURRENT_SESSION.headers
assert session._creds is not None
resp, state = common.login(
base_url=api_base_url,
assert session.session_token is None
assert not session.logged_in
@pytest.mark.api_con_required
def test_login_logout(session, credentials):
assert not session.logged_in
resp, status = session.login()
assert resp is not None
assert status.code == 0
assert session.session_token is not None
resp, status = session.logout()
assert resp is None
assert status.code == 0
assert session.session_token is None
assert "DelecoToken" not in session.headers
session.set_credentials(
user_name=credentials["user"],
password="WRONG_PASSWORD",
database=credentials["db"],
mandant=credentials["mandant"],
)
resp, status = session.login()
assert resp is not None
assert state.code == DEFAULT_API_ERR_CODE
assert state.api_server_error is not None
assert state.api_server_error.status_code == 409
assert state.api_server_error.message == "Nutzer oder Passwort falsch."
assert status.code == DEFAULT_API_ERR_CODE
assert status.api_server_error is not None
assert status.api_server_error.status_code == 409
assert status.api_server_error.message == "Nutzer oder Passwort falsch."
@pytest.mark.api_con_required
def test_get_sales_prognosis_data(credentials, api_base_url):
resp, state = common.login(
base_url=api_base_url,
user_name=credentials["user"],
password=credentials["pwd"],
database=credentials["db"],
mandant=credentials["mandant"],
)
# test without company ID
assert state.code == 0
date = Datetime(2022, 6, 1)
resp, state = common.get_sales_prognosis_data(api_base_url, None, date)
assert state.code == 0
assert len(resp.daten) > 0
date = Datetime(2030, 1, 1)
resp, state = common.get_sales_prognosis_data(api_base_url, None, date)
assert state.code == 0
assert len(resp.daten) == 0
# test with company ID
assert state.code == 0
date = Datetime(2022, 6, 1)
company_id = 1024
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, date)
assert state.code == 0
assert len(resp.daten) > 0
date = Datetime(2030, 1, 1)
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, date)
assert state.code == 0
assert len(resp.daten) == 0
# test with non-existent company ID
assert state.code == 0
date = Datetime(2022, 6, 1)
company_id = 1000024
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, date)
# TODO check if this behaviour is still considered "successful"
assert state.code == 0
assert len(resp.daten) == 0
# test without date
company_id = 1024
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, None)
assert state.code == 0
assert len(resp.daten) > 0
# test without filters
resp, state = common.get_sales_prognosis_data(api_base_url, None, None)
assert state.code == 0
assert len(resp.daten) > 0
# close connection
resp, state = common.logout(
base_url=api_base_url,
)
assert state.code == 0
def test_assert_login_while_logged_out(session):
assert session.session_token is None
assert session._creds is not None
# test logged out state
resp, status = session.assert_login()
assert resp is not None
assert status.code == 0
assert session.session_token is not None
resp, status = session.logout()
assert status.code == 0
# test already logged in
assert session.session_token is None
assert session._creds is not None
_, status = session.login()
assert status.code == 0
resp, status = session.assert_login()
assert resp is not None
assert status.code == 0
assert session.session_token is not None
resp, status = session.logout()
assert status.code == 0
# test invalid token
assert session.session_token is None
assert session._creds is not None
_, status = session.login()
assert status.code == 0
session._session_token = "WRONGTOKEN"
resp, status = session.assert_login()
assert resp is not None
assert status.code == 0
assert session.session_token is not None
resp, status = session.logout()
assert status.code == 0

51
tests/api/test_requests.py Normal file
View File

@ -0,0 +1,51 @@
from datetime import datetime as Datetime
import pytest
from delta_barth.api import requests as requests_
@pytest.mark.api_con_required
def test_get_sales_prognosis_data(session):
resp, state = session.login()
# test without company ID
assert state.code == 0
date = Datetime(2022, 6, 1)
resp, state = requests_.get_sales_prognosis_data(session, None, date)
assert state.code == 0
assert len(resp.daten) > 0
date = Datetime(2030, 1, 1)
resp, state = requests_.get_sales_prognosis_data(session, None, date)
assert state.code == 0
assert len(resp.daten) == 0
# test with company ID
assert state.code == 0
date = Datetime(2022, 6, 1)
company_id = 1024
resp, state = requests_.get_sales_prognosis_data(session, company_id, date)
assert state.code == 0
assert len(resp.daten) > 0
date = Datetime(2030, 1, 1)
resp, state = requests_.get_sales_prognosis_data(session, company_id, date)
assert state.code == 0
assert len(resp.daten) == 0
# test with non-existent company ID
assert state.code == 0
date = Datetime(2022, 6, 1)
company_id = 1000024
resp, state = requests_.get_sales_prognosis_data(session, company_id, date)
# TODO check if this behaviour is still considered "successful"
assert state.code == 0
assert len(resp.daten) == 0
# test without date
company_id = 1024
resp, state = requests_.get_sales_prognosis_data(session, company_id, None)
assert state.code == 0
assert len(resp.daten) > 0
# test without filters
resp, state = requests_.get_sales_prognosis_data(session, None, None)
assert state.code == 0
assert len(resp.daten) > 0
# close connection
resp, state = session.logout()
assert state.code == 0

2
tests/conftest.py
View File

@ -9,7 +9,7 @@ import pandas as pd
import pytest
if TYPE_CHECKING:
from delta_barth.api.common import SalesPrognosisResponse
from delta_barth.api.requests import SalesPrognosisResponse
@pytest.fixture(scope="session")

53
tests/test_errors.py
View File

@ -4,20 +4,20 @@ from dataclasses import asdict
from typing import cast
import pytest
from pydantic import ValidationError
import delta_barth._management
from delta_barth import errors
from delta_barth.constants import DEFAULT_API_ERR_CODE
from delta_barth.types import DelBarApiError, Status
def test_state_handler_parsing():
def test_status_handler_parsing():
predef_errs = errors.DATA_PIPELINE_STATUS_DESCR
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
status_hdlr = errors.StatusHandler()
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:
@ -27,16 +27,16 @@ def test_state_handler_parsing():
assert dopt_err.description == err[2]
assert dopt_err.message == ""
state_hdlr._parse_data_pipe_states()
status_hdlr._parse_data_pipe_states()
def test_state_handler_internal():
def test_status_handler_internal():
DESCRIPTION = "test case"
MESSAGE = "an error occurred"
ERR_CODE = 101
state_hdlr = delta_barth._management.StateHandler()
new_err = state_hdlr.error(
status_hdlr = errors.StatusHandler()
new_err = status_hdlr.error(
description=DESCRIPTION,
message=MESSAGE,
code=ERR_CODE,
@ -47,28 +47,28 @@ def test_state_handler_internal():
# failure cases
err_code = 50 # default lower bound: 100
with pytest.raises(ValueError):
new_err = state_hdlr.error(
new_err = status_hdlr.error(
description=DESCRIPTION,
message=MESSAGE,
code=err_code,
)
err_code = 500 # default upper bound: 400
with pytest.raises(ValueError):
new_err = state_hdlr.error(
new_err = status_hdlr.error(
description=DESCRIPTION,
message=MESSAGE,
code=err_code,
)
def test_state_handler_api_error():
def test_status_handler_api_error():
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"
state_hdlr = delta_barth._management.StateHandler()
new_err = state_hdlr.api_error(error=api_err)
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
@ -76,19 +76,18 @@ def test_state_handler_api_error():
assert new_err.api_server_error == api_err
@pytest.mark.new
def test_state_handler_raising():
state_hdlr = delta_barth._management.StateHandler()
def test_status_handler_raising():
status_hdlr = errors.StatusHandler()
# success: should not raise
err_state = state_hdlr.SUCCESS
assert state_hdlr.unwrap(err_state) is None
err_status = status_hdlr.SUCCESS
assert status_hdlr.unwrap(err_status) is None
# data related errors (predefined)
err_state = state_hdlr.pipe_states.BAD_QUALITY
err_descr = err_state.description
err_status = status_hdlr.pipe_states.BAD_QUALITY
err_descr = err_status.description
with pytest.raises(errors.UDataProcessingError):
try:
state_hdlr.unwrap(err_state)
status_hdlr.unwrap(err_status)
except errors.UDataProcessingError as err:
descr = str(err)
assert err_descr in descr
@ -97,14 +96,14 @@ def test_state_handler_raising():
description = "test case"
message = "an error occurred"
err_code = 101
err_state = state_hdlr.error(
err_status = status_hdlr.error(
description=description,
message=message,
code=err_code,
)
with pytest.raises(errors.UInternalError):
try:
state_hdlr.unwrap(err_state)
status_hdlr.unwrap(err_status)
except errors.UInternalError as err:
descr = str(err)
assert description in descr
@ -113,10 +112,10 @@ def test_state_handler_raising():
api_err = DelBarApiError(status_code=401, message="test case", code="1234")
description = "Kommunikation mit dem API-Server aufgetreten"
msg = "Bitte beachten Sie die"
err_state = state_hdlr.api_error(error=api_err)
err_status = status_hdlr.api_error(error=api_err)
with pytest.raises(errors.UApiError):
try:
state_hdlr.unwrap(err_state)
status_hdlr.unwrap(err_status)
except errors.UApiError as err:
descr = str(err)
assert description in descr