test_rule_processor
Tests for RuleBasedModel class
fixture_example_rule()
Inititaion of StepFunctionRule to be reused in the following tests
Source code in tests/business/entities/test_rule_processor.py
@pytest.fixture(name="example_rule")
def fixture_example_rule():
"""Inititaion of StepFunctionRule to be reused in the following tests"""
return StepFunctionRule(
"rule_name",
"input_variable_name",
[0, 1, 2, 5, 10],
[10, 11, 12, 15, 20],
)
test_creating_rule_processor_without_input_datasets_should_throw_exception()
Tests if input datasets are correctly checked during creation of the processor.
Source code in tests/business/entities/test_rule_processor.py
def test_creating_rule_processor_without_input_datasets_should_throw_exception():
"""
Tests if input datasets are correctly checked during creation of the processor.
"""
# Arrange
rule = Mock(IRule)
# Act
with pytest.raises(ValueError) as exc_info:
RuleProcessor([rule], None)
exception_raised = exc_info.value
# Assert
expected_message = "No datasets defined."
assert exception_raised.args[0] == expected_message
test_creating_rule_processor_without_rules_should_throw_exception()
Tests if absence of rules is correctly checked during creation of the processor.
Source code in tests/business/entities/test_rule_processor.py
def test_creating_rule_processor_without_rules_should_throw_exception():
"""
Tests if absence of rules is correctly checked during creation of the processor.
"""
# Arrange
dataset = _xr.Dataset()
dataset["test"] = _xr.DataArray([32, 94, 9])
rules = []
# Act
with pytest.raises(ValueError) as exc_info:
RuleProcessor(rules, dataset)
exception_raised = exc_info.value
# Assert
expected_message = "No rules defined."
assert exception_raised.args[0] == expected_message
test_execute_rule_throws_error_for_unknown_input_variable()
Tests that trying to execute a rule with an unknown input variable throws an error, and the error message.
Source code in tests/business/entities/test_rule_processor.py
def test_execute_rule_throws_error_for_unknown_input_variable():
"""Tests that trying to execute a rule with an unknown input variable
throws an error, and the error message."""
# Arrange
output_dataset = _xr.Dataset()
input_array = _xr.DataArray([32, 94, 9])
output_dataset["test"] = input_array
logger = Mock(ILogger)
rule = Mock(IRule)
rule.name = "test"
rule.input_variable_names = ["unexisting"]
rule.output_variable_name = "output"
processor = RuleProcessor([rule], output_dataset)
# Act
with pytest.raises(KeyError) as exc_info:
processor._execute_rule(rule, output_dataset, logger)
exception_raised = exc_info.value
# Assert
expected_message = (
f"Key {rule.input_variable_names[0]} was not found "
+ "in input datasets or in calculated output dataset."
)
assert exception_raised.args[0] == expected_message
test_initialization_for_different_rule_dependencies(indices_to_remove, expected_result)
Tests if the processor can initialize given the rule dependencies.
Source code in tests/business/entities/test_rule_processor.py
@pytest.mark.parametrize(
"indices_to_remove, expected_result",
[
[[0], False],
[[1], False],
[[2], False],
[[3], True],
[[2, 3], True],
[[1, 2, 3], True],
],
)
def test_initialization_for_different_rule_dependencies(
indices_to_remove: List[int], expected_result: bool
):
"""Tests if the processor can initialize given the rule dependencies."""
# Arrange
dataset = _xr.Dataset()
dataset["test"] = _xr.DataArray([32, 94, 9])
logger = Mock(ILogger)
rules = _create_test_rules()
processor = RuleProcessor(rules, dataset)
rules_to_remove = [rules[index] for index in indices_to_remove]
# remove rules
for rule in rules_to_remove:
rules.remove(rule)
# Act & Assert
assert expected_result == processor.initialize(logger)
test_initialization_given_rule_dependencies()
Tests if the processor can correctly initialize given the rule dependencies.
Source code in tests/business/entities/test_rule_processor.py
def test_initialization_given_rule_dependencies():
"""Tests if the processor can correctly initialize given
the rule dependencies.
"""
# Arrange
dataset = _xr.Dataset()
dataset["test"] = _xr.DataArray([32, 94, 9])
logger = Mock(ILogger)
rules = _create_test_rules()
processor = RuleProcessor(rules, dataset)
# Act & Assert
assert processor.initialize(logger)
test_process_rules_calls_array_based_rule_execute_correctly()
Tests if during processing the rule its execute method of an IArrayBasedRule is called with the right parameter.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_calls_array_based_rule_execute_correctly():
"""Tests if during processing the rule its execute method of
an IArrayBasedRule is called with the right parameter."""
# Arrange
output_dataset = _xr.Dataset()
input_array = _xr.DataArray([32, 94, 9])
output_dataset["test"] = input_array
logger = Mock(ILogger)
rule = Mock(IArrayBasedRule)
rule.input_variable_names = ["test"]
rule.output_variable_name = "output"
rule.execute.return_value = _xr.DataArray([4, 3, 2])
processor = RuleProcessor([rule], output_dataset)
# Act
assert processor.initialize(logger)
processor.process_rules(output_dataset, logger)
# Assert
assert len(output_dataset) == 2
assert rule.output_variable_name in output_dataset.keys()
rule.execute.assert_called_once_with(ANY, logger)
# get first call, first argument
array: _xr.DataArray = rule.execute.call_args[0][0]
_xr.testing.assert_equal(array, input_array)
test_process_rules_calls_cell_based_rule_execute_correctly()
Tests if during processing the rule its execute method of an ICellBasedRule is called with the right parameter.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_calls_cell_based_rule_execute_correctly():
"""Tests if during processing the rule its execute method of
an ICellBasedRule is called with the right parameter."""
# Arrange
dataset = _xr.Dataset()
input_array = _xr.DataArray(_np.array([[1, 2, 3], [4, 5, 6]], _np.int32))
dataset["test"] = input_array
logger = Mock(ILogger)
rule = Mock(ICellBasedRule)
rule.input_variable_names = ["test"]
rule.output_variable_name = "output"
# expected return value = 1; number of warnings (min and max) = 0 and 0
rule.execute.return_value = [1, [0, 0]]
processor = RuleProcessor([rule], dataset)
# Act
assert processor.initialize(logger)
processor.process_rules(dataset, logger)
# Assert
assert len(dataset) == 2
assert rule.output_variable_name in dataset.keys()
assert rule.execute.call_count == 6
test_process_rules_calls_multi_array_based_rule_execute_correctly()
Tests if during processing the rule its execute method of an IMultiArrayBasedRule is called with the right parameters.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_calls_multi_array_based_rule_execute_correctly():
"""Tests if during processing the rule its execute method of
an IMultiArrayBasedRule is called with the right parameters."""
# Arrange
dataset = _xr.Dataset()
array1 = _xr.DataArray([32, 94, 9])
array2 = _xr.DataArray([7, 93, 6])
dataset["test1"] = array1
dataset["test2"] = array2
logger = Mock(ILogger)
rule = Mock(IMultiArrayBasedRule)
rule.input_variable_names = ["test1", "test2"]
rule.output_variable_name = "output"
rule.execute.return_value = _xr.DataArray([4, 3, 2])
processor = RuleProcessor([rule], dataset)
# Act
assert processor.initialize(logger)
processor.process_rules(dataset, logger)
# Assert
assert len(dataset) == 3
assert rule.output_variable_name in dataset.keys()
rule.execute.assert_called_once_with(ANY, logger)
# get first call, first argument
array_lookup: Dict[str, _xr.DataArray] = rule.execute.call_args[0][0]
_xr.testing.assert_equal(array_lookup["test1"], array1)
_xr.testing.assert_equal(array_lookup["test2"], array2)
test_process_rules_calls_multi_cell_based_fails_with_different_dims()
MultiCellBasedRule allows for values with less dimensions, but not with different dimensions.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_calls_multi_cell_based_fails_with_different_dims():
"""MultiCellBasedRule allows for values with less dimensions, but not
with different dimensions."""
# Arrange
dataset = _xr.Dataset()
input_array1 = _xr.DataArray(
_np.array([1, 2], _np.int32),
dims=["x"],
coords={"x": [0, 1]},
)
input_array2 = _xr.DataArray(
_np.array([1, 2], _np.int32),
dims=["y"],
coords={"y": [0, 1]},
)
dataset["test1"] = input_array1
dataset["test2"] = input_array2
logger = Mock(ILogger)
rule = Mock(IMultiCellBasedRule)
rule.name = "test_rule"
rule.input_variable_names = ["test1", "test2"]
rule.output_variable_name = "output"
rule.execute.return_value = 1
processor = RuleProcessor([rule], dataset)
processor.initialize(logger)
# Act
with pytest.raises(NotImplementedError) as exc_info:
processor.process_rules(dataset, logger)
exception_raised = exc_info.value
# Assert
expected = f"Can not execute rule {rule.name} with variables with different \
dimensions. Variable test1 with dimensions:('x',) is \
different than test2 with dimensions:('y',)"
assert exception_raised.args[0] == expected
test_process_rules_calls_multi_cell_based_rule_execute_correctly()
Tests if during processing the rule its execute method of an IMultiCellBasedRule is called with the right parameter.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_calls_multi_cell_based_rule_execute_correctly():
"""Tests if during processing the rule its execute method of
an IMultiCellBasedRule is called with the right parameter."""
# Arrange
dataset = _xr.Dataset()
input_array1 = _xr.DataArray(_np.array([[1, 2, 3], [4, 5, 6]], _np.int32))
input_array2 = _xr.DataArray(_np.array([[1, 2, 3], [4, 5, 6]], _np.int32))
dataset["test1"] = input_array1
dataset["test2"] = input_array2
logger = Mock(ILogger)
rule = Mock(IMultiCellBasedRule)
rule.input_variable_names = ["test1", "test2"]
rule.output_variable_name = "output"
rule.execute.return_value = 1
processor = RuleProcessor([rule], dataset)
# Act
assert processor.initialize(logger)
processor.process_rules(dataset, logger)
# Assert
assert len(dataset) == 3
assert rule.output_variable_name in dataset.keys()
assert rule.execute.call_count == 6
test_process_rules_calls_multi_cell_based_rule_special_cases(input_array1, input_array2, dims)
Some exceptional cases need to be tested for the multi_cell rule: 1. variables with different dimensions (1D vs 2D) 2. variables with different dimensions (1D vs 3D)
Source code in tests/business/entities/test_rule_processor.py
@pytest.mark.parametrize(
"input_array1, input_array2, dims",
[
(
_xr.DataArray(
_np.array([1, 2], _np.int32),
dims=["x"],
coords={"x": [0, 1]},
),
_xr.DataArray(
_np.array([[1, 2], [3, 4]], _np.int32),
dims=["x", "y"],
coords={"x": [0, 1], "y": [0, 1]},
),
{"x": 2, "y": 2},
),
(
_xr.DataArray(
_np.array([1, 2], _np.int32),
dims=["x"],
coords={"x": [0, 1]},
),
_xr.DataArray(
_np.array([[[1, 2], [3, 4]], [[1, 2], [3, 4]]], _np.int32),
dims=["x", "y", "z"],
coords={"x": [0, 1], "y": [0, 1], "z": [0, 1]},
),
{"x": 2, "y": 2, "z": 2},
),
],
)
def test_process_rules_calls_multi_cell_based_rule_special_cases(
input_array1, input_array2, dims
):
"""Some exceptional cases need to be tested for the multi_cell rule:
1. variables with different dimensions (1D vs 2D)
2. variables with different dimensions (1D vs 3D)"""
# Arrange
dataset = _xr.Dataset()
dataset["test1"] = input_array1
dataset["test2"] = input_array2
logger = Mock(ILogger)
rule = Mock(IMultiCellBasedRule)
rule.input_variable_names = ["test1", "test2"]
rule.output_variable_name = "output"
rule.execute.return_value = 1
processor = RuleProcessor([rule], dataset)
# Act
assert processor.initialize(logger)
output_dataset = processor.process_rules(dataset, logger)
# Assert
print(output_dataset.output, output_dataset.dims, output_dataset.dims == dims)
assert output_dataset.dims == dims
test_process_rules_copies_multi_coords_correctly()
Tests if during processing the coords are copied to the output array and there are no duplicates.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_copies_multi_coords_correctly():
"""Tests if during processing the coords are copied to the output array
and there are no duplicates."""
# Arrange
output_dataset = _xr.Dataset()
output_dataset["test"] = _xr.DataArray([32, 94, 9])
logger = Mock(ILogger)
rule = Mock(IArrayBasedRule)
rule_2 = Mock(IArrayBasedRule)
result_array = _xr.DataArray([27, 45, 93])
result_array = result_array.assign_coords({"test": _xr.DataArray([2, 4, 5])})
result_array_2 = _xr.DataArray([1, 2, 93])
result_array_2 = result_array_2.assign_coords({"test": _xr.DataArray([2, 4, 5])})
rule.input_variable_names = ["test"]
rule.output_variable_name = "output"
rule.execute.return_value = result_array
rule_2.input_variable_names = ["test"]
rule_2.output_variable_name = "output_2"
rule_2.execute.return_value = result_array_2
processor = RuleProcessor([rule, rule_2], output_dataset)
# Act
assert processor.initialize(logger)
result_dataset = processor.process_rules(output_dataset, logger)
# Assert
assert "test" in result_dataset.coords
# compare coords at the level of variable
result_array_coords = result_array.coords["test"]
result_output_var_coords = result_dataset.output.coords["test"] # output variable
assert (result_output_var_coords == result_array_coords).all()
# compare coords at the level of dataset /
# check if the coordinates are correctly copied to the dataset
result_dataset_coords = result_dataset.coords["test"]
assert (result_output_var_coords == result_dataset_coords).all()
# check if havnig an extra rule with coordinates then they are not copy pasted too
assert len(result_dataset.output.coords) == 1
test_process_rules_fails_for_uninitialized_processor()
Tests if an error is thrown if process_rules is called on the processor when it is not properly initialized.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_fails_for_uninitialized_processor():
"""Tests if an error is thrown if process_rules is called on the processor
when it is not properly initialized."""
# Arrange
input_dataset = _xr.Dataset()
output_dataset = _xr.Dataset()
input_dataset["test"] = _xr.DataArray([32, 94, 9])
logger = Mock(ILogger)
rule = Mock(IRule)
processor = RuleProcessor([rule], input_dataset)
# Act
with pytest.raises(RuntimeError) as exc_info:
processor.process_rules(output_dataset, logger)
exception_raised = exc_info.value
# Assert
expected_message = "Processor is not properly initialized, please initialize."
assert exception_raised.args[0] == expected_message
test_process_rules_given_rule_dependencies()
Tests if the processor can correctly process_rules given the rule dependencies.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_given_rule_dependencies():
"""Tests if the processor can correctly process_rules given
the rule dependencies.
"""
# Arrange
dataset = _xr.Dataset()
dataset["test"] = _xr.DataArray([32, 94, 9])
rule1 = Mock(IArrayBasedRule, id="rule1")
rule2 = Mock(IArrayBasedRule, id="rule2")
rule3 = Mock(IMultiArrayBasedRule, id="rule3")
logger = Mock(ILogger)
rule1.input_variable_names = ["test"]
rule2.input_variable_names = ["test"]
rule3.input_variable_names = ["out1", "out2"]
rule1.output_variable_name = "out1"
rule2.output_variable_name = "out2"
rule3.output_variable_name = "out3"
rule1.execute.return_value = _xr.DataArray([1, 2, 3])
rule2.execute.return_value = _xr.DataArray([4, 5, 6])
rule3.execute.return_value = _xr.DataArray([7, 8, 9])
rules: List[IRule] = [rule1, rule2, rule3]
processor = RuleProcessor(rules, dataset)
assert processor.initialize(logger)
# Act
processor.process_rules(dataset, logger)
# Assert
assert len(dataset) == 4
for rule in [rule1, rule2, rule3]:
rule.execute.assert_called_once_with(ANY, logger)
assert rule.output_variable_name in dataset.keys()
test_process_rules_throws_exception_for_array_based_rule_with_multiple_inputs()
Tests if an error is thrown during processing of an IArrayBasedRule if two inputs were defined.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_throws_exception_for_array_based_rule_with_multiple_inputs():
"""Tests if an error is thrown during processing of an IArrayBasedRule
if two inputs were defined."""
# Arrange
output_dataset = _xr.Dataset()
output_dataset["test1"] = _xr.DataArray([32, 94, 9])
output_dataset["test2"] = _xr.DataArray([32, 94, 9])
logger = Mock(ILogger)
rule = Mock(IArrayBasedRule)
rule.input_variable_names = ["test1", "test2"]
rule.output_variable_name = "output"
processor = RuleProcessor([rule], output_dataset)
assert processor.initialize(logger)
# Act
with pytest.raises(NotImplementedError) as exc_info:
processor.process_rules(output_dataset, logger)
exception_raised = exc_info.value
# Assert
expected_message = "Array based rule only supports one input array."
assert exception_raised.args[0] == expected_message
test_process_rules_throws_exception_for_unsupported_rule()
Tests if an error is thrown when trying to execute a rule that is not supported.
Source code in tests/business/entities/test_rule_processor.py
def test_process_rules_throws_exception_for_unsupported_rule():
"""Tests if an error is thrown when trying to execute a rule that is
not supported."""
# Arrange
output_dataset = _xr.Dataset()
input_array = _xr.DataArray([32, 94, 9])
output_dataset["test"] = input_array
logger = Mock(ILogger)
rule = Mock(IRule)
rule.name = "test"
rule.input_variable_names = ["test"]
rule.output_variable_name = "output"
processor = RuleProcessor([rule], output_dataset)
assert processor.initialize(logger)
# Act
with pytest.raises(NotImplementedError) as exc_info:
processor.process_rules(output_dataset, logger)
exception_raised = exc_info.value
# Assert
expected_message = f"Can not execute rule {rule.name}."
assert exception_raised.args[0] == expected_message
test_process_values_outside_limits(example_rule, input_value, expected_output_value, expected_log_message)
Test the function execution with input values outside the interval limits.
Source code in tests/business/entities/test_rule_processor.py
@pytest.mark.parametrize(
"input_value, expected_output_value, expected_log_message",
[
(-1, (10, [1, 0]), "value less than min: 1 occurence(s)"),
(11, (20, [0, 1]), "value greater than max: 1 occurence(s)"),
],
)
def test_process_values_outside_limits(
example_rule,
input_value: int,
expected_output_value: int,
expected_log_message: str,
):
"""
Test the function execution with input values outside the interval limits.
"""
# Arrange
logger = Mock(ILogger)
dataset = _xr.Dataset()
dataset["test1"] = _xr.DataArray(input_value)
rule = Mock(ICellBasedRule)
rule.input_variable_names = ["test1"]
rule.output_variable_name = "output"
rule.execute.return_value = expected_output_value
processor = RuleProcessor([rule], dataset)
# Act
assert processor.initialize(logger)
processor.process_rules(dataset, logger)
# Assert
assert example_rule.execute(input_value, logger) == expected_output_value
processor.process_rules(dataset, logger)
logger.log_warning.assert_called_with(expected_log_message)