TP2000 360 - Refactor automated business rule checks.

.github/workflows/django.yml

@@ -46,6 +46,13 @@ jobs:
       run: |
         python manage.py makemigrations --dry-run --check
 
+    - name: Check for business rule data migrations
+      env:
+        DATABASE_URL: postgres://postgres:postgres@localhost:5432/postgres
+        DJANGO_SETTINGS_MODULE: settings.test
+      run: |
+        python manage.py sync_business_rules --check
+
     - name: Run tests
       env:
         DATABASE_URL: postgres://postgres:postgres@localhost:5432/postgres

checks/checks.py

@@ -1,226 +1,133 @@
-from functools import cached_property
-from typing import Collection
-from typing import Dict
-from typing import Iterator
+import abc
+import logging
+import typing
+from collections import defaultdict
 from typing import Optional
+from typing import Set
 from typing import Tuple
-from typing import Type
-from typing import TypeVar
 
+from checks.models import BusinessRuleModel
+from checks.models import BusinessRuleResult
 from checks.models import TrackedModelCheck
-from checks.models import TransactionCheck
-from common.business_rules import ALL_RULES
 from common.business_rules import BusinessRule
-from common.business_rules import BusinessRuleViolation
-from common.models.trackedmodel import TrackedModel
+from common.models import TrackedModel
+from common.models import Transaction
 from common.models.utils import get_current_transaction
-from common.models.utils import override_current_transaction
-
-CheckResult = Tuple[bool, Optional[str]]
 
+logger = logging.getLogger(__name__)
 
-Self = TypeVar("Self")
-
+CheckResult = Tuple[bool, Optional[str]]
 
-class Checker:
-    """
-    A ``Checker`` is an object that knows how to perform a certain kind of check
-    against a model.
 
-    Checkers can be applied against a model. The logic of the checker will be
-    run and the result recorded as a ``TrackedModelCheck``.
-    """
+class Checker(abc.ABC):
+    @classmethod
+    @abc.abstractmethod
+    def get_model_rule_mapping(
+        cls: abc.abstractclassmethod,
+        model: TrackedModel,
+        rules: Optional[Set[str]] = None,
+    ) -> typing.Dict[TrackedModel, Set[str]]:
+        """Implementing classes should return a dict mapping classes to sets of
+        business rules that apply to them."""
+        return {}
 
-    @cached_property
-    def name(self) -> str:
+    @classmethod
+    def apply_rules(
+        cls,
+        rules: typing.Sequence[BusinessRule],
+        transaction: Transaction,
+        model: TrackedModel,
+    ):
         """
-        The name string that on a per-model basis uniquely identifies the
-        checker.
-
-        The name should be deterministic (i.e. not rely on the current
-        environment, memory locations or random data) so that the system can
-        record the name in the database and later use it to work out whether
-        this check has been run. The name doesn't need to include any details
-        about the model.
-
-        By default this is the name of the class, but it can include any other
-        non-model data that is unique to the checker. For a more complex
-        example, see ``IndirectBusinessRuleChecker.name``.
+        TODO - Get rules_to_run - set of rules that have not been run.
         """
-        return type(self).__name__
+        # model.content_hash().digest()
 
-    @classmethod
-    def checkers_for(cls: Type[Self], model: TrackedModel) -> Collection[Self]:
-        """
-        Returns instances of this ``Checker`` that should apply to the model.
+        # rule_models = {
+        #     rule_model.name: rule_model for rule_model in type(model).get_business_rule_models()
+        # }
+        # TrackedModel check represents and ongoing check
 
-        What checks apply to a model is sometimes data-dependent, so it is the
-        responsibility of the ``Checker`` class to tell the system what
-        instances of itself it would expect to run against the model. For an
-        example, see ``IndirectBusinessRuleChecker.checkers_for``.
-        """
-        raise NotImplementedError()
-
-    def run(self, model: TrackedModel) -> CheckResult:
-        """Runs Checker-dependent logic and returns an indication of success."""
-        raise NotImplementedError()
-
-    def apply(self, model: TrackedModel, context: TransactionCheck):
-        """Applies the check to the model and records success."""
-
-        success, message = False, None
-        try:
-            with override_current_transaction(context.transaction):
-                success, message = self.run(model)
-        except Exception as e:
-            success, message = False, str(e)
-        finally:
-            return TrackedModelCheck.objects.create(
-                model=model,
-                transaction_check=context,
-                check_name=self.name,
-                successful=success,
-                message=message,
-            )
+        # To minimise the amount of queries, data is fetched up front and results are batched where possible.
+        rule_models = [*BusinessRuleModel.from_rules(rules)]
 
