-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
1 changed file
with
135 additions
and
183 deletions.
There are no files selected for viewing
318 changes: 135 additions & 183 deletions
318
validphys2/src/validphys/tests/test_inconsistent_ct.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,202 +1,154 @@ | ||
""" | ||
Module to test the InconsistentCommonData class. | ||
Module for testing the InconsistentCommonData class in the inconsistent_closuretest module. | ||
Testing is done by mocking the class's methods and properties. | ||
""" | ||
|
||
from numpy.testing import assert_allclose | ||
import unittest | ||
from unittest.mock import MagicMock, patch | ||
import pandas as pd | ||
from io import StringIO | ||
|
||
from validphys.tests.conftest import SINGLE_DATASET | ||
from validphys.closuretest.inconsistent_closuretest.inconsistent_ct import InconsistentCommonData | ||
from validphys.api import API | ||
|
||
class TestInconsistentCommonData(unittest.TestCase): | ||
|
||
cd = API.commondata(**{"dataset_input": {**SINGLE_DATASET}}).load() | ||
|
||
inconsys_cd = InconsistentCommonData( | ||
setname=cd.setname, | ||
ndata=cd.ndata, | ||
commondataproc=cd.commondataproc, | ||
nkin=cd.nkin, | ||
nsys=cd.nsys, | ||
commondata_table=cd.commondata_table, | ||
systype_table=cd.systype_table, | ||
) | ||
|
||
|
||
def test_with_MULT_sys(): | ||
""" | ||
test if MULT commondata_table is | ||
replaced correctly by | ||
dataclasses.replace(self, commondata_table = new_table) | ||
""" | ||
|
||
mult_sys_tab = 3 * cd.commondata_table["MULT"].to_numpy() | ||
|
||
inc_mult_sys_tab = inconsys_cd.with_MULT_sys(mult_sys_tab).commondata_table["MULT"].to_numpy() | ||
|
||
assert_allclose(mult_sys_tab, inc_mult_sys_tab) | ||
|
||
|
||
def test_with_ADD_sys(): | ||
""" | ||
test if ADD commondata_table is | ||
replaced correctly by | ||
dataclasses.replace(self, commondata_table = new_table) | ||
""" | ||
|
||
mult_sys_tab = 3 * cd.commondata_table["ADD"].to_numpy() | ||
|
||
inc_mult_sys_tab = inconsys_cd.with_ADD_sys(mult_sys_tab).commondata_table["ADD"].to_numpy() | ||
|
||
assert_allclose(mult_sys_tab, inc_mult_sys_tab) | ||
|
||
|
||
def test_rescale_sys_CORR_MULT(): | ||
""" | ||
Check whether rescaling of | ||
CORR MULT uncertainties works | ||
as expected | ||
""" | ||
|
||
rescaling_factor = 2.0 | ||
treatment_err = "MULT" | ||
new_icd = inconsys_cd.with_MULT_sys( | ||
inconsys_cd.rescale_sys( | ||
treatment_err=treatment_err, | ||
CORR=True, | ||
UNCORR=False, | ||
SPECIAL=False, | ||
sys_rescaling_factor=rescaling_factor, | ||
) | ||
@patch( | ||
'validphys.closuretest.inconsistent_closuretest.inconsistent_ct.InconsistentCommonData', | ||
autospec=True, | ||
) | ||
|
||
# get indices of CORR sys | ||
systype_corr = cd.systype_table[ | ||
(cd.systype_table["treatment"] == treatment_err) | ||
& (~cd.systype_table["name"].isin(["UNCORR", "THEORYUNCORR"])) | ||
] | ||
|
||
tab2 = rescaling_factor * cd.systematics_table.iloc[:, systype_corr.index - 1].to_numpy() | ||
|
||
tab1 = new_icd.systematics_table.iloc[:, systype_corr.index - 1] | ||
|
||
assert_allclose(tab1, tab2) | ||
|
||
|
||
def test_rescale_sys_CORR_ADD(): | ||
""" | ||
Check whether rescaling of | ||
CORR ADD uncertainties works | ||
as expected | ||
""" | ||
|
||
rescaling_factor = 2.0 | ||
treatment_err = "ADD" | ||
new_icd = inconsys_cd.with_ADD_sys( | ||
inconsys_cd.rescale_sys( | ||
treatment_err, | ||
CORR=True, | ||
UNCORR=False, | ||
SPECIAL=False, | ||
sys_rescaling_factor=rescaling_factor, | ||
def setUp(self, MockInconsistentCommonData): | ||
""" | ||
Set up mock instance of InconsistentCommonData for all tests. | ||
""" | ||
self.mock_instance = MockInconsistentCommonData.return_value | ||
|
||
# Mocking the DataFrames in the instance | ||
self.mock_instance.systype_table = pd.DataFrame( | ||
{"treatment": ["ADD", "MULT", "ADD"], "name": ["CORR", "UNCORR", "SPECIAL"]} | ||
) | ||
) | ||
|
||
# get indices of CORR sys | ||
systype_corr = cd.systype_table[ | ||
(cd.systype_table["treatment"] == treatment_err) | ||
& (~cd.systype_table["name"].isin(["UNCORR", "THEORYUNCORR"])) | ||
] | ||
|
||
tab2 = rescaling_factor * cd.systematics_table.iloc[:, systype_corr.index - 1].to_numpy() | ||
self.mock_instance.systematic_errors = pd.DataFrame( | ||
{"sys1": [0.1, 0.2, 0.3], "sys2": [0.4, 0.5, 0.6]} | ||
) | ||
|
||
tab1 = new_icd.systematics_table.iloc[:, systype_corr.index - 1] | ||
def test_systematic_errors_getter(self): | ||
""" | ||
Test the getter for the systematic_errors property. | ||
""" | ||
# Set the _systematic_errors to None so the getter is triggered | ||
self.mock_instance._systematic_errors = None | ||
|
||
# Mock the return value of the superclass's systematic_errors method | ||
with patch( | ||
'validphys.coredata.CommonData.systematic_errors', | ||
return_value=self.mock_instance.systematic_errors, | ||
): | ||
result = self.mock_instance.systematic_errors | ||
|
||
# Assert that the result matches the mock | ||
pd.testing.assert_frame_equal(result, self.mock_instance.systematic_errors) | ||
|
||
def test_systematic_errors_setter(self): | ||
""" | ||
Test the setter for the systematic_errors property. | ||
""" | ||
new_systematic_errors = pd.DataFrame({"sys1": [0.2, 0.3, 0.4], "sys2": [0.5, 0.6, 0.7]}) | ||
|
||
self.mock_instance.systematic_errors = new_systematic_errors | ||
pd.testing.assert_frame_equal(self.mock_instance.systematic_errors, new_systematic_errors) | ||
|
||
def test_select_systype_table_indices(self): | ||
""" | ||
Test select_systype_table_indices method with valid input. | ||
""" | ||
treatment_names = ["ADD"] | ||
names_uncertainties = ["CORR", "SPECIAL"] | ||
|
||
# Mock return of select_systype_table_indices call | ||
self.mock_instance.select_systype_table_indices.return_value = pd.Index([0, 2]) | ||
|
||
result = self.mock_instance.select_systype_table_indices( | ||
treatment_names, names_uncertainties | ||
) | ||
|
||
assert_allclose(tab1, tab2) | ||
self.mock_instance.select_systype_table_indices.assert_called_once_with( | ||
treatment_names, names_uncertainties | ||
) | ||
pd.testing.assert_index_equal(result, pd.Index([0, 2])) | ||
|
||
def test_select_systype_table_indices_invalid_uncertainties(self): | ||
""" | ||
Test select_systype_table_indices with invalid uncertainties. | ||
""" | ||
treatment_names = ["ADD"] | ||
names_uncertainties = ["INVALID"] | ||
|
||
# Mock the behavior of raising a ValueError | ||
self.mock_instance.select_systype_table_indices.side_effect = ValueError( | ||
"names_uncertainties should only contain either CORR, UNCORR, THEORYCORR, THEORYUNCORR or SPECIAL" | ||
) | ||
|
||
with self.assertRaises(ValueError): | ||
self.mock_instance.select_systype_table_indices(treatment_names, names_uncertainties) | ||
|
||
def test_rescale_systematics(self): | ||
""" | ||
Test rescale_systematics method. | ||
""" | ||
self.mock_instance.systematic_errors = self.mock_instance.systematic_errors.copy() | ||
treatment_names = ["ADD"] | ||
names_uncertainties = ["CORR"] | ||
sys_rescaling_factor = 2.0 | ||
|
||
# Mock return of rescale_systematics | ||
rescaled_table = self.mock_instance.systematic_errors.copy() | ||
rescaled_table.iloc[:, 0] *= sys_rescaling_factor | ||
self.mock_instance.rescale_systematics.return_value = rescaled_table | ||
|
||
result = self.mock_instance.rescale_systematics( | ||
treatment_names, names_uncertainties, sys_rescaling_factor | ||
) | ||
|
||
def test_process_commondata(): | ||
""" | ||
Check whether process_commondata | ||
leaves the commondata instance | ||
unchanged when told to do so. | ||
""" | ||
# Assert that rescale_systematics was called once and that the return value matches the mock | ||
self.mock_instance.rescale_systematics.assert_called_once_with( | ||
treatment_names, names_uncertainties, sys_rescaling_factor | ||
) | ||
pd.testing.assert_frame_equal(result, rescaled_table) | ||
|
||
def test_process_commondata(self): | ||
""" | ||
Test process_commondata method when the dataset is inconsistent. | ||
""" | ||
inconsistent_datasets = ["test_dataset"] | ||
treatment_names = ["ADD"] | ||
names_uncertainties = ["CORR"] | ||
sys_rescaling_factor = 2.0 | ||
|
||
# Mock the return of process_commondata | ||
modified_commondata = MagicMock() | ||
self.mock_instance.process_commondata.return_value = modified_commondata | ||
|
||
result = self.mock_instance.process_commondata( | ||
treatment_names, names_uncertainties, sys_rescaling_factor, inconsistent_datasets | ||
) | ||
|
||
new_icd = inconsys_cd.process_commondata( | ||
ADD=False, | ||
MULT=False, | ||
CORR=False, | ||
UNCORR=False, | ||
SPECIAL=False, | ||
inconsistent_datasets=[SINGLE_DATASET['dataset']], | ||
sys_rescaling_factor=1, | ||
) | ||
tab1 = new_icd.commondata_table.drop(['process'], axis=1).to_numpy() | ||
tab2 = inconsys_cd.commondata_table.drop(['process'], axis=1).to_numpy() | ||
|
||
assert_allclose(tab1, tab2) | ||
|
||
|
||
def test_process_commondata_CORR_MULT(): | ||
""" | ||
Check whether rescaling of | ||
CORR MULT uncertainties works | ||
as expected with process_commondata | ||
method | ||
""" | ||
|
||
treatment_err = "MULT" | ||
rescaling_factor = 2.0 | ||
new_icd = inconsys_cd.process_commondata( | ||
ADD=False, | ||
MULT=True, | ||
CORR=True, | ||
UNCORR=False, | ||
SPECIAL=False, | ||
inconsistent_datasets=[SINGLE_DATASET['dataset']], | ||
sys_rescaling_factor=rescaling_factor, | ||
) | ||
# Assert that the method was called with correct parameters | ||
self.mock_instance.process_commondata.assert_called_once_with( | ||
treatment_names, names_uncertainties, sys_rescaling_factor, inconsistent_datasets | ||
) | ||
self.assertEqual(result, modified_commondata) | ||
|
||
# get indices of CORR sys | ||
systype_corr = cd.systype_table[ | ||
(cd.systype_table["treatment"] == treatment_err) | ||
& (~cd.systype_table["name"].isin(["UNCORR", "THEORYUNCORR"])) | ||
] | ||
|
||
tab2 = rescaling_factor * cd.systematics_table.iloc[:, systype_corr.index - 1].to_numpy() | ||
|
||
tab1 = new_icd.systematics_table.iloc[:, systype_corr.index - 1] | ||
|
||
assert_allclose(tab1, tab2) | ||
|
||
|
||
def test_process_commondata_CORR_ADD(): | ||
""" | ||
Check whether rescaling of | ||
CORR ADD uncertainties works | ||
as expected with process_commondata | ||
method | ||
""" | ||
|
||
treatment_err = "ADD" | ||
rescaling_factor = 2.0 | ||
new_icd = inconsys_cd.process_commondata( | ||
ADD=True, | ||
MULT=False, | ||
CORR=True, | ||
UNCORR=False, | ||
SPECIAL=False, | ||
inconsistent_datasets=[SINGLE_DATASET['dataset']], | ||
sys_rescaling_factor=rescaling_factor, | ||
) | ||
def test_export_uncertainties(self): | ||
""" | ||
Test the export_uncertainties method. | ||
""" | ||
buffer = StringIO() | ||
|
||
# get indices of CORR sys | ||
systype_corr = cd.systype_table[ | ||
(cd.systype_table["treatment"] == treatment_err) | ||
& (~cd.systype_table["name"].isin(["UNCORR", "THEORYUNCORR"])) | ||
] | ||
# Mock the export_uncertainties method | ||
self.mock_instance.export_uncertainties.return_value = None | ||
|
||
tab2 = rescaling_factor * cd.systematics_table.iloc[:, systype_corr.index - 1].to_numpy() | ||
self.mock_instance.export_uncertainties(buffer) | ||
self.mock_instance.export_uncertainties.assert_called_once_with(buffer) | ||
|
||
tab1 = new_icd.systematics_table.iloc[:, systype_corr.index - 1] | ||
|
||
assert_allclose(tab1, tab2) | ||
if __name__ == "__main__": | ||
unittest.main() |