Add possibility to check if upgrade check rule applies (#12981)

closes: #12897

airflow/upgrade/formatters.py

@@ -74,20 +74,23 @@ def end_checking(self, rule_statuses):
     @staticmethod
     def display_recommendations(rule_statuses):
         for rule_status in rule_statuses:
+            # Show recommendations only if there are any messaged
             if not rule_status.messages:
                 continue
 
-            # Show recommendations only if there are any messaged
             rule = rule_status.rule
-            lines = [
-                rule.title,
-                "-" * len(rule.title),
-                rule.description,
-                "",
-                "Problems:",
-                "",
-            ]
-            lines.extend(['{:>3}.  {}'.format(i, m) for i, m in enumerate(rule_status.messages, 1)])
+            lines = [rule.title, "-" * len(rule.title)]
+            if rule_status.skipped:
+                lines.extend([rule_status.messages[0]])
+            else:
+                if rule.description:
+                    lines.extend([rule.description])
+                lines.extend([
+                    "",
+                    "Problems:",
+                    "",
+                ])
+                lines.extend(['{:>3}.  {}'.format(i, m) for i, m in enumerate(rule_status.messages, 1)])
             msg = "\n".join(lines)
 
             formatted_msg = pygments.highlight(
@@ -96,7 +99,12 @@ def display_recommendations(rule_statuses):
             print(formatted_msg)
 
     def on_next_rule_status(self, rule_status):
-        status = colorize("green", "SUCCESS") if rule_status.is_success else colorize("red", "FAIL")
+        if rule_status.skipped:
+            status = colorize("yellow", "SKIPPED")
+        elif rule_status.is_success:
+            status = colorize("green", "SUCCESS")
+        else:
+            status = colorize("red", "FAIL")
         status_line_fmt = self.prepare_status_line_format()
         print(status_line_fmt.format(rule_status.rule.title, status))
 

airflow/upgrade/problem.py

@@ -25,21 +25,25 @@ class RuleStatus(NamedTuple(
     'RuleStatus',
     [
         ('rule', BaseRule),
-        ('messages', List[str])
+        ('messages', List[str]),
+        ('skipped', bool)
     ]
 )):
-
     @property
     def is_success(self):
         return len(self.messages) == 0
 
     @classmethod
     def from_rule(cls, rule):
         # type: (BaseRule) -> RuleStatus
+        msg = rule.should_skip()
+        if msg:
+            return cls(rule=rule, messages=[msg], skipped=True)
+
         messages = []  # type: List[str]
         result = rule.check()
         if isinstance(result, str):
             messages = [result]
         elif isinstance(result, Iterable):
             messages = list(result)
-        return cls(rule=rule, messages=messages)
+        return cls(rule=rule, messages=messages, skipped=False)

airflow/upgrade/rules/base_rule.py

@@ -33,5 +33,12 @@ def description(self):
         """A long description explaining the problem in detail. This can be an entry from UPDATING.md file."""
         pass
 
+    def should_skip(self):
+        """
+        Executes a pre check of configuration. If returned value is
+        True then the checking the rule is omitted.
+        """
+        pass
+
     def check(self):
         pass

airflow/upgrade/rules/pod_template_file_rule.py

@@ -52,6 +52,11 @@ class PodTemplateFileRule(BaseRule):
 needed once this pod_template_file has been generated.
 """
 
+    def should_skip(self):
+        # Check this rule only if users use KubernetesExecutor
+        if conf.get("core", "executor") != "KubernetesExecutor":
+            return "Skipped because this rule applies only to environment using KubernetesExecutor."
+
     def check(self):
         pod_template_file = conf.get("kubernetes", "pod_template_file", fallback=None)
         if not pod_template_file:

tests/upgrade/test_formattes.py

@@ -34,15 +34,28 @@ def check(self):
         return MESSAGES
 
 
+class MockRuleToSkip(BaseRule):
+    title = "title"
+    description = "description"
+
+    def should_skip(self):
+        return "Yes, skipp this rule"
+
+
 class TestJSONFormatter:
-    @mock.patch("airflow.upgrade.checker.ALL_RULES", [MockRule()])
+    @mock.patch("airflow.upgrade.checker.ALL_RULES", [MockRule(), MockRuleToSkip()])
     @mock.patch("airflow.upgrade.checker.logging.disable")  # mock to avoid side effects
     def test_output(self, _):
         expected = [
             {
                 "rule": MockRule.__name__,
                 "title": MockRule.title,
                 "messages": MESSAGES,
+            },
+            {
+                "rule": MockRuleToSkip.__name__,
+                "title": MockRuleToSkip.title,
+                "messages": ["Yes, skipp this rule"],
             }
         ]
         parser = cli.CLIFactory.get_parser()

tests/upgrade/test_problem.py

@@ -22,14 +22,15 @@
 
 class TestRuleStatus:
     def test_is_success(self):
-        assert RuleStatus(rule=mock.MagicMock(), messages=[]).is_success is True
-        assert RuleStatus(rule=mock.MagicMock(), messages=["aaa"]).is_success is False
+        assert RuleStatus(rule=mock.MagicMock(), messages=[], skipped=False).is_success is True
+        assert RuleStatus(rule=mock.MagicMock(), messages=["aaa"], skipped=False).is_success is False
 
     def test_rule_status_from_rule(self):
         msgs = ["An interesting problem to solve"]
         rule = mock.MagicMock()
         rule.check.return_value = msgs
+        rule.should_skip.return_value = None
 
         result = RuleStatus.from_rule(rule)
         rule.check.assert_called_once_with()
-        assert result == RuleStatus(rule, msgs)
+        assert result == RuleStatus(rule, msgs, skipped=False)