diff --git a/src/delta_barth/_management.py b/src/delta_barth/_management.py index 628fd3a..34a8f32 100644 --- a/src/delta_barth/_management.py +++ b/src/delta_barth/_management.py @@ -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"] diff --git a/src/delta_barth/analysis/forecast.py b/src/delta_barth/analysis/forecast.py index 654f893..21300ce 100644 --- a/src/delta_barth/analysis/forecast.py +++ b/src/delta_barth/analysis/forecast.py @@ -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 diff --git a/src/delta_barth/api/_session.py b/src/delta_barth/api/_session.py deleted file mode 100644 index 27ac6b1..0000000 --- a/src/delta_barth/api/_session.py +++ /dev/null @@ -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 diff --git a/src/delta_barth/api/common.py b/src/delta_barth/api/common.py index 6f35be8..7b26001 100644 --- a/src/delta_barth/api/common.py +++ b/src/delta_barth/api/common.py @@ -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) +# 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 - # err = DelBarApiError(status_code=resp.status_code, **resp.json()) - # response = SalesPrognosisResponse(daten=tuple(), error=err) - # else: # pragma: no cover - # _raise_for_unknown_error(resp) - - return response, status +# return response, status diff --git a/src/delta_barth/api/requests.py b/src/delta_barth/api/requests.py new file mode 100644 index 0000000..5ca5a43 --- /dev/null +++ b/src/delta_barth/api/requests.py @@ -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 diff --git a/src/delta_barth/constants.py b/src/delta_barth/constants.py index 029a2e6..0ba61ec 100644 --- a/src/delta_barth/constants.py +++ b/src/delta_barth/constants.py @@ -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", diff --git a/src/delta_barth/errors.py b/src/delta_barth/errors.py index 243cee0..677408b 100644 --- a/src/delta_barth/errors.py +++ b/src/delta_barth/errors.py @@ -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() diff --git a/src/delta_barth/types.py b/src/delta_barth/types.py index f815d2d..5b25a70 100644 --- a/src/delta_barth/types.py +++ b/src/delta_barth/types.py @@ -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 = "" diff --git a/tests/api/conftest.py b/tests/api/conftest.py new file mode 100644 index 0000000..ac6338c --- /dev/null +++ b/tests/api/conftest.py @@ -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 diff --git a/tests/api/test_common.py b/tests/api/test_common.py index f653984..adb1e19 100644 --- a/tests/api/test_common.py +++ b/tests/api/test_common.py @@ -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 diff --git a/tests/api/test_requests.py b/tests/api/test_requests.py new file mode 100644 index 0000000..7ccc46f --- /dev/null +++ b/tests/api/test_requests.py @@ -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 diff --git a/tests/conftest.py b/tests/conftest.py index a3db0ad..56bccd1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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") diff --git a/tests/test_errors.py b/tests/test_errors.py index 7ef54d0..74a1d75 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -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