[v3-1-test] Fix cron expression display for Day-of-Month and Day-of-Week conflicts (#54644)

* Fix cron expression display for Day-of-Month and Day-of-Week conflicts

* Add test case for CronMixin description attribute

* Add test case for CronMixin description attribute

* Add test case for CronMixin description attribute

* Add test case for CronMixin description attribute

* Add test case for CronMixin description attribute

---------
(cherry picked from commit c6531bb)

Co-authored-by: shreyaskj-0710 <[email protected]>
Co-authored-by: Ryan Hatter <[email protected]>

Author: 2 people

airflow-core/src/airflow/timetables/_cron.py

@@ -71,17 +71,52 @@ def __init__(self, cron: str, timezone: str | Timezone | FixedTimezone) -> None:
         self._timezone = timezone
 
         try:
-            descriptor = ExpressionDescriptor(
-                expression=self._expression, casing_type=CasingTypeEnum.Sentence, use_24hour_time_format=True
-            )
             # checking for more than 5 parameters in Cron and avoiding evaluation for now,
             # as Croniter has inconsistent evaluation with other libraries
             if len(croniter(self._expression).expanded) > 5:
                 raise FormatException()
-            interval_description: str = descriptor.get_description()
+
+            self.description = self._describe_with_dom_dow_fix(self._expression)
+
         except (CroniterBadCronError, FormatException, MissingFieldException):
-            interval_description = ""
-        self.description: str = interval_description
+            self.description = ""
+
+    def _describe_with_dom_dow_fix(self, expression: str) -> str:
+        """
+        Return cron description with fix for DOM+DOW conflicts.
+
+        If both DOM and DOW are restricted, explain them as OR.
+        """
+        cron_fields = expression.split()
+
+        if len(cron_fields) < 5:
+            return ExpressionDescriptor(
+                expression, casing_type=CasingTypeEnum.Sentence, use_24hour_time_format=True
+            ).get_description()
+
+        dom = cron_fields[2]
+        dow = cron_fields[4]
+
+        if dom != "*" and dow != "*":
+            # Case: conflict → DOM OR DOW
+            cron_fields_dom = cron_fields.copy()
+            cron_fields_dom[4] = "*"
+            day_of_month_desc = ExpressionDescriptor(
+                " ".join(cron_fields_dom), casing_type=CasingTypeEnum.Sentence, use_24hour_time_format=True
+            ).get_description()
+
+            cron_fields_dow = cron_fields.copy()
+            cron_fields_dow[2] = "*"
+            day_of_week_desc = ExpressionDescriptor(
+                " ".join(cron_fields_dow), casing_type=CasingTypeEnum.Sentence, use_24hour_time_format=True
+            ).get_description()
+
+            return f"{day_of_month_desc} (or) {day_of_week_desc}"
+
+        # no conflict → return normal description
+        return ExpressionDescriptor(
+            expression, casing_type=CasingTypeEnum.Sentence, use_24hour_time_format=True
+        ).get_description()
 
     def __eq__(self, other: object) -> bool:
         """

airflow-core/tests/unit/timetables/test_cron_mixin.py

@@ -0,0 +1,41 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.timetables._cron import CronMixin
+
+SAMPLE_TZ = "UTC"
+
+
+def test_valid_cron_expression():
+    cm = CronMixin("* * 1 * *", SAMPLE_TZ)  # every day at midnight
+    assert isinstance(cm.description, str)
+    assert "Every minute" in cm.description or "month" in cm.description
+
+
+def test_invalid_cron_expression():
+    cm = CronMixin("invalid cron", SAMPLE_TZ)
+    assert cm.description == ""
+
+
+def test_dom_and_dow_conflict():
+    cm = CronMixin("* * 1 * 1", SAMPLE_TZ)  # 1st of month or Monday
+    desc = cm.description
+
+    assert "(or)" in desc
+    assert "Every minute, on day 1 of the month" in desc
+    assert "Every minute, only on Monday" in desc