+        head_transaction = Transaction.objects.approved().last()
+        check, created = TrackedModelCheck.objects.get_or_create(
+            model=model,
+            head_transaction=head_transaction,
+            # content_hash=model.content_hash().digest(),
+        )
 
-class BusinessRuleChecker(Checker):
-    """
-    A ``Checker`` that runs a ``BusinessRule`` against a model.
+        # TODO: Get exclude existing rules
+        results = [
+            rule_model.get_result(model.transaction, model)
+            for rule_model in rule_models
+        ]
 
-    This class is expected to be sub-typed for a specific rule by a call to
-    ``of()``.
+        results = BusinessRuleResult.objects.bulk_create(results)
 
-    Attributes:
-        checker_cache (dict): (class attribute)  Cache of Business checkers created by ``of()``.
-    """
+        check.results.add(*results)
+        print(results)
+        return check
 
-    rule: Type[BusinessRule]
 
-    _checker_cache: Dict[str, BusinessRule] = {}
+class BusinessRuleChecker(Checker):
+    """A``Checker`` that runs a ``BusinessRule`` against a model."""
 
     @classmethod
-    def of(cls: Type, rule_type: Type[BusinessRule]) -> Type:
+    def get_model_rule_mapping(
+        cls,
+        model: TrackedModel,
+        rules: Optional[Set[str]] = None,
+    ):
         """
-        Return a subclass of a Checker, e.g. BusinessRuleChecker,
-        IndirectBusinessRuleChecker that runs the passed in business rule.
-
-        Example, creating a BusinessRuleChecker for ME32:
-
-        >>> BusinessRuleChecker.of(measures.business_rules.ME32)
-        <class 'checks.checks.BusinessRuleCheckerOf[measures.business_rules.ME32]'>
-
-        This API is usually called by .applicable_to, however this docstring should
-        illustrate what it does.
+        Return a dict mapping business rules to the passed in model.
 
-        Checkers are created once and then cached in _checker_cache.
+        This returns a dict, with the passed in model used as a key (this allows LinkedModelsBusinessRuleChecker to map models other than the passed in model to rules.)
 
-        As well as a small performance improvement, caching aids debugging by ensuring
-        the same checker instance is returned if the same cls is passed to ``of``.
+        :param model: TrackedModel instance
+        :param rules: Optional list of rule names to filter by.
+        :return: Dict with one entry for the passed in model the values are the rule instances to apply.
         """
-        checker_name = f"{cls.__name__}Of[{rule_type.__module__}.{rule_type.__name__}]"
+        if rules is None:
+            return {model: set(model.business_rules)}
 
-        # If the checker class was already created, return it.
-        checker_class = cls._checker_cache.get(checker_name)
-        if checker_class is not None:
-            return checker_class
+        # User passed in a certain set of rule names to run, filter the business rules by these names
+        filtered_rules = {
+            rule for rule in model.business_rules if rule.__name__ in rules
+        }
+        return {model: filtered_rules}
 
-        # No existing checker was found, so create it:
 
-        class BusinessRuleCheckerOf(cls):
-            # Creating this class explicitly in code is more readable than using type(...)
-            # Once created the name will be mangled to include the rule to be checked.
-
-            f"""Apply the following checks as specified in {rule_type.__name__}"""
-            rule = rule_type
-
-            def __repr__(self):
-                return f"<{checker_name}>"
-
-        BusinessRuleCheckerOf.__name__ = checker_name
-
-        cls._checker_cache[checker_name] = BusinessRuleCheckerOf
-        return BusinessRuleCheckerOf
+class LinkedModelsBusinessRuleChecker(Checker):
+    """A ``Checker`` that runs a ``BusinessRule`` against a model that is linked
+    to the model being checked, and for which a change in the checked model
+    could result in a business rule failure against the linked model."""
 
     @classmethod
