Run formatting

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

triage_rules/apps.py

@@ -2,4 +2,4 @@
 
 
 class TriageRulesConfig(AppConfig):
-    name = 'triage_rules'
+    name = "triage_rules"

triage_rules/engine.py

@@ -28,20 +28,14 @@ def evaluate(self, decision_point):
         if target == DecisionPoint.TARGET_VULNERABILITY:
             queryset = decision_point.query.get_qs(user=self.user)
             return queryset.filter(
-                Q(
-                    pk__in=self.product.get_vulnerability_qs().values("pk")
-                )
-                | Q(
-                    pk__in=self.product.affected_by_vulnerabilities.values("pk")
-                )
+                Q(pk__in=self.product.get_vulnerability_qs().values("pk"))
+                | Q(pk__in=self.product.affected_by_vulnerabilities.values("pk"))
             ).exists()
 
         if target == DecisionPoint.TARGET_PRODUCT:
             return self._product_matches_query(decision_point.query)
 
-        raise ValueError(
-            f"Unknown target {target}"
-        )
+        raise ValueError(f"Unknown target {target}")
 
     def _product_matches_query(self, query):
         q_objects = []
@@ -53,10 +47,8 @@ def _product_matches_query(self, query):
         if not q_objects:
             return False
 
-        queryset = (
-            Product.unsecured_objects
-            .scope(self.product.dataspace)
-            .filter(pk=self.product.pk)
+        queryset = Product.unsecured_objects.scope(self.product.dataspace).filter(
+            pk=self.product.pk
         )
         operator_type = operator.or_ if query.operator == "or" else operator.and_
         return queryset.filter(reduce(operator_type, q_objects)).exists()
@@ -66,7 +58,6 @@ class RuleMatcher:
 
     @staticmethod
     def matches(rule, vector):
-
         """
         vector
 
@@ -82,20 +73,12 @@ def matches(rule, vector):
 
             expected = condition.expected
 
-            actual = vector[
-                condition.decision_point.name
-            ]
+            actual = vector[condition.decision_point.name]
 
-            if (
-                expected == ExpectedValue.TRUE
-                and not actual
-            ):
+            if expected == ExpectedValue.TRUE and not actual:
                 return False
 
-            if (
-                expected == ExpectedValue.FALSE
-                and actual
-            ):
+            if expected == ExpectedValue.FALSE and actual:
                 return False
 
         return True
@@ -109,14 +92,10 @@ def __init__(self, ruleset):
 
     def evaluate(self, vector):
 
-        rules = (
-            self.ruleset.rules
-            .prefetch_related(
-                "conditions",
-                "decision",
-            )
-            .order_by("priority")
-        )
+        rules = self.ruleset.rules.prefetch_related(
+            "conditions",
+            "decision",
+        ).order_by("priority")
 
         for rule in rules:
 
@@ -130,7 +109,6 @@ def evaluate(self, vector):
 
 
 class EvaluationEngine:
-
     """
     Public entry point.
 
@@ -142,33 +120,20 @@ class EvaluationEngine:
     )
     """
 
-    def evaluate(
-        self,
-        product,
-        rulesets,
-        user=None,
-    ):
-
-        decision_point_evaluator = (
-            DecisionPointEvaluator(
-                product,
-                user=user,
-            )
+    def evaluate(self, product, rulesets, user=None):
+
+        decision_point_evaluator = DecisionPointEvaluator(
+            product,
+            user=user,
         )
 
         vector = {}
 
-        decision_points = (
-            self.get_all_decision_points(
-                rulesets
-            )
-        )
+        decision_points = self.get_all_decision_points(rulesets)
 
         for dp in decision_points:
 
-            vector[dp.name] = (
-                decision_point_evaluator.evaluate(dp)
-            )
+            vector[dp.name] = decision_point_evaluator.evaluate(dp)
 
         winning = None
 
@@ -178,48 +143,29 @@ def evaluate(
             if not ruleset.enabled:
                 continue
 
-            decision = (
-                RulesetEvaluator(
-                    ruleset
-                ).evaluate(vector)
-            )
+            decision = RulesetEvaluator(ruleset).evaluate(vector)
 
-            if (
-                winning is None
-                or ruleset.precedence
-                > winning_precedence
-            ):
+            if winning is None or ruleset.precedence > winning_precedence:
                 winning = decision
-                winning_precedence = (
-                    ruleset.precedence
-                )
+                winning_precedence = ruleset.precedence
 
         return EvaluationResult(
             decision=winning,
             decision_vector=vector,
         )
 
     @staticmethod
-    def get_all_decision_points(
-        rulesets,
-    ):
+    def get_all_decision_points(rulesets):
 
         seen = {}
 
         for ruleset in rulesets:
             if not ruleset.enabled:
                 continue
 
-            for rule in (
-                ruleset.rules
-                .prefetch_related(
-                    "conditions__decision_point"
-                )
-            ):
+            for rule in ruleset.rules.prefetch_related("conditions__decision_point"):
 
-                for condition in (
-                    rule.conditions.all()
-                ):
+                for condition in rule.conditions.all():
                     dp = condition.decision_point
                     if not dp.enabled:
                         continue
@@ -231,23 +177,11 @@ def get_all_decision_points(
 
 class EvaluationResult:
 
-    def __init__(
-        self,
-        decision,
-        decision_vector,
-    ):
-
+    def __init__(self, decision, decision_vector):
         self.decision = decision
-
         self.vector = decision_vector
 
     def __repr__(self):
-
         return (
-            f"<EvaluationResult "
-            f"decision={self.decision} "
-            f"vector={self.vector}>"
+            f"<EvaluationResult " f"decision={self.decision} " f"vector={self.vector}>"
         )
-
-
-# Have policies that are more related to real life examples

triage_rules/loader.py

@@ -85,15 +85,11 @@ def _load_decisions(decisions_data, dataspace):
 
         action = config.get("action")
         if action not in dict(TriageDecision.ACTION_CHOICES):
-            raise PolicyLoadError(
-                f"Decision '{name}' has invalid action '{action}'."
-            )
+            raise PolicyLoadError(f"Decision '{name}' has invalid action '{action}'.")
 
         timeline_days = config.get("timeline_days")
         if timeline_days is None:
-            raise PolicyLoadError(
-                f"Decision '{name}' requires 'timeline_days'."
-            )
+            raise PolicyLoadError(f"Decision '{name}' requires 'timeline_days'.")
 
         decision, _ = TriageDecision.objects.update_or_create(
             name=name,
@@ -253,9 +249,7 @@ def _load_rule(ruleset, rule_config, decisions, decision_points):
 
     rule_name = rule_config.get("name")
     if not rule_name:
-        raise PolicyLoadError(
-            f"Each rule in ruleset '{ruleset.name}' requires 'name'."
-        )
+        raise PolicyLoadError(f"Each rule in ruleset '{ruleset.name}' requires 'name'.")
 
     decision_name = rule_config.get("decision")
     decision = decisions.get(decision_name)
@@ -304,9 +298,7 @@ def _iter_conditions(conditions_data):
                 raise PolicyLoadError(
                     "List-style rule conditions require 'decision_point'."
                 )
-            items.append(
-                (decision_point_name, condition.get("expected"))
-            )
+            items.append((decision_point_name, condition.get("expected")))
         return items
 
     raise PolicyLoadError("Rule conditions must be a mapping or list.")