103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tomllib
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from delta_barth.api.requests import SalesPrognosisResponse
|
|
|
|
|
|
@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"]
|
|
|
|
|
|
# TODO: maybe include in main package depending if needed in future
|
|
def _cvt_str_float(value: str) -> float:
|
|
import locale
|
|
|
|
locale.setlocale(locale.LC_NUMERIC, "de_DE.UTF-8")
|
|
return locale.atof(value)
|
|
|
|
|
|
def _cvt_str_ts(value: str) -> Any:
|
|
date = value.split("_")[0]
|
|
|
|
return pd.to_datetime(date, format="%Y%m%d", errors="coerce")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def sales_data_db_export() -> pd.DataFrame:
|
|
pwd = Path.cwd()
|
|
assert "barth" in pwd.parent.name.lower(), "not in project root directory"
|
|
data_pth = pwd / "./tests/_test_data/swm_f_umsatz_fakt.csv"
|
|
assert data_pth.exists(), "file to sales data not found"
|
|
data = pd.read_csv(data_pth, sep="\t")
|
|
data["betrag"] = data["betrag"].apply(_cvt_str_float)
|
|
data["buchungs_datum"] = data["buchungs_datum"].apply(_cvt_str_ts)
|
|
data = data.dropna(
|
|
how="any",
|
|
subset=["firma_refid", "beleg_typ", "buchungs_datum", "betrag"],
|
|
ignore_index=True,
|
|
)
|
|
data["buchungs_datum"] = pd.to_datetime(data["buchungs_datum"])
|
|
|
|
return data
|
|
|
|
|
|
@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_ouput.pkl"
|
|
assert data_pth.exists(), "file to API sales data not found"
|
|
|
|
return pd.read_pickle(data_pth)
|