delta-barth/delta-barth-py
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 3 Releases 3 Wiki Activity

fix return values in pipeline wrapper

Browse Source
This commit is contained in:
Florian Förster 2025-03-20 12:29:04 +01:00
parent fc94badcf3
commit 0eccccafe5
3 changed files with 28 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[project] [project]
name = "delta-barth" name = "delta-barth"
version = "0.2.2" version = "0.2.3"
description = "prognosis module" description = "prognosis module"
authors = [ authors = [
{name = "Florian Förster", email = "f.foerster@d-opt.com"}, {name = "Florian Förster", email = "f.foerster@d-opt.com"},

9
src/delta_barth/pipelines.py
View File

@ -10,12 +10,13 @@ from delta_barth.types import JsonExportResponse, JsonResponse, JsonStatus
def pipeline_sales_forecast( def pipeline_sales_forecast(
company_id: int | None, company_id: int | None,
start_date: Datetime | None, start_date: Datetime | None,
) -> tuple[JsonResponse, JsonStatus]: ) -> JsonExportResponse:
result = forecast.pipeline_sales(SESSION, company_id=company_id, start_date=start_date) result = forecast.pipeline_sales(SESSION, company_id=company_id, start_date=start_date)
response = JsonResponse(result.response.model_dump_json()) export = JsonExportResponse(result.model_dump_json())
status = JsonStatus(result.status.model_dump_json()) # response = JsonResponse(result.response.model_dump_json())
# status = JsonStatus(result.status.model_dump_json())
return response, status return export
def pipeline_sales_forecast_dummy( def pipeline_sales_forecast_dummy(

47
tests/test_pipelines.py
View File

@ -10,38 +10,35 @@ from delta_barth.errors import STATUS_HANDLER
def test_sales_prognosis_pipeline(exmpl_api_sales_prognosis_resp): def test_sales_prognosis_pipeline(exmpl_api_sales_prognosis_resp):
def mock_request(*args, **kwargs): # pragma: no cover
return exmpl_api_sales_prognosis_resp, STATUS_HANDLER.SUCCESS
with patch( with patch(
"delta_barth.api.requests.get_sales_prognosis_data", "delta_barth.analysis.forecast.get_sales_prognosis_data",
new=mock_request, ) as mock:
): mock.return_value = (exmpl_api_sales_prognosis_resp, STATUS_HANDLER.SUCCESS)
importlib.reload(delta_barth.pipelines) importlib.reload(delta_barth.pipelines)
json_resp, json_stat = pl.pipeline_sales_forecast(None, None) json_export = pl.pipeline_sales_forecast(None, None)
assert isinstance(json_resp, str) assert isinstance(json_export, str)
assert isinstance(json_stat, str) parsed_resp = json.loads(json_export)
parsed_resp = json.loads(json_resp) assert "response" in parsed_resp
assert "daten" in parsed_resp assert "status" in parsed_resp
assert len(parsed_resp["daten"]) > 0 assert "daten" in parsed_resp["response"]
parsed_stat = json.loads(json_stat) assert len(parsed_resp["response"]["daten"]) > 0
assert "code" in parsed_stat assert "code" in parsed_resp["status"]
assert parsed_stat["code"] == 0 assert parsed_resp["status"]["code"] == 0
def test_sales_prognosis_pipeline_dummy(): def test_sales_prognosis_pipeline_dummy():
json_resp, json_stat = pl.pipeline_sales_forecast_dummy(None, None) json_export = pl.pipeline_sales_forecast_dummy(None, None)
assert isinstance(json_resp, str) assert isinstance(json_export, str)
assert isinstance(json_stat, str) parsed_resp = json.loads(json_export)
parsed_resp = json.loads(json_resp) assert "response" in parsed_resp
assert "daten" in parsed_resp assert "status" in parsed_resp
assert len(parsed_resp["daten"]) > 0 assert "daten" in parsed_resp["response"]
entry = parsed_resp["daten"][0] assert len(parsed_resp["response"]["daten"]) > 0
entry = parsed_resp["response"]["daten"][0]
assert entry["jahr"] == 2022 assert entry["jahr"] == 2022
assert entry["monat"] == 11 assert entry["monat"] == 11
assert entry["vorhersage"] == pytest.approx(47261.058594) assert entry["vorhersage"] == pytest.approx(47261.058594)
parsed_stat = json.loads(json_stat) assert "code" in parsed_resp["status"]
assert "code" in parsed_stat assert parsed_resp["status"]["code"] == 0
assert parsed_stat["code"] == 0