test_model_builder

Tests for ModelFactory class

test_create_multiply_rule_based_model()

Test creating a multiply-rule-based model via factory

Source code in tests/business/workflow/test_model_builder.py

def test_create_multiply_rule_based_model():
    """Test creating a multiply-rule-based model via factory"""

test_create_rule_based_model()

Test creating a rule-based model via builder

Source code in tests/business/workflow/test_model_builder.py

def test_create_rule_based_model():
    """Test creating a rule-based model via builder"""

    # Arrange
    logger = Mock(ILogger)
    model_data = Mock(IModelData)
    dataset = Mock()
    dataset_data = Mock(IDatasetData)
    da_layer = Mock(IDataAccessLayer)

    multiply_rule_data = MultiplyRuleData("abc", [[2.0, 5.86]], "a")
    multiply_rule_data.output_variable = "b"

    step_function_rule_data = StepFunctionRuleData(
        "step_function_name", [0.0, 20.0, 100.0], [1.0, 2.0, 3.0], "input_name"
    )
    step_function_rule_data.description = "descript_step_func_rule"
    step_function_rule_data.output_variable = "output_step_func_name"

    rule_data_layer_filter_rule = LayerFilterRuleData("lfrname", 2, "var1")
    rule_data_layer_filter_rule.output_variable = "output_name"

    time_aggregation_rule = TimeAggregationRuleData(
        "taname", TimeOperationType.MIN, "var1"
    )
    time_aggregation_rule.time_scale = "Month"
    time_aggregation_rule.output_variable = "output"

    combine_results_rule_data = CombineResultsRuleData(
        "test_rule_name", ["foo", "hello"], "MULTIPLY"
    )
    combine_results_rule_data.output_variable = "output"

    formula_rule_data = FormulaRuleData("test_rule_name", ["foo", "bar"], "foo + bar")
    formula_rule_data.output_variable = "output"

    model_data.name = "Test model"
    model_data.datasets = [dataset_data]
    model_data.rules = [
        multiply_rule_data,
        step_function_rule_data,
        combine_results_rule_data,
        rule_data_layer_filter_rule,
        time_aggregation_rule,
        formula_rule_data,
    ]
    model_data.partition = ""

    da_layer.read_input_dataset.return_value = dataset
    model_builder = ModelBuilder(da_layer, logger)

    # Act
    model = model_builder.build_model(model_data)

    # Assert

    assert isinstance(model, RuleBasedModel)
    assert model.name == "Test model"
    assert dataset in model.input_datasets
    assert len(model.rules) == 6

    # logs info about model creation
    logger.log_info.assert_called_once()

test_create_rule_based_model_with_non_supported_rule()

Test creating a rule-based model with a rule that is not supported/recognized by the builder. This should throw an exception

Source code in tests/business/workflow/test_model_builder.py

def test_create_rule_based_model_with_non_supported_rule():
    """Test creating a rule-based model with a rule that is
    not supported/recognized by the builder.
    This should throw an exception"""

    # Arrange
    logger = Mock(ILogger)
    model_data = Mock(IModelData)
    dataset_data = Mock(IDatasetData)
    da_layer = Mock(IDataAccessLayer)

    rules_data = Mock(IRuleData)

    rules_data.name = "test"

    model_data.name = "Test model"
    model_data.datasets = [dataset_data]
    model_data.rules = [rules_data]

    model_builder = ModelBuilder(da_layer, logger)

    # Act & Assert
    with pytest.raises(NotImplementedError) as exc_info:
        model_builder.build_model(model_data)

    exception_raised = exc_info.value

    # Assert
    expected_message = "The rule type of rule 'test' is currently not implemented"
    assert exception_raised.args[0] == expected_message