gitea_admin/sientia-dataops-model-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1645: code snapshot (part 1)

Browse Source
This commit is contained in:
vitor-aignosi
2026-07-16 13:29:23 -03:00
commit 40f26bcf17
114 changed files with 18654 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

630
tests/utils/repository/test_data_manager_repository.py Normal file
View File

@@ -0,0 +1,630 @@
"""Unit tests for DataManagerRepository and module helpers."""
from __future__ import annotations
import json
from typing import Any
from unittest.mock import MagicMock, Mock, patch
import numpy as np
import pandas as pd
import pytest
from model_manager.runtime_paths import REPORTS_ROOT
from model_manager.utils.models.train_model_params import TrainModelParams
from model_manager.utils.models.train_model_result import TrainModelResult
from model_manager.utils.repository import data_manager_repository as dmr
def test_train_test_split_dataframe_shuffle():
df = pd.DataFrame({'a': range(10)})
tr, te = dmr.train_test_split(df, train_size=0.7, random_state=0, shuffle=True)
assert len(tr) == 7 and len(te) == 3
def test_train_test_split_dataframe_no_shuffle():
df = pd.DataFrame({'a': range(10)})
tr, te = dmr.train_test_split(df, train_size=0.5, shuffle=False)
assert list(tr['a']) == [0, 1, 2, 3, 4]
def test_train_test_split_dataframe_returns_dataframes():
df = pd.DataFrame(np.arange(20).reshape(10, 2), columns=['a', 'b'])
tr, te = dmr.train_test_split(df, train_size=0.5, shuffle=False, random_state=None)
assert isinstance(tr, pd.DataFrame)
assert isinstance(te, pd.DataFrame)
assert tr.shape[0] == 5 and te.shape[0] == 5
def _params(**kwargs) -> TrainModelParams:
base: dict[str, Any] = {
'variable_columns': ['v1'],
'target_variable': 't',
'bucket_name': 'b',
'file_name': 'f.csv',
'line_separator': ',',
'decimal_separator': '.',
'date_column': 'timestamp',
'date_format': 'yyyy-MM-dd HH:mm:ss',
'train_size': 80,
'shuffle': True,
'random_state': 42,
'experiment_run_id': 1,
'model_name': 'Linear Regression',
'val_file_name': None,
'data_model_kwargs': {},
'model_kwargs': {},
'opt_params': {},
'model_type': 'linear_regression',
'model_id': None,
'model_metadata': {'schemas': {'components': {'schemas': {}}}},
}
base.update(kwargs)
return TrainModelParams.from_dict(base)
def test_ensure_date_column_parsed_missing_column_raises():
df = pd.DataFrame({'a': [1]})
p = _params(date_column='missing')
with pytest.raises(ValueError, match='not found in dataset columns'):
dmr._ensure_date_column_parsed(df, p)
def test_ensure_date_column_parsed_success():
df = pd.DataFrame({'a': range(3), 'ts': ['2024-01-01 10:00:00'] * 3})
p = _params(date_column='ts')
out = dmr._ensure_date_column_parsed(df, p)
assert pd.api.types.is_datetime64_any_dtype(out['ts'])
def test_ensure_date_column_parsed_naive_with_frontend_format():
"""CSV timestamps without timezone use params.date_format strftime mapping."""
df = pd.DataFrame({'ts': ['2025-06-02 00:00:00', '2025-06-02 01:00:00']})
p = _params(date_column='ts', date_format='yyyy-MM-dd HH:mm:ss')
out = dmr._ensure_date_column_parsed(df, p)
assert pd.api.types.is_datetime64_any_dtype(out['ts'])
def test_ensure_date_column_parsed_invalid_raises():
df = pd.DataFrame({'a': range(3), 'ts': ['not-a-date'] * 3})
p = _params(date_column='ts')
with pytest.raises(ValueError, match='Failed to parse date column'):
dmr._ensure_date_column_parsed(df, p)
def test_prepare_training_data_csv_load_failure():
repo = dmr.DataManagerRepository(MagicMock())
p = _params()
with patch(
'model_manager.utils.repository.data_manager_repository.pd.read_csv',
side_effect=pd.errors.ParserError('bad'),
):
with pytest.raises(ValueError, match='Failed to load training CSV'):
repo.prepare_training_data(b'x', None, p, {})
def test_prepare_training_data_empty_after_load():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
# Headers only; timestamp column present but no data rows
csv_bytes = b'timestamp,v1,t\n'
with pytest.raises(ValueError, match='Training data view is empty after transformation'):
repo.prepare_training_data(csv_bytes, None, p, {})
def test_prepare_training_data_empty_after_transformation(monkeypatch):
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
monkeypatch.setattr(
repo,
'_configure_datetime_index',
lambda *_args, **_kwargs: pd.DataFrame(columns=['v1', 't']),
)
monkeypatch.setattr(repo, '_set_timezone_on_index', lambda data, *_args, **_kwargs: data)
with pytest.raises(ValueError, match='Training data view is empty after transformation'):
repo.prepare_training_data(b'timestamp,v1,t\n', None, p, {})
def _minimal_dict_for_prepare():
return {
'variable_columns': ['v1'],
'target_variable': 't',
'bucket_name': 'b',
'file_name': 'f.csv',
'line_separator': ',',
'decimal_separator': '.',
'date_column': 'timestamp',
'date_format': 'yyyy-MM-dd HH:mm:ss',
'train_size': 80,
'shuffle': True,
'random_state': 42,
'experiment_run_id': 1,
'model_name': 'Linear Regression',
'val_file_name': None,
'data_model_kwargs': {},
'model_kwargs': {},
'opt_params': {},
'model_type': 'linear_regression',
'model_id': None,
'model_metadata': {'schemas': {'components': {'schemas': {}}}},
}
def _csv_bytes_with_ts(n_rows: int = 20) -> bytes:
"""CSV with leading timestamp column (naive, matches default date_format)."""
lines = ['timestamp,v1,t']
for i in range(n_rows):
lines.append(f'2024-01-{i + 1:02d} 00:00:00,{i},{i + 1}')
return '\n'.join(lines).encode()
def test_prepare_training_data_validation_csv_invalid():
from io import BytesIO
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
train_csv = _csv_bytes_with_ts(5)
train_df = pd.read_csv(BytesIO(train_csv), sep=',', decimal='.')
with patch.object(
dmr.pd,
'read_csv',
side_effect=[train_df, pd.errors.ParserError('bad val')],
):
with pytest.raises(ValueError, match='Failed to load validation CSV'):
repo.prepare_training_data(train_csv, b'broken', p, {})
def test_prepare_training_data_validation_empty_val():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
train_csv = _csv_bytes_with_ts(5)
val_csv = b'timestamp,v1,t\n'
with pytest.raises(ValueError, match='Validation data view is empty'):
repo.prepare_training_data(train_csv, val_csv, p, {})
def test_prepare_training_data_drops_row_with_blank_timestamp():
"""Rows with empty date_column values are removed before datetime parsing."""
repo = dmr.DataManagerRepository(MagicMock())
d = _minimal_dict_for_prepare()
d['date_column'] = 'timestamp'
d['date_format'] = 'yyyy-MM-dd HH:mm:ss'
p = TrainModelParams.from_dict(d)
lines = ['timestamp,v1,t']
for i in range(10):
if i == 3:
lines.append(',1.0,2.0')
else:
lines.append(f'2025-06-01 {i:02d}:00:00,1.0,2.0')
csv = '\n'.join(lines).encode()
res = repo.prepare_training_data(csv, None, p, {})
assert len(res.train_data) + len(res.val_data) == 9
def test_prepare_training_data_split_path():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
train_csv = _csv_bytes_with_ts(20)
res = repo.prepare_training_data(train_csv, None, p, {})
assert res.train_data is not None and res.val_data is not None
def test_prepare_training_data_explicit_validation_success():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
train_csv = _csv_bytes_with_ts(10)
val_csv = _csv_bytes_with_ts(5)
res = repo.prepare_training_data(train_csv, val_csv, p, {})
assert len(res.val_data) == 5
def test_coerce_non_timestamp_columns_to_numeric_success():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
idx = pd.date_range('2024-01-01', periods=2, freq='h', tz='UTC')
df = pd.DataFrame(
{'v1': ['1.25', '2.75'], 't': ['10', '11']},
index=idx,
)
out = repo._coerce_non_timestamp_columns_to_numeric(df, p, {})
assert pd.api.types.is_numeric_dtype(out['v1'])
assert pd.api.types.is_numeric_dtype(out['t'])
assert float(out['v1'].iloc[0]) == 1.25
assert float(out['t'].iloc[1]) == 11.0
def test_coerce_non_timestamp_columns_to_numeric_invalid_values_to_nan():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
idx = pd.date_range('2024-01-01', periods=2, freq='h', tz='UTC')
df = pd.DataFrame(
{'v1': ['1.25', 'oops'], 't': ['10', 'bad']},
index=idx,
)
out = repo._coerce_non_timestamp_columns_to_numeric(df, p, {})
assert np.isnan(out['v1'].iloc[1])
assert np.isnan(out['t'].iloc[1])
def test_prepare_training_data_coerces_non_timestamp_columns_to_numeric():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
lines = ['timestamp,v1,t']
for i in range(10):
v1 = 'bad' if i == 4 else f'{i + 0.5}'
t = 'bad' if i == 7 else f'{i + 1.0}'
lines.append(f'2025-06-01 {i:02d}:00:00,{v1},{t}')
csv = '\n'.join(lines).encode()
res = repo.prepare_training_data(csv, None, p, {})
joined = pd.concat([res.train_data, res.val_data], axis=0).sort_index()
assert pd.api.types.is_numeric_dtype(joined['v1'])
assert pd.api.types.is_numeric_dtype(joined['t'])
assert joined['v1'].isna().sum() == 1
assert joined['t'].isna().sum() == 1
def test_as_series_series():
repo = dmr.DataManagerRepository(MagicMock())
s = pd.Series([1.0, 2.0])
assert repo._as_series(s).equals(s)
def test_as_series_one_column_df():
repo = dmr.DataManagerRepository(MagicMock())
df = pd.DataFrame({'x': [1.0, 2.0]})
out = repo._as_series(df)
assert isinstance(out, pd.Series)
def test_as_series_multi_column_raises():
repo = dmr.DataManagerRepository(MagicMock())
df = pd.DataFrame({'a': [1.0], 'b': [2.0]})
with pytest.raises(ValueError, match='single-column'):
repo._as_series(df)
def test_compute_regression_metrics_requires_y_pred():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'t': [1.0]}),
val_data=pd.DataFrame({'t': [1.0]}),
y_pred=None,
)
with pytest.raises(ValueError, match='y_pred must be set'):
repo.compute_regression_metrics(tmr, MagicMock())
def test_compute_regression_metrics_no_overlap():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'t': [1.0]}),
val_data=pd.DataFrame({'t': [1.0]}, index=[10]),
y_pred=pd.DataFrame({'p': [1.0]}, index=[20]),
)
with pytest.raises(ValueError, match='No overlapping indices'):
repo.compute_regression_metrics(tmr, MagicMock())
def test_compute_regression_metrics_linear_equation():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
p.model_type = 'linear_regression'
idx = pd.Index([0, 1])
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
val_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
y_pred=pd.DataFrame({'p': [1.0, 2.0]}, index=idx),
)
regr = MagicMock()
regr.coef_ = np.array([0.5])
regr.intercept_ = 1.0
wrapper = MagicMock()
wrapper.model = MagicMock()
wrapper.model.regr = regr
out = repo.compute_regression_metrics(tmr, wrapper)
assert out.mse_val is not None and out.equation is not None
def test_compute_regression_metrics_linear_skips_equation_without_sklearn_regr():
"""E2E dummy wrappers expose model without sklearn .regr; metrics still compute."""
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
p.model_type = 'linear_regression'
idx = pd.Index([0, 1])
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
val_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
y_pred=pd.DataFrame({'p': [1.0, 2.0]}, index=idx),
)
wrapper = MagicMock()
wrapper.model = object()
out = repo.compute_regression_metrics(tmr, wrapper)
assert out.mse_val is not None and out.equation is None
def test_compute_regression_metrics_non_linear_skips_equation():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
p.model_type = 'xgboost'
idx = pd.Index([0, 1])
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
val_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
y_pred=pd.DataFrame({'p': [1.0, 2.0]}, index=idx),
)
out = repo.compute_regression_metrics(tmr, MagicMock())
assert out.mse_val is not None and out.equation is None
def test_configure_datetime_index_none_raises():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
with pytest.raises(ValueError, match='Data is None'):
repo._configure_datetime_index(None, p, {})
def test_configure_datetime_index_already_datetime_index():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
existing_idx = pd.date_range('2024-01-01', periods=3, freq='h')
df = pd.DataFrame(
{
'timestamp': pd.to_datetime(
['2024-01-03 00:00:00', '2024-01-01 00:00:00', '2024-01-02 00:00:00']
),
'v1': [1, 2, 3],
't': [1, 2, 3],
},
index=existing_idx,
)
out = repo._configure_datetime_index(df, p, {})
assert isinstance(out.index, pd.DatetimeIndex)
assert out.index.equals(pd.DatetimeIndex(pd.to_datetime(sorted(df['timestamp'].tolist()))))
def test_configure_datetime_index_from_date_column():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict({**_minimal_dict_for_prepare(), 'date_column': 'mydate'})
df = pd.DataFrame(
{
'mydate': pd.date_range('2024-01-01', periods=3, freq='D'),
'v1': [1, 2, 3],
't': [1, 2, 3],
}
)
out = repo._configure_datetime_index(df, p, {})
assert isinstance(out.index, pd.DatetimeIndex)
def test_configure_datetime_index_missing_date_column_raises():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict({**_minimal_dict_for_prepare(), 'date_column': 'mydate'})
df = pd.DataFrame(
{
'timestamp': pd.date_range('2024-01-01', periods=3, freq='D'),
'v1': [1, 2, 3],
't': [1, 2, 3],
}
)
with pytest.raises(ValueError, match='date_column "mydate" not found'):
repo._configure_datetime_index(df, p, {})
def test_configure_datetime_index_non_datetime_date_column_raises():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict({**_minimal_dict_for_prepare(), 'date_column': 'mydate'})
df = pd.DataFrame(
{
'mydate': ['2024-01-01', '2024-01-02', '2024-01-03'],
'v1': [1.0, 2.0, 3.0],
't': [1.0, 2.0, 3.0],
}
)
with pytest.raises(ValueError, match='must be datetime before index configuration'):
repo._configure_datetime_index(df, p, {})
def test_create_run_directory_permission_error():
repo = dmr.DataManagerRepository(MagicMock())
with patch(
'model_manager.utils.repository.data_manager_repository.makedirs',
side_effect=PermissionError('no'),
):
with pytest.raises(PermissionError, match='Permission denied'):
repo._create_run_directory('/tmp', 'run', {})
def test_create_run_directory_os_error():
repo = dmr.DataManagerRepository(MagicMock())
with patch(
'model_manager.utils.repository.data_manager_repository.makedirs',
side_effect=OSError('disk'),
):
with pytest.raises(OSError, match='Failed to create directory'):
repo._create_run_directory('/tmp', 'run', {})
def test_generate_report_success(tmp_path):
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'v1': [1.0, 2.0], 't': [1.0, 2.0]}),
val_data=pd.DataFrame({'v1': [1.0, 2.0], 't': [1.0, 2.0]}),
y_train_pred=pd.DataFrame({'t': [1.0, 2.0]}),
y_pred=pd.DataFrame({'t': [1.0, 2.0]}),
run_name='testrun',
)
tmr.equation = {'target_variable': 't'}
with (
patch.object(repo, '_get_reports_directory', return_value=str(tmp_path)),
patch('model_manager.utils.repository.data_manager_repository.Reports') as mrep,
):
instance = mrep.return_value
instance.save_all_sections_html = Mock()
out = repo.generate_report(tmr, {})
assert out.report_path and out.train_data_path and out.test_data_path
if out.equation_path:
with open(out.equation_path, encoding='utf-8') as f:
json.load(f)
def test_generate_report_adds_target_alias_for_reports(tmp_path):
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'v1': [1.0, 2.0], 't': [1.0, 2.0]}),
val_data=pd.DataFrame({'v1': [1.0, 2.0], 't': [1.0, 2.0]}),
y_train_pred=pd.DataFrame({'t': [1.0, 2.0]}),
y_pred=pd.DataFrame({'t': [1.0, 2.0]}),
run_name='testrun',
)
with (
patch.object(repo, '_get_reports_directory', return_value=str(tmp_path)),
patch('model_manager.utils.repository.data_manager_repository.Reports') as mrep,
):
instance = mrep.return_value
instance.save_all_sections_html = Mock()
out = repo.generate_report(tmr, {})
kwargs = mrep.call_args.kwargs
reference_data = kwargs['reference_data']
current_data = kwargs['current_data']
assert 'target' in reference_data.columns
assert 'target' in current_data.columns
assert reference_data['target'].equals(reference_data['t'])
assert current_data['target'].equals(current_data['t'])
assert out.train_data_path is not None
assert out.test_data_path is not None
train_csv = pd.read_csv(out.train_data_path)
test_csv = pd.read_csv(out.test_data_path)
assert 'target' in train_csv.columns
assert 'target' in test_csv.columns
assert train_csv['target'].equals(train_csv['t'])
assert test_csv['target'].equals(test_csv['t'])
def test_generate_report_skips_equation_file_when_not_linear(tmp_path):
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
p.model_type = 'other'
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'v1': [1.0, 2.0], 't': [1.0, 2.0]}),
val_data=pd.DataFrame({'v1': [1.0, 2.0], 't': [1.0, 2.0]}),
y_train_pred=pd.DataFrame({'t': [1.0, 2.0]}),
y_pred=pd.DataFrame({'t': [1.0, 2.0]}),
run_name='testrun',
equation={'k': 'v'},
)
with (
patch.object(repo, '_get_reports_directory', return_value=str(tmp_path)),
patch('model_manager.utils.repository.data_manager_repository.Reports'),
):
out = repo.generate_report(tmr, {})
assert out.equation_path is None
def test_generate_report_run_name_missing():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'t': [1.0]}),
val_data=pd.DataFrame({'t': [1.0]}),
run_name=None,
)
with pytest.raises(ValueError, match='run_name is not set'):
repo.generate_report(tmr, {})
def test_generate_report_requires_predictions():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
tmr = TrainModelResult(
params=p,
train_data=pd.DataFrame({'t': [1.0]}),
val_data=pd.DataFrame({'t': [1.0]}),
run_name='testrun',
y_train_pred=None,
y_pred=None,
)
with pytest.raises(ValueError, match='y_train_pred or y_pred is not set'):
repo.generate_report(tmr, {})
def test_cleanup_run_directory_empty():
repo = dmr.DataManagerRepository(MagicMock())
repo.cleanup_run_directory('', {})
def test_cleanup_run_directory_exists(tmp_path):
repo = dmr.DataManagerRepository(MagicMock())
d = tmp_path / 'subdir'
d.mkdir()
repo.cleanup_run_directory(str(d), {})
assert not d.exists()
def test_cleanup_run_directory_missing(tmp_path):
repo = dmr.DataManagerRepository(MagicMock())
repo.cleanup_run_directory(str(tmp_path / 'nope'), {})
def test_extract_model_equation_polynomial_poly_names():
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
p.model_kwargs = {'degree': 2, 'poly_feature_names': ['f1', 'f2']}
regr = MagicMock()
regr.coef_ = np.array([1.0, 2.0])
regr.intercept_ = 3.0
wrapper = MagicMock()
wrapper.model = MagicMock()
wrapper.model.regr = regr
eq = repo._extract_model_equation(wrapper.model, p)
assert 'equation_string' in eq and eq['degree'] == 2
def test_extract_model_equation_extra_coefficients_ignored():
"""More coefficients than feature names: only the first len(names) are used."""
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
regr = MagicMock()
regr.coef_ = np.array([1.0, 2.0, 3.0])
regr.intercept_ = 0.0
wrapper = MagicMock()
wrapper.model = MagicMock()
wrapper.model.regr = regr
eq = repo._extract_model_equation(wrapper.model, p)
assert len(eq['coefficients']) == len(p.variable_columns)
def test_extract_model_equation_more_features_than_coefficients():
"""Polynomial feature names longer than coef array: extra names get no coefficient entry."""
repo = dmr.DataManagerRepository(MagicMock())
p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
p.model_kwargs = {'degree': 2, 'poly_feature_names': ['a', 'b', 'c']}
regr = MagicMock()
regr.coef_ = np.array([1.0, 2.0])
regr.intercept_ = 0.0
wrapper = MagicMock()
wrapper.model = MagicMock()
wrapper.model.regr = regr
eq = repo._extract_model_equation(wrapper.model, p)
assert list(eq['coefficients'].keys()) == ['a', 'b']
def test_get_reports_directory_path():
repo = dmr.DataManagerRepository(MagicMock())
reports_dir = repo._get_reports_directory()
assert reports_dir == REPORTS_ROOT