test_rolling_statistics_rule

Tests for rolling statistics rule

test_create_rolling_statistics_rule_should_set_defaults()

Test creating a rolling statistics rule with defaults

Source code in tests/business/entities/rules/test_rolling_statistics_rule.py

def test_create_rolling_statistics_rule_should_set_defaults():
    """Test creating a rolling statistics rule with defaults"""

    # Arrange & Act
    rule = RollingStatisticsRule(
        name="test",
        input_variable_names=["foo"],
        operation_type=TimeOperationType.MIN,
    )

    # Assert
    assert rule.name == "test"
    assert rule.description == ""
    assert isinstance(rule, RollingStatisticsRule)

test_execute_value_array_rolling_statistics_average()

RullingStatisticsRule (average, yearly)

Source code in tests/business/entities/rules/test_rolling_statistics_rule.py

def test_execute_value_array_rolling_statistics_average():
    """RullingStatisticsRule (average, yearly)"""

    # create test set
    logger = Mock(ILogger)
    rule = RollingStatisticsRule(
        name="test",
        input_variable_names=["foo"],
        operation_type=TimeOperationType.MEDIAN,
    )

    rule.settings.time_scale = "day"
    rule.period = 2

    rolling_statistic = rule.execute(value_array, logger)

    result_data = [np.nan, np.nan, 0.2, 0.2, 0.2, 0.2]
    result_array = _xr.DataArray(result_data, coords=[time], dims=["time"])

    # Assert
    assert _xr.testing.assert_equal(rolling_statistic, result_array) is None

test_execute_value_array_rolling_statistics_max()

RollingStatisticsRule (max, yearly)

Source code in tests/business/entities/rules/test_rolling_statistics_rule.py

def test_execute_value_array_rolling_statistics_max():
    """RollingStatisticsRule (max, yearly)"""

    # create test set
    logger = Mock(ILogger)
    rule = RollingStatisticsRule(
        name="test",
        input_variable_names=["foo"],
        operation_type=TimeOperationType.MAX,
    )

    rule.settings.time_scale = "day"
    rule.period = 2

    rolling_statistic = rule.execute(value_array, logger)

    result_data = [np.nan, np.nan, 0.7, 0.7, 0.3, 0.3]
    result_array = _xr.DataArray(result_data, coords=[time], dims=["time"])

    # Assert
    assert _xr.testing.assert_equal(rolling_statistic, result_array) is None

test_execute_value_array_rolling_statistics_min()

RullingStatisticsRule (min, yearly)

Source code in tests/business/entities/rules/test_rolling_statistics_rule.py

def test_execute_value_array_rolling_statistics_min():
    """RullingStatisticsRule (min, yearly)"""

    # create test set
    logger = Mock(ILogger)
    rule = RollingStatisticsRule(
        name="test", input_variable_names=["foo"], operation_type=TimeOperationType.MIN
    )
    rule.settings.time_scale = "day"
    rule.period = 2

    rolling_statistic = rule.execute(value_array, logger)

    result_data = [np.nan, np.nan, 0.1, 0.2, 0.2, 0.1]
    result_array = _xr.DataArray(result_data, coords=[time], dims=["time"])

    # Assert
    assert _xr.testing.assert_equal(rolling_statistic, result_array) is None

test_operation_type_not_implemented()

Test that the rulling statistics rule gives an error if no operation_type is given

Source code in tests/business/entities/rules/test_rolling_statistics_rule.py

def test_operation_type_not_implemented():
    """Test that the rulling statistics rule gives an error
    if no operation_type is given"""

    # create test set
    logger = Mock(ILogger)
    rule = RollingStatisticsRule(
        name="test", input_variable_names=["foo"], operation_type="test"
    )
    rule.settings.time_scale = "day"
    rule.period = 2

    with pytest.raises(NotImplementedError) as exc_info:
        rule.execute(value_array, logger)

    exception_raised = exc_info.value

    # Assert
    expected_message = "The operation type 'test' is currently not supported"
    assert exception_raised.args[0] == expected_message

test_rolling_statistics_rule_without_time_dimension()

RollingStatisticsRule should give an error when no time dim is defined

Source code in tests/business/entities/rules/test_rolling_statistics_rule.py

def test_rolling_statistics_rule_without_time_dimension():
    """RollingStatisticsRule should give an error when no time dim is defined"""
    # create test set
    logger = Mock(ILogger)
    rule = RollingStatisticsRule(
        name="test", input_variable_names=["foo"], operation_type=TimeOperationType.ADD
    )
    rule.settings.time_scale = "day"
    rule.period = 365

    test_data = [1.2, 0.4]
    test_array = _xr.DataArray(test_data, name="test_with_error")

    with pytest.raises(ValueError) as exc_info:
        rule.execute(test_array, logger)

    exception_raised = exc_info.value

    # Assert
    expected_message = "No time dimension found for test_with_error"
    assert exception_raised.args[0] == expected_message