290 lines
8.9 KiB
Python
290 lines
8.9 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from delta_barth.api import common
|
|
from delta_barth.constants import DEFAULT_API_ERR_CODE, HTTP_BASE_CONTENT_HEADERS
|
|
from delta_barth.errors import (
|
|
UnspecifiedRequestType,
|
|
)
|
|
from delta_barth.types import HttpRequestTypes
|
|
|
|
|
|
def test_validate_path_Success():
|
|
str_pth = str(Path.cwd())
|
|
path = common.validate_path(str_pth)
|
|
assert path.name == Path.cwd().name
|
|
|
|
|
|
def test_validate_path_FailNotExisting():
|
|
str_pth = str(Path.cwd() / "test")
|
|
with pytest.raises(FileNotFoundError, match=r"seems not to exist"):
|
|
_ = common.validate_path(str_pth)
|
|
|
|
|
|
def test_validate_path_FailNoDirectory(tmp_path):
|
|
file = tmp_path / "test.txt"
|
|
file.write_text("test", encoding="utf-8")
|
|
|
|
str_pth = str(file)
|
|
with pytest.raises(FileNotFoundError, match=r"seems not to be a directory"):
|
|
_ = common.validate_path(str_pth)
|
|
|
|
|
|
def test_session_set_DataPath(tmp_path):
|
|
str_path = str(tmp_path)
|
|
session = common.Session(HTTP_BASE_CONTENT_HEADERS)
|
|
|
|
assert session._data_path is None
|
|
|
|
session.set_data_path(str_path)
|
|
assert session._data_path is not None
|
|
assert isinstance(session.data_path, Path)
|
|
|
|
|
|
def test_validate_creds(credentials):
|
|
creds = common.validate_credentials(
|
|
username=credentials["user"],
|
|
password=credentials["pwd"],
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
assert creds.username == 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(
|
|
username=user,
|
|
password=credentials["pwd"],
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
assert user != credentials["user"]
|
|
assert creds.username == credentials["user"]
|
|
# invalid type
|
|
user = 123
|
|
with pytest.raises(ValidationError):
|
|
creds = common.validate_credentials(
|
|
username=user, # type: ignore
|
|
password=credentials["pwd"],
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_ping(api_base_url):
|
|
resp = common.ping(api_base_url, HttpRequestTypes.GET)
|
|
assert resp.status_code == 204
|
|
resp = common.ping(api_base_url, HttpRequestTypes.PUT)
|
|
assert resp.status_code == 204
|
|
resp = common.ping(api_base_url, HttpRequestTypes.DELETE)
|
|
assert resp.status_code == 204
|
|
|
|
with pytest.raises(UnspecifiedRequestType):
|
|
resp = common.ping(api_base_url, HttpRequestTypes.POST)
|
|
|
|
|
|
def test_session_set_ApiInfo_LoggedOut(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(
|
|
username=credentials["user"],
|
|
password=credentials["pwd"],
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
assert session._creds is not None
|
|
|
|
assert session.session_token is None
|
|
assert not session.logged_in
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_session_set_ApiInfo_LoggedIn(credentials, api_base_url):
|
|
session = common.Session(HTTP_BASE_CONTENT_HEADERS)
|
|
# prepare login
|
|
assert session.session_token is None
|
|
assert session._creds is None
|
|
assert session._base_url is None
|
|
session.set_base_url(api_base_url)
|
|
session.set_credentials(
|
|
username=credentials["user"],
|
|
password=credentials["pwd"],
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
session.login()
|
|
assert session._base_url is not None
|
|
assert session.logged_in
|
|
# reset base URL
|
|
session.set_base_url(api_base_url)
|
|
assert session._base_url is not None
|
|
assert not session.logged_in
|
|
assert session.session_token is None
|
|
# reset credentials
|
|
session.login()
|
|
assert session.logged_in
|
|
session.set_credentials(
|
|
username=credentials["user"],
|
|
password=credentials["pwd"],
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
assert session._creds is not None
|
|
assert not session.logged_in
|
|
assert session.session_token is None
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_login_logout_Success(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(
|
|
username=credentials["user"],
|
|
password="WRONG_PASSWORD",
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
resp, status = session.login()
|
|
assert resp is not None
|
|
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."
|
|
|
|
|
|
def test_login_logout_FailApiServer(session, mock_put):
|
|
code = 401
|
|
json = {
|
|
"message": "GenericError",
|
|
"code": "TestLogin",
|
|
"hints": "TestCase",
|
|
}
|
|
|
|
mock_put.return_value.status_code = code
|
|
mock_put.return_value.json.return_value = json
|
|
resp, status = session.login()
|
|
assert resp is not None
|
|
assert not resp.token
|
|
assert status.code == 400
|
|
assert status.api_server_error is not None
|
|
assert status.api_server_error.status_code == code
|
|
assert status.api_server_error.message == json["message"]
|
|
assert status.api_server_error.code == json["code"]
|
|
assert status.api_server_error.hints == json["hints"]
|
|
resp, status = session.logout()
|
|
assert resp is None
|
|
assert status.code == 400
|
|
assert status.api_server_error is not None
|
|
assert status.api_server_error.status_code == code
|
|
assert status.api_server_error.message == json["message"]
|
|
assert status.api_server_error.code == json["code"]
|
|
assert status.api_server_error.hints == json["hints"]
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_assert_login_SuccessLoggedOut(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
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_assert_login_SuccessStillLoggedIn(session):
|
|
assert session.session_token is None
|
|
assert session._creds is not None
|
|
resp, status = session.login()
|
|
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
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_assert_login_ReloginNoValidAuth(session, mock_get):
|
|
code = 401
|
|
json = {
|
|
"message": "AuthentificationError",
|
|
"code": "TestAssertLoginAfter",
|
|
"hints": "TestCase",
|
|
}
|
|
mock_get.return_value.status_code = code
|
|
mock_get.return_value.json.return_value = json
|
|
|
|
resp, status = session.login()
|
|
|
|
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
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_assert_login_ReloginWrongToken(session):
|
|
# triggers code 401
|
|
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
|
|
|
|
|
|
@pytest.mark.api_con_required
|
|
def test_assert_login_FailApiServer(session, mock_get):
|
|
code = 500
|
|
json = {
|
|
"message": "ServerError",
|
|
"code": "TestExternalServerError",
|
|
"hints": "TestCase",
|
|
}
|
|
mock_get.return_value.status_code = code
|
|
mock_get.return_value.json.return_value = json
|
|
|
|
resp, status = session.login()
|
|
|
|
resp, status = session.assert_login()
|
|
assert resp is not None
|
|
assert not resp.token
|
|
assert status.code == 400
|
|
assert status.api_server_error is not None
|
|
assert status.api_server_error.status_code == code
|
|
assert status.api_server_error.message == json["message"]
|
|
assert status.api_server_error.code == json["code"]
|
|
assert status.api_server_error.hints == json["hints"]
|