test_rule_handler.py

#!/usr/bin/env python3

# Copyright (C) 2023-2024:
#   Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.


import logging
import pytest

from rulelib import RuleHandler, RuleError

logger = logging.getLogger()


def test_rule_handler_init(building_tags_rule, source_db_config):
    """
    Test if the rule handler can be initialized without any rules, or with one rule.
    """

    # Test initialization of a rule handler without any rules provided.
    rule_handler = RuleHandler()
    assert len(rule_handler.rules) == 0

    # Test initialization of a rule handler with an empty rule dictionary.
    rule_handler = RuleHandler(rules={})
    assert len(rule_handler.rules) == 0

    # Test initialization of a rule handler with one rule.
    rule_handler = RuleHandler(
        rules={building_tags_rule.name: building_tags_rule},
        databases={building_tags_rule.db_name: source_db_config},
    )
    assert len(rule_handler.rules) == 1


def test_rule_handler(rule_handler, building_tags_rule, source_db_config):
    """
    Test if the rule handler has one rule and if that rule works when evaluated. The test is
    also executed for a subset of rules, using the `rule_names` input parameter.
    """

    assert "FloorspaceRule" in rule_handler.rules

    result = rule_handler.eval({"footprint": 100.0, "floors": 2})
    assert result["area"] == 200.0

    # Register another rule, but do not execute that rule.
    rule_handler.register_database(building_tags_rule.db_name, source_db_config)
    rule_handler.register(building_tags_rule)
    result = rule_handler.eval({"footprint": 100.0, "floors": 3}, rule_names=["FloorspaceRule"])
    assert "TagRule" in rule_handler.rules
    assert result["area"] == 300.0
    assert "tags" not in result


def test_rule_handler_register(
    floorspace_rule, building_tags_rule, rule_handler, source_db_config
):
    """
    Test if the registration of new rules works properly: If a rule is added with the same name,
    the number of rules should stay the same. If a different rule is added, the number of rules
    should increase.
    """

    # Add the already existing rule once more. This should not add another rule to the total.
    rule_handler.register(rule=floorspace_rule)
    assert len(rule_handler.rules.values()) == 1

    # Add another rule.
    rule_handler.register_database(building_tags_rule.db_name, source_db_config)
    rule_handler.register(rule=building_tags_rule)
    assert len(rule_handler.rules.values()) == 2
    assert "TagRule" in rule_handler.rules


def test_rule_handler_xml(rule_handler, source_db_config):
    """
    Test if the registration of XML rules works.
    """

    # Open XML-string and change rule name to `floorspace-2`
    xml_string = open("./tests/data/building_tags.xml", "r").read()

    rule_handler.register_database(source_db_config["dbname"], source_db_config)

    # Register rule
    rule_handler.register_xml(xml_string=xml_string)
    assert "TagRule" in rule_handler.rules


def test_rule_handler_unregister(rule_handler):
    """
    Test if unregistering rules works: if the only existing rule is removed, there should be no
    rules anymore in the `RuleHandler`.
    """

    rule_handler.unregister(rule_name="FloorspaceRule")
    assert len(rule_handler.rules.values()) == 0


def test_rule_handler_unregister_cascade(
    rule_handler, apartment_size_rule, apartments_rule, source_db_config
):
    """
    Test if the cascading of unregistering rules work.
    """

    rule_handler.register_database(apartments_rule.db_name, source_db_config)
    rule_handler.register(apartment_size_rule)
    rule_handler.register(apartments_rule)

    # The RuleHandler now has three rules: the FloorspaceRule is registered already.
    assert len(rule_handler.rules.values()) == 3

    rule_handler.unregister(rule_name="FloorspaceRule", cascade=True)

    # The `apartments_rule` is dependent on the FloorspaceRule, therefore it is unregistered
    # too.
    assert len(rule_handler.rules.values()) == 1


def test_fail_rule_handler_unregister_cascade(
    rule_handler, apartment_size_rule, apartments_rule, source_db_config
):
    """
    Test if the cascading of unregistering rules fails if set to `False` for rules that are
    dependent on other rules.
    """

    rule_handler.register_database(apartments_rule.db_name, source_db_config)
    rule_handler.register(apartment_size_rule)
    rule_handler.register(apartments_rule)

    # The RuleHandler now has three rules: the FloorspaceRule is registered already.
    assert len(rule_handler.rules.values()) == 3

    with pytest.raises(ValueError):
        # This results in a value error, because the `cascade` option default is `False`.
        rule_handler.unregister(rule_name="FloorspaceRule")


def test_fail_rule_handler(rule_handler):
    """
    Test if the parsing of the example results in a failure due to an incorrectly typed
    variable.
    """

    with pytest.raises(RuleError):
        rule_handler.eval({"footprint": 100.0, "floors": "5"})


def test_rule_dependencies(
    rule_handler, apartment_size_rule, apartments_rule, source_db_config
):
    """ "
    Test if the rule dependencies work properly
    """
    rule_handler.register_database(apartments_rule.db_name, source_db_config)
    rule_handler.register(apartment_size_rule)
    rule_handler.register(apartments_rule)

    # Check if the apartment rule is dependent on the floorspace rule
    assert "FloorspaceRule" in apartments_rule.dependencies
    assert "ApartmentSizeRule" in apartments_rule.dependencies

    # The apartment rule is dependent on the floorspace rule, therefore the subset contains
    # both. The floorspace rule is not dependent on anything, therefore the subset only
    # contains that rule
    assert len(list(rule_handler.subset(["FloorspaceRule"]))) == 1
    assert len(list(rule_handler.subset(["ApartmentsRule"]))) == 3

    result = rule_handler.eval(
        {
            "footprint": 100.0,
            "floors": 2,
            "country_iso_code": "BEL",
            "location": "Urban",
        }
    )
    assert result["number_apartments"] == 4


def test_fail_rule_dependencies(rule_handler, apartments_rule):
    """
    Test if the registering a rule with a missing dependency fails.
    """
    with pytest.raises(ValueError):

        rule_handler.register(apartments_rule)


def test_execute_cancel_command(rule_handler, execute_cancel_command_rule, floorspace_rule):
    """
    Test if the `eval` method inside the `RuleHandler` is stopped, when the value `CANCEL` is
    returned by the `ExecuteCancelCommandRule`.
    """

    # Test if the `floorspace_rule` functions without registering the
    # `execute_cancel_command_rule`.
    result = rule_handler.eval({"footprint": 100.0, "floors": 2})
    assert result["area"] == 200.0

    rule_handler.register(execute_cancel_command_rule)

    # Overwrite the dependencies of the `floorspace_rule` and re-register.
    floorspace_rule.dependencies = ["ExecuteCancelCommandRule"]
    rule_handler.register(floorspace_rule, overwrite=True)

    # Test if the processing of the rules is cancelled before the `floorspace_rule` is executed.
    result = rule_handler.eval({"footprint": 100.0, "floors": 2})
    assert "area" not in result