test_time_aggregation_rule

Tests for time aggregation rule

test_aggregate_time_rule_without_time_dimension()

TimeAggregationRule should give an error when no time dim is defined

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

def test_aggregate_time_rule_without_time_dimension():
    """TimeAggregationRule should give an error when no time dim is defined"""
    # create test set
    logger = Mock(ILogger)
    rule = TimeAggregationRule(
        name="test",
        input_variable_names=["foo"],
        operation_type=TimeOperationType.ADD,
    )

    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

test_create_time_aggregation_rule_should_set_defaults()

Test creating a time aggregation rule with defaults

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

def test_create_time_aggregation_rule_should_set_defaults():
    """Test creating a time aggregation rule with defaults"""

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

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

test_execute_value_array_aggregate_time_monthly_add()

Aggregate input_variable_names of a TimeAggregationRule (add, monthly)

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

def test_execute_value_array_aggregate_time_monthly_add():
    """Aggregate input_variable_names of a TimeAggregationRule (add, monthly)"""

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

    rule.settings.time_scale = "month"

    time_aggregation = rule.execute(value_array_monthly, logger)

    result_data = [0.1, 0.9, 0.5]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_monthly], dims=["time_month"]
    )

    # Assert
    assert (
        _xr.testing.assert_allclose(time_aggregation, result_array, atol=1e-11) is None
    )

test_execute_value_array_aggregate_time_monthly_average()

Aggregate input_variable_names of a TimeAggregationRule (average, monthly)

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

def test_execute_value_array_aggregate_time_monthly_average():
    """Aggregate input_variable_names of a TimeAggregationRule (average, monthly)"""

    # create test set
    logger = Mock(ILogger)
    rule = TimeAggregationRule(
        name="test",
        input_variable_names=["foo"],
        operation_type=TimeOperationType.AVERAGE,
    )
    rule.settings.time_scale = "month"

    time_aggregation = rule.execute(value_array_monthly, logger)

    result_data = [0.1, 0.45, 0.25]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_monthly], dims=["time_month"]
    )

    # Assert
    assert (
        _xr.testing.assert_allclose(time_aggregation, result_array, atol=1e-11) is None
    )

test_execute_value_array_aggregate_time_monthly_max()

Aggregate input_variable_names of a TimeAggregationRule (max, monthly)

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

def test_execute_value_array_aggregate_time_monthly_max():
    """Aggregate input_variable_names of a TimeAggregationRule (max, monthly)"""

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

    time_aggregation = rule.execute(value_array_monthly, logger)

    result_data = [0.1, 0.7, 0.3]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_monthly], dims=["time_month"]
    )

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

test_execute_value_array_aggregate_time_monthly_median()

Test aggregate input_variable_names of a TimeAggregationRule (median, monthly)

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

def test_execute_value_array_aggregate_time_monthly_median():
    """Test aggregate input_variable_names of a TimeAggregationRule (median, monthly)"""

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

    time_aggregation = rule.execute(value_array_monthly, logger)

    result_data = [0.1, 0.45, 0.25]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_monthly], dims=["time_month"]
    )

    # Assert
    assert (
        _xr.testing.assert_allclose(time_aggregation, result_array, atol=1e-11) is None
    )

test_execute_value_array_aggregate_time_monthly_min()

Aggregate input_variable_names of a TimeAggregationRule (min, monthly)

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

def test_execute_value_array_aggregate_time_monthly_min():
    """Aggregate input_variable_names of a TimeAggregationRule (min, monthly)"""

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

    time_aggregation = rule.execute(value_array_monthly, logger)

    result_data = [0.1, 0.2, 0.2]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_monthly], dims=["time_month"]
    )

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

test_execute_value_array_aggregate_time_monthly_percentile()

Test aggregate input_variable_names of a TimeAggregationRule (PERCENTILE, monthly)

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