-    def checkers_for(cls: Type[Self], model: TrackedModel) -> Collection[Self]:
-        """If the rule attribute on this BusinessRuleChecker matches any in the
-        supplied TrackedModel instance's business_rules, return it in a list,
-        otherwise there are no matches so return an empty list."""
-        if cls.rule in model.business_rules:
-            return [cls()]
-        return []
-
-    def run(self, model: TrackedModel) -> CheckResult:
+    def get_model_rule_mapping(cls, model: TrackedModel, rules: Optional[Set] = None):
         """
-        :return CheckResult, a Tuple(rule_passed: str, violation_reason: Optional[str]).
-        """
-        transaction = get_current_transaction()
-        try:
-            self.rule(transaction).validate(model)
-            return True, None
-        except BusinessRuleViolation as violation:
-            return False, violation.args[0]
-
-
-class IndirectBusinessRuleChecker(BusinessRuleChecker):
-    """
-    A ``Checker`` that runs a ``BusinessRule`` against a model that is linked to
-    the model being checked, and for which a change in the checked model could
-    result in a business rule failure against the linked model.
-
-    This is a base class: subclasses for checking specific rules are created by
-    calling ``of()``.
-    """
-
-    rule: Type[BusinessRule]
-    linked_model: TrackedModel
-
-    def __init__(self, linked_model: TrackedModel) -> None:
-        self.linked_model = linked_model
-        super().__init__()
-
-    @cached_property
-    def name(self) -> str:
-        # Include the identity of the linked model in the checker name, so that
-        # each linked model needs to be checked for all checks to be complete.
-        return f"{super().name}[{self.linked_model.pk}]"
-
-    @classmethod
-    def checkers_for(cls: Type[Self], model: TrackedModel) -> Collection[Self]:
-        """Return a set of IndirectBusinessRuleCheckers for every model found on
-        rule.get_linked_models."""
-        rules = set()
-        transaction = get_current_transaction()
-        if cls.rule in model.indirect_business_rules:
-            for linked_model in cls.rule.get_linked_models(model, transaction):
-                rules.add(cls(linked_model))
-        return rules
-
-    def run(self, model: TrackedModel) -> CheckResult:
+        :param model: Initial TrackedModel instance
+        :param rules: Optional list of rule names to filter by.
+        :return: Dict mapping linked models with sets of the BusinessRules that apply to them.
         """
-        Return the result of running super.run, passing self.linked_model, and.
+        tx = get_current_transaction()
 
-        return it as a CheckResult - a Tuple(rule_passed: str, violation_reason: Optional[str])
-        """
-        result, message = super().run(self.linked_model)
-        message = f"{self.linked_model}: " + message if message else None
-        return result, message
+        model_rules = defaultdict(set)
 
+        for rule in [*model.indirect_business_rules]:
+            for linked_model in rule.get_linked_models(model, tx):
+                if rules is not None and rule.__name__ not in rules:
+                    continue
 
-def checker_types() -> Iterator[Type[Checker]]:
-    """
-    Return all registered Checker types.
+                model_rules[linked_model].add(rule)
 
-    See ``checks.checks.BusinessRuleChecker.of``.
-    """
-    for rule in ALL_RULES:
-        yield BusinessRuleChecker.of(rule)
-        yield IndirectBusinessRuleChecker.of(rule)
+        # Downcast to a dict - this API (and unit testing) a little more sane.
+        return {**model_rules}
 
 
-def applicable_to(model: TrackedModel) -> Iterator[Checker]:
-    """Return instances of any Checker classes applicable to the supplied
-    TrackedModel instance."""
-    for checker_type in checker_types():
-        yield from checker_type.checkers_for(model)
+# Checkers in priority list order, checkers for linked models come first.
+ALL_CHECKERS = {
+    "LinkedModelsBusinessRuleChecker": LinkedModelsBusinessRuleChecker,
+    "BusinessRuleChecker": BusinessRuleChecker,
+}

checks/management/commands/list_business_rules.py

@@ -0,0 +1,40 @@
+from collections import defaultdict
+
+from django.core.management import BaseCommand
+
+from checks.models import BusinessRuleModel
+from common.business_rules import ALL_RULES
+
+
+class Command(BaseCommand):
+    """Display the business rules in the system and database."""
+
+    def handle(self, *app_labels, **options):
+        self.stdout.write("Rule Name,  In System,  In Database,  Status")
+
+        # Create a dictionary of rule_names, then a couple of flags
+        # to determine status.
+        rule_info = defaultdict(dict)
+        for rule_name in BusinessRuleModel.objects.values("name"):
+            rule_info[rule_name["name"]]["in_database"] = True
+
+        for rule_name in ALL_RULES.keys():
+            rule_info[rule_name]["in_system"] = True
+
+        for rule_name, info in rule_info.items():
+            in_database = info.get("in_database", False)
+            in_system = info.get("in_system", False)
+
+            if in_database and in_system:
+                status = "In Sync"
+            elif in_database:
+                status = "Pending Removal"
+            elif in_system:
+                status = "Pending Addition"
+
+            self.stdout.write(
+                f"{rule_name},"
+                f" {'Y' if in_system else 'N'},"
+                f" {'Y' if in_database else 'N'},"
+                f" {status}",
+            )