324 lines
9.9 KiB
Python
324 lines
9.9 KiB
Python
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
import delta_barth.session
|
|
from delta_barth import logging
|
|
from delta_barth.constants import (
|
|
DEFAULT_API_ERR_CODE,
|
|
HTTP_BASE_CONTENT_HEADERS,
|
|
LOG_FILENAME,
|
|
)
|
|
|
|
|
|
def test_validate_path_Success():
|
|
str_pth = str(Path.cwd())
|
|
path = delta_barth.session.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"):
|
|
_ = delta_barth.session.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"):
|
|
_ = delta_barth.session.validate_path(str_pth)
|
|
|
|
|
|
def test_session_set_DataPath(tmp_path):
|
|
str_path = str(tmp_path)
|
|
session = delta_barth.session.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_session_setup_db_management(tmp_path):
|
|
str_path = str(tmp_path)
|
|
foldername: str = "data_test"
|
|
target_db_dir = tmp_path / foldername
|
|
|
|
session = delta_barth.session.Session(HTTP_BASE_CONTENT_HEADERS, db_folder=foldername)
|
|
session.set_data_path(str_path)
|
|
db_path = session.db_path
|
|
assert db_path.parent.exists()
|
|
assert db_path.parent == target_db_dir
|
|
assert not db_path.exists()
|
|
session.setup()
|
|
db_path2 = session.db_path
|
|
assert db_path2 == db_path
|
|
assert session._db_engine is not None
|
|
assert db_path.exists()
|
|
|
|
|
|
@patch("delta_barth.logging.ENABLE_LOGGING", True)
|
|
@patch("delta_barth.logging.LOGGING_TO_FILE", True)
|
|
def test_session_setup_logging(tmp_path):
|
|
str_path = str(tmp_path)
|
|
foldername: str = "logging_test"
|
|
target_log_dir = tmp_path / foldername
|
|
|
|
session = delta_barth.session.Session(
|
|
HTTP_BASE_CONTENT_HEADERS, logging_folder=foldername
|
|
)
|
|
session.set_data_path(str_path)
|
|
log_dir = session.logging_dir
|
|
|
|
assert log_dir.exists()
|
|
assert log_dir == target_log_dir
|
|
# write file
|
|
target_file = target_log_dir / LOG_FILENAME
|
|
assert not target_file.exists()
|
|
session.setup() # calls setup code for logging
|
|
log_dir2 = session.logging_dir
|
|
assert log_dir2 == log_dir
|
|
assert target_file.exists()
|
|
|
|
|
|
@patch("delta_barth.logging.ENABLE_LOGGING", True)
|
|
@patch("delta_barth.logging.LOGGING_TO_FILE", True)
|
|
def test_session_disable_logging(tmp_path):
|
|
str_path = str(tmp_path)
|
|
foldername: str = "logging_test"
|
|
target_log_dir = tmp_path / foldername
|
|
|
|
session = delta_barth.session.Session(
|
|
HTTP_BASE_CONTENT_HEADERS, logging_folder=foldername
|
|
)
|
|
session.set_data_path(str_path)
|
|
log_dir = session.logging_dir
|
|
assert log_dir.exists()
|
|
assert log_dir == target_log_dir
|
|
# write file
|
|
target_file = target_log_dir / LOG_FILENAME
|
|
assert not target_file.exists()
|
|
session.setup() # calls setup code for logging
|
|
assert target_file.exists()
|
|
# provoke entry
|
|
msg = "this is a test"
|
|
logging.logger_base.critical(msg)
|
|
session.disable_logging()
|
|
with open(target_file, "r") as file:
|
|
content = file.readlines()
|
|
last_line = content[-1]
|
|
assert msg in last_line.lower()
|
|
# log new entry which should not be added as logging is disabled
|
|
msg = "this is a second test"
|
|
logging.logger_base.critical(msg)
|
|
with open(target_file, "r") as file:
|
|
content = file.readlines()
|
|
last_line = content[-1]
|
|
assert msg not in last_line.lower()
|
|
|
|
|
|
def test_session_set_ApiInfo_LoggedOut(credentials, api_base_url):
|
|
session = delta_barth.session.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 = delta_barth.session.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"]
|