def test_execute_value_array_aggregate_time_monthly_percentile():
    """Test aggregate input_variable_names of a TimeAggregationRule
    (PERCENTILE, monthly)"""

    # create test set
    logger = Mock(ILogger)
    rule = TimeAggregationRule(
        name="test",
        input_variable_names=["foo"],
        operation_type=TimeOperationType.PERCENTILE,
    )
    rule.settings.time_scale = "month"
    rule.settings.percentile_value = 10

    time_aggregation = rule.execute(value_array_monthly, logger)
    result_data = [0.1, 0.25, 0.21]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_monthly], dims=["time_month"]
    )

    # Assert
    assert (
        _xr.testing.assert_allclose(time_aggregation, result_array, atol=1e-11) is None
    )

test_execute_value_array_aggregate_time_monthly_stdev()

Test aggregate input_variable_names of a TimeAggregationRule (STDEV, monthly)

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

def test_execute_value_array_aggregate_time_monthly_stdev():
    """Test aggregate input_variable_names of a TimeAggregationRule
    (STDEV, monthly)"""

    # create test set
    logger = Mock(ILogger)
    rule = TimeAggregationRule(
        name="test",
        input_variable_names=["foo"],
        operation_type=TimeOperationType.STDEV,
    )
    rule.settings.time_scale = "month"

    time_aggregation = rule.execute(value_array_monthly, logger)
    result_data = [0.0, 0.25, 0.05]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_monthly], dims=["time_month"]
    )

    # Assert
    assert (
        _xr.testing.assert_allclose(time_aggregation, result_array, atol=1e-11) is None
    )

test_execute_value_array_aggregate_time_yearly_add()

Aggregate input_variable_names of a TimeAggregationRule (add, yearly)

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

def test_execute_value_array_aggregate_time_yearly_add():
    """Aggregate input_variable_names of a TimeAggregationRule (add, yearly)"""

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

    time_aggregation = rule.execute(value_array_yearly, logger)

    result_data = [1.2, 0.4]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_yearly], dims=["time_year"]
    )

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

test_execute_value_array_aggregate_time_yearly_average()

Aggregate input_variable_names of a TimeAggregationRule (average, yearly)

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

def test_execute_value_array_aggregate_time_yearly_average():
    """Aggregate input_variable_names of a TimeAggregationRule (average, yearly)"""

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

    time_aggregation = rule.execute(value_array_yearly, logger)

    result_data = [0.3, 0.2]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_yearly], dims=["time_year"]
    )

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

test_execute_value_array_aggregate_time_yearly_max()

Aggregate input_variable_names of a TimeAggregationRule (max, yearly)

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

def test_execute_value_array_aggregate_time_yearly_max():
    """Aggregate input_variable_names of a TimeAggregationRule (max, yearly)"""

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

    time_aggregation = rule.execute(value_array_yearly, logger)

    result_data = [0.7, 0.3]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_yearly], dims=["time_year"]
    )

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

test_execute_value_array_aggregate_time_yearly_median()

Test aggregate input_variable_names of a TimeAggregationRule (median, yearly)

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

def test_execute_value_array_aggregate_time_yearly_median():
    """Test aggregate input_variable_names of a TimeAggregationRule (median, yearly)"""

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

    time_aggregation = rule.execute(value_array_yearly, logger)

    result_data = [0.2, 0.2]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_yearly], dims=["time_year"]
    )

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

test_execute_value_array_aggregate_time_yearly_min()

Aggregate input_variable_names of a TimeAggregationRule (min, yearly)

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

def test_execute_value_array_aggregate_time_yearly_min():
    """Aggregate input_variable_names of a TimeAggregationRule (min, yearly)"""

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

    time_aggregation = rule.execute(value_array_yearly, logger)

    result_data = [0.1, 0.1]
    result_array = _xr.DataArray(
        result_data, coords=[result_time_yearly], dims=["time_year"]
    )

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

test_operation_type_not_implemented()

Test that the time aggregation rule gives an error if no operation_type is given

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

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

    # create test set
    logger = Mock(ILogger)
    rule = TimeAggregationRule(
        name="test",
        input_variable_names=["foo"],
        operation_type="test",
    )
    rule.settings.time_scale = "month"

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

    exception_raised = exc_info.value

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