diff --git a/src/delta_barth/api/common.py b/src/delta_barth/api/common.py index f74e02c..47f5896 100644 --- a/src/delta_barth/api/common.py +++ b/src/delta_barth/api/common.py @@ -2,32 +2,26 @@ from __future__ import annotations import re from datetime import datetime as Datetime -from typing import Final, Never +from typing import TYPE_CHECKING, Final, Never import requests from pydantic import BaseModel, PositiveInt, SkipValidation from requests import Response +from delta_barth._management import STATE_HANDLER from delta_barth.constants import HTTP_CURRENT_CONNECTION, KnownDelBarApiErrorCodes from delta_barth.errors import ( ApiConnectionError, - UnknownApiErrorCode, UnspecifiedRequestType, ) from delta_barth.types import DelBarApiError, HttpRequestTypes +if TYPE_CHECKING: + from delta_barth.types import Status + LOGIN_ERROR_CODES_KNOWN: Final[frozenset[int]] = frozenset((400, 401, 409, 500)) -def _raise_for_unknown_error( - resp: Response, -) -> Never: - raise UnknownApiErrorCode( - f"Unknown response code for request. Status Code: {resp.status_code}, " - f"Content: {resp.text}" - ) - - def _assert_login_status() -> None: if not HTTP_CURRENT_CONNECTION.logged_in: raise ApiConnectionError("Curent session is not logged in") @@ -73,7 +67,6 @@ class LoginRequest(BaseModel): class LoginResponse(BaseModel): token: str - error: DelBarApiError | None = None def login( @@ -82,7 +75,7 @@ def login( password: str, database: str, mandant: str, -) -> LoginResponse: +) -> tuple[LoginResponse, Status]: ROUTE: Final[str] = "user/login" URL: Final = combine_route(base_url, ROUTE) @@ -99,26 +92,23 @@ def login( ) response: LoginResponse + status: Status if resp.status_code == 200: response = LoginResponse(**resp.json()) + status = STATE_HANDLER.pipe_states.SUCCESS HTTP_CURRENT_CONNECTION.add_session_token(response.token) - elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value: + else: + response = LoginResponse(token="") err = DelBarApiError(status_code=resp.status_code, **resp.json()) - response = LoginResponse(token="", error=err) - else: # pragma: no cover - _raise_for_unknown_error(resp) + status = STATE_HANDLER.api_error(err) - return response + return response, status # ** logout -class LogoutResponse(BaseModel): - error: DelBarApiError | None = None - - def logout( base_url: str, -) -> LogoutResponse: +) -> tuple[None, Status]: ROUTE: Final[str] = "user/logout" URL: Final = combine_route(base_url, ROUTE) @@ -127,17 +117,16 @@ def logout( headers=HTTP_CURRENT_CONNECTION.headers, # type: ignore ) - response: LogoutResponse + response = None + status: Status if resp.status_code == 200: - response = LogoutResponse() + status = STATE_HANDLER.SUCCESS HTTP_CURRENT_CONNECTION.remove_session_token() - elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value: + else: err = DelBarApiError(status_code=resp.status_code, **resp.json()) - response = LogoutResponse(error=err) - else: # pragma: no cover - _raise_for_unknown_error(resp) + status = STATE_HANDLER.api_error(err) - return response + return response, status # ** sales data @@ -156,14 +145,14 @@ class SalesPrognosisResponseEntry(BaseModel): class SalesPrognosisResponse(BaseModel): daten: tuple[SalesPrognosisResponseEntry, ...] - error: DelBarApiError | None = None + # error: DelBarApiError | None = None def get_sales_prognosis_data( base_url: str, company_id: int | None = None, start_date: Datetime | None = None, -) -> SalesPrognosisResponse: +) -> tuple[SalesPrognosisResponse, Status]: _assert_login_status() ROUTE: Final[str] = "verkauf/umsatzprognosedaten" URL: Final = combine_route(base_url, ROUTE) @@ -179,12 +168,19 @@ def get_sales_prognosis_data( ) response: SalesPrognosisResponse + status: Status if resp.status_code == 200: response = SalesPrognosisResponse(**resp.json()) - elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value: # pragma: no cover + status = STATE_HANDLER.SUCCESS + else: + response = SalesPrognosisResponse(daten=tuple()) 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 + # 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 diff --git a/src/delta_barth/errors.py b/src/delta_barth/errors.py index 941033d..4f7a268 100644 --- a/src/delta_barth/errors.py +++ b/src/delta_barth/errors.py @@ -13,10 +13,6 @@ class UnspecifiedRequestType(Exception): """exception raised if for a given API endpoint a not defined operation is requested""" -class UnknownApiErrorCode(Exception): - """exception raised if for a given request a unknown error response code is transmitted""" - - class ApiConnectionError(Exception): """exception raised if an established connection is needed, but the current session is not connected""" diff --git a/tests/api/test_common.py b/tests/api/test_common.py index 9bffcde..63d8671 100644 --- a/tests/api/test_common.py +++ b/tests/api/test_common.py @@ -3,10 +3,9 @@ from datetime import datetime as Datetime import pytest from delta_barth.api import common -from delta_barth.constants import HTTP_CURRENT_CONNECTION +from delta_barth.constants import DEFAULT_API_ERR_CODE, HTTP_CURRENT_CONNECTION from delta_barth.errors import ( ApiConnectionError, - UnknownApiErrorCode, UnspecifiedRequestType, ) from delta_barth.types import HttpRequestTypes @@ -68,46 +67,43 @@ def test_ping(api_base_url): resp = common.ping(api_base_url, HttpRequestTypes.POST) -@pytest.mark.api_con_required -def test_raise_unknown_error(api_base_url): - resp = common.ping(api_base_url, HttpRequestTypes.GET) - with pytest.raises(UnknownApiErrorCode): - common._raise_for_unknown_error(resp) - - @pytest.mark.api_con_required def test_login_logout(credentials, api_base_url): assert HTTP_CURRENT_CONNECTION.session_token is None - resp = common.login( + resp, state = common.login( base_url=api_base_url, user_name=credentials["user"], password=credentials["pwd"], database=credentials["db"], mandant=credentials["mandant"], ) - assert resp.error is None + assert state.code == 0 assert HTTP_CURRENT_CONNECTION.session_token is not None - resp = common.logout( + resp, state = common.logout( base_url=api_base_url, ) - assert resp.error is None + assert resp is None + assert state.code == 0 assert HTTP_CURRENT_CONNECTION.session_token is None assert "DelecoToken" not in HTTP_CURRENT_CONNECTION.headers - resp = common.login( + + resp, state = common.login( base_url=api_base_url, user_name=credentials["user"], password="WRONG_PASSWORD", database=credentials["db"], mandant=credentials["mandant"], ) - assert resp.error is not None - assert resp.error.status_code == 409 - assert resp.error.message == "Nutzer oder Passwort falsch." + 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." @pytest.mark.api_con_required def test_get_sales_prognosis_data(credentials, api_base_url): - resp = common.login( + resp, state = common.login( base_url=api_base_url, user_name=credentials["user"], password=credentials["pwd"], @@ -115,35 +111,45 @@ def test_get_sales_prognosis_data(credentials, api_base_url): mandant=credentials["mandant"], ) # test without company ID - assert resp.error is None + assert state.code == 0 date = Datetime(2022, 6, 1) - resp = common.get_sales_prognosis_data(api_base_url, None, date) - assert resp.error is None + 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 = common.get_sales_prognosis_data(api_base_url, None, date) - assert resp.error is None + 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 resp.error is None + assert state.code == 0 date = Datetime(2022, 6, 1) company_id = 1024 - resp = common.get_sales_prognosis_data(api_base_url, company_id, date) - assert resp.error is None + 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 = common.get_sales_prognosis_data(api_base_url, company_id, date) - assert resp.error is None + 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 resp.error is None + assert state.code == 0 date = Datetime(2022, 6, 1) company_id = 1000024 - resp = common.get_sales_prognosis_data(api_base_url, company_id, date) - assert resp.error is None + 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 = common.logout( + resp, state = common.logout( base_url=api_base_url, ) - assert resp.error is None + assert state.code == 0