add env setup for runtime to enable multiprocessing parameter search by joblib, closes #23

tests/analysis/test_forecast.py

@@ -430,6 +430,7 @@ def test_export_on_fail():
     assert res.status.description == status.description
 
 
+@patch("delta_barth.session.CFG_HOT_RELOAD", False)
 def test_pipeline_sales_forecast_SuccessDbWrite(exmpl_api_sales_prognosis_resp, session):
     with (
         patch(

tests/test_env.py

@@ -0,0 +1,49 @@
+import importlib
+import sys
+from unittest.mock import patch
+
+import pytest
+
+import delta_barth.constants
+from delta_barth import _env
+
+
+@patch("delta_barth._env.PY_RUNTIME_FOLDER", "test123456")
+def test_prepare_env_NoRuntimeFolder(tmp_path):
+    ret = _env.prepare_env(tmp_path)
+    assert ret is None
+
+
+@patch("delta_barth._env.PY_RUNTIME_FOLDER", "base")
+def test_prepare_env_FailNoInterpreter(tmp_path_factory):
+    mocked_lib_pth = tmp_path_factory.mktemp("path") / "to/base/folder/lib/"
+    mocked_lib_pth.mkdir(parents=True, exist_ok=True)
+    with pytest.raises(FileNotFoundError):
+        _ = _env.prepare_env(mocked_lib_pth)
+
+
+@patch("delta_barth._env.PY_RUNTIME_FOLDER", "base")
+def test_prepare_env_Success(tmp_path_factory):
+    mocked_lib_pth = tmp_path_factory.mktemp("path") / "to/base/folder/lib/"
+    mocked_lib_pth.mkdir(parents=True, exist_ok=True)
+    rt_path = mocked_lib_pth.parents[1]
+    mocked_interpreter = rt_path / "python.exe"
+    mocked_interpreter.touch()
+    assert mocked_interpreter.exists()
+    ret = _env.prepare_env(mocked_lib_pth)
+    assert ret == rt_path
+    # sys attributes
+    executable = getattr(sys, "executable")
+    assert executable == str(mocked_interpreter)
+    base_executable = getattr(sys, "_base_executable")
+    assert base_executable == str(mocked_interpreter)
+
+    class MockPath:
+        def __init__(self, *args, **kwargs):
+            self.parent = mocked_lib_pth
+
+    with patch("pathlib.Path", MockPath):
+        (mocked_lib_pth / "_dummy_data").mkdir(exist_ok=True)
+        importlib.reload(delta_barth.constants)
+        assert delta_barth.constants.DEPLOYMENT_STATUS
+        assert delta_barth.constants.RUNTIME_PATH == rt_path

tests/test_pipelines.py

@@ -45,6 +45,7 @@ def test_write_performance_metrics_FailStartingTime(session):
             )
 
 
+@patch("delta_barth.session.CFG_HOT_RELOAD", False)
 def test_sales_prognosis_pipeline(exmpl_api_sales_prognosis_resp, session, monkeypatch):
     with (
         patch(

tests/test_session.py

@@ -1,15 +1,18 @@
+import tomllib
 from pathlib import Path
 from unittest.mock import patch
 
 import pytest
+import tomli_w
 
+import delta_barth.config
 import delta_barth.session
 from delta_barth import logging
 from delta_barth.constants import (
     DEFAULT_API_ERR_CODE,
     HTTP_BASE_CONTENT_HEADERS,
-    LOG_FILENAME,
 )
+from delta_barth.logging import LOG_FILENAME
 
 
 def test_validate_path_Success():
@@ -62,7 +65,7 @@ def test_session_setup_db_management(tmp_path):
     assert db_path.exists()
 
 
-def test_session_setup_config(tmp_path, pth_cfg):
+def test_session_setup_config(tmp_path):
     str_path = str(tmp_path)
     foldername: str = "cfg_test"
     target_cfg_dir = tmp_path / foldername
@@ -80,6 +83,61 @@ def test_session_setup_config(tmp_path, pth_cfg):
     assert session.cfg.forecast.threshold_month_data_points == 28
 
 
+@patch("delta_barth.session.CFG_HOT_RELOAD", False)
+def test_session_reload_config_NoHotReload(tmp_path):
+    str_path = str(tmp_path)
+    foldername: str = "cfg_test"
+    target_cfg_dir = tmp_path / foldername
+    session = delta_barth.session.Session(HTTP_BASE_CONTENT_HEADERS, cfg_folder=foldername)
+    session.set_data_path(str_path)
+    cfg_path = session.cfg_path
+    assert cfg_path.parent.exists()
+    assert cfg_path.parent == target_cfg_dir
+    assert not cfg_path.exists()
+    session.setup()
+    assert cfg_path.exists()
+    parsed_cfg = session.cfg
+    assert isinstance(parsed_cfg, delta_barth.config.Config)
+    # modify config and reload
+    with open(cfg_path, "rb") as file:
+        cfg_data = tomllib.load(file)
+    cfg_data["forecast"]["threshold_month_data_points"] = 30
+    with open(cfg_path, "wb") as file:
+        tomli_w.dump(cfg_data, file)
+
+    assert session.cfg.forecast.threshold_month_data_points == 28
+
+    session.reload_cfg()
+    reload_cfg = session.cfg
+    assert isinstance(reload_cfg, delta_barth.config.Config)
+    assert reload_cfg.forecast.threshold_month_data_points == 30
+
+
+@patch("delta_barth.session.CFG_HOT_RELOAD", True)
+def test_session_reload_config_HotReload(tmp_path):
+    str_path = str(tmp_path)
+    foldername: str = "cfg_test"
+    target_cfg_dir = tmp_path / foldername
+    session = delta_barth.session.Session(HTTP_BASE_CONTENT_HEADERS, cfg_folder=foldername)
+    session.set_data_path(str_path)
+    cfg_path = session.cfg_path
+    assert cfg_path.parent.exists()
+    assert cfg_path.parent == target_cfg_dir
+    assert not cfg_path.exists()
+    session.setup()
+    assert cfg_path.exists()
+    parsed_cfg = session.cfg
+    assert isinstance(parsed_cfg, delta_barth.config.Config)
+    # modify config and reload
+    with open(cfg_path, "rb") as file:
+        cfg_data = tomllib.load(file)
+    cfg_data["forecast"]["threshold_month_data_points"] = 30
+    with open(cfg_path, "wb") as file:
+        tomli_w.dump(cfg_data, file)
+
+    assert session.cfg.forecast.threshold_month_data_points == 30
+
+
 @patch("delta_barth.logging.ENABLE_LOGGING", True)
 @patch("delta_barth.logging.LOGGING_TO_FILE", True)
 @patch("delta_barth.logging.LOGGING_TO_STDERR", True)