125 lines
3.5 KiB
Python
125 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tomllib
|
|
from pathlib import Path
|
|
from typing import cast
|
|
from unittest.mock import patch
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
import tomli_w
|
|
|
|
import delta_barth.session
|
|
from delta_barth.api.requests import SalesPrognosisResponse
|
|
from delta_barth.constants import HTTP_BASE_CONTENT_HEADERS
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def credentials() -> dict[str, str]:
|
|
pwd = Path.cwd()
|
|
assert "barth" in pwd.parent.name.lower(), "not in project root directory"
|
|
creds_pth = pwd / "./CREDENTIALS.toml"
|
|
assert creds_pth.exists(), "file to credentials data not found"
|
|
with open(creds_pth, "rb") as file:
|
|
cfg = tomllib.load(file)
|
|
|
|
creds = cast(dict[str, str], cfg["delta-barth-server"]["api"])
|
|
|
|
return creds
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def api_base_url(credentials) -> str:
|
|
return credentials["base_url"]
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def pth_dummy_cfg() -> Path:
|
|
pwd = Path.cwd()
|
|
assert "barth" in pwd.parent.name.lower(), "not in project root directory"
|
|
data_pth = pwd / "./tests/_test_data/dopt-cfg.toml"
|
|
assert data_pth.exists(), "file to dummy CFG not found"
|
|
return data_pth
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def pth_cfg(pth_dummy_cfg, tmp_path) -> Path:
|
|
with open(pth_dummy_cfg, "rb") as file:
|
|
cfg_data = tomllib.load(file)
|
|
|
|
target = tmp_path / "dummy_cfg.toml"
|
|
target.touch()
|
|
with open(target, "wb") as file:
|
|
tomli_w.dump(cfg_data, file)
|
|
|
|
return target
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def sales_data_real() -> pd.DataFrame:
|
|
pwd = Path.cwd()
|
|
assert "barth" in pwd.parent.name.lower(), "not in project root directory"
|
|
data_pth = pwd / "./tests/_test_data/exmp_sales_prognosis_resp.json"
|
|
assert data_pth.exists(), "file to API sales data not found"
|
|
|
|
with open(data_pth, "r") as file:
|
|
data = json.load(file)
|
|
|
|
parsed = SalesPrognosisResponse(**data)
|
|
data = parsed.model_dump()["daten"]
|
|
|
|
return pd.DataFrame(data)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def exmpl_api_sales_prognosis_resp() -> SalesPrognosisResponse:
|
|
pwd = Path.cwd()
|
|
assert "barth" in pwd.parent.name.lower(), "not in project root directory"
|
|
data_pth = pwd / "./tests/_test_data/exmp_sales_prognosis_resp.json"
|
|
assert data_pth.exists(), "file to API sales data not found"
|
|
|
|
with open(data_pth, "r") as file:
|
|
data = json.load(file)
|
|
|
|
return SalesPrognosisResponse(**data)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def exmpl_api_sales_prognosis_output() -> pd.DataFrame:
|
|
pwd = Path.cwd()
|
|
assert "barth" in pwd.parent.name.lower(), "not in project root directory"
|
|
data_pth = pwd / "./tests/_test_data/exmp_sales_prognosis_output.pkl"
|
|
assert data_pth.exists(), "file to API sales data not found"
|
|
|
|
return pd.read_pickle(data_pth)
|
|
|
|
|
|
# ** sessions
|
|
@pytest.fixture(scope="function")
|
|
def session(credentials, api_base_url, tmp_path) -> delta_barth.session.Session:
|
|
session = delta_barth.session.Session(HTTP_BASE_CONTENT_HEADERS)
|
|
session.set_data_path(str(tmp_path))
|
|
session.set_base_url(api_base_url)
|
|
session.set_credentials(
|
|
username=credentials["user"],
|
|
password=credentials["pwd"],
|
|
database=credentials["db"],
|
|
mandant=credentials["mandant"],
|
|
)
|
|
session.setup()
|
|
|
|
return session
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_put():
|
|
with patch("requests.put") as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_get():
|
|
with patch("requests.get") as mock:
|
|
yield mock
|