add logging of pipeline metrics in database

tests/test_pipelines.py

@@ -3,23 +3,44 @@ import json
 from unittest.mock import patch
 
 import pytest
+import sqlalchemy as sql
 
 import delta_barth.pipelines
+from delta_barth import databases as db
 from delta_barth import pipelines as pl
 from delta_barth.errors import STATUS_HANDLER
 
 
-def test_write_performance_metrics(session): ...
+def test_write_performance_metrics(session):
+    pipe_name = "test_pipe"
+    t_start = 20_000_000_000
+    t_end = 30_000_000_000
+
+    with patch("delta_barth.pipelines.SESSION", session):
+        metrics = pl._write_performance_metrics(
+            pipeline_name=pipe_name,
+            time_start=t_start,
+            time_end=t_end,
+        )
+    assert metrics["pipeline_name"] == pipe_name
+    assert metrics["execution_duration"] == 10
+
+    with session.db_engine.begin() as con:
+        ret = con.execute(sql.select(db.perf_meas))
+
+    metrics = ret.all()[-1]
+    assert metrics.pipeline_name == pipe_name
+    assert metrics.execution_duration == 10
 
 
 @patch("delta_barth.analysis.forecast.SALES_BASE_NUM_DATAPOINTS_MONTHS", 1)
-def test_sales_prognosis_pipeline(exmpl_api_sales_prognosis_resp):
+def test_sales_prognosis_pipeline(exmpl_api_sales_prognosis_resp, session):
     with patch(
         "delta_barth.analysis.forecast.get_sales_prognosis_data",
     ) as mock:
         mock.return_value = (exmpl_api_sales_prognosis_resp, STATUS_HANDLER.SUCCESS)
-        importlib.reload(delta_barth.pipelines)
-        json_export = pl.pipeline_sales_forecast(None, None)
+        with patch("delta_barth.pipelines.SESSION", session):
+            json_export = pl.pipeline_sales_forecast(None, None)
 
     assert isinstance(json_export, str)
     parsed_resp = json.loads(json_export)
@@ -30,9 +51,18 @@ def test_sales_prognosis_pipeline(exmpl_api_sales_prognosis_resp):
     assert "code" in parsed_resp["status"]
     assert parsed_resp["status"]["code"] == 0
 
+    with session.db_engine.begin() as con:
+        ret = con.execute(sql.select(db.perf_meas))
 
-def test_sales_prognosis_pipeline_dummy():
-    json_export = pl.pipeline_sales_forecast_dummy(None, None)
+    metrics = ret.all()[-1]
+    assert metrics.pipeline_name == "sales_forecast"
+    assert metrics.execution_duration > 0
+
+
+@pytest.mark.new
+def test_sales_prognosis_pipeline_dummy(session):
+    with patch("delta_barth.pipelines.SESSION", session):
+        json_export = pl.pipeline_sales_forecast_dummy(None, None)
 
     assert isinstance(json_export, str)
     parsed_resp = json.loads(json_export)
@@ -46,3 +76,10 @@ def test_sales_prognosis_pipeline_dummy():
     assert entry["vorhersage"] == pytest.approx(47261.058594)
     assert "code" in parsed_resp["status"]
     assert parsed_resp["status"]["code"] == 0
+
+    with session.db_engine.begin() as con:
+        ret = con.execute(sql.select(db.perf_meas))
+
+    metrics = ret.all()[-1]
+    assert metrics.pipeline_name == "sales_forecast_dummy"
+    assert metrics.execution_duration > 0