Default logical_date to now in airflowctl dagrun trigger to match UI behavior (apache#61047)

* Default logical_date to now in airflowctl dagrun trigger to match UI behavior

  When triggering DAG runs via airflowctl without specifying --logical-date, the
  parameter now defaults to the current timestamp instead of None. This aligns
  with the Airflow UI behavior where the trigger form pre-populates logical_date
  with the current time, providing a more intuitive user experience.

* Add tests

Author: dheerajturaga

airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py

@@ -86,6 +86,8 @@ def date_param():
     "dags list-warning",
     # Order of trigger and pause/unpause is important for test stability because state checked
     f"dags trigger --dag-id=example_bash_operator --logical-date={ONE_DATE_PARAM} --run-after={ONE_DATE_PARAM}",
+    # Test trigger without logical-date (should default to now)
+    "dags trigger --dag-id=example_bash_operator",
     "dags pause example_bash_operator",
     "dags unpause example_bash_operator",
     # DAG Run commands

airflow-ctl/src/airflowctl/ctl/cli_config.py

@@ -621,6 +621,16 @@ def _get_func(args: Namespace, api_operation: dict, api_client: Client = NEW_API
 
             if datamodel:
                 if datamodel_param_name:
+                    # Special handling for TriggerDAGRunPostBody: default logical_date to now
+                    # This matches the Airflow UI behavior where the form pre-fills with current time
+                    if (
+                        datamodel.__name__ == "TriggerDAGRunPostBody"
+                        and "logical_date" in method_params[datamodel_param_name]
+                        and method_params[datamodel_param_name]["logical_date"] is None
+                    ):
+                        method_params[datamodel_param_name]["logical_date"] = datetime.datetime.now(
+                            datetime.timezone.utc
+                        )
                     method_params[datamodel_param_name] = datamodel.model_validate(
                         method_params[datamodel_param_name]
                     )

airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py

@@ -355,3 +355,51 @@ def test_merge_commands(self, no_op_method):
                 assert "subcommand2" in sub_command_names
                 assert "subcommand3" in sub_command_names
                 assert "subcommand4" in sub_command_names
+
+    def test_trigger_dag_run_defaults_logical_date_to_now(self):
+        """Test that trigger command defaults logical_date to now when not provided."""
+        from datetime import datetime, timezone
+
+        from airflowctl.api.datamodels.generated import TriggerDAGRunPostBody
+
+        # Simulate the logic in _get_func from cli_config.py
+        # This is the actual code path that runs when user doesn't provide --logical-date
+
+        # Step 1: Simulate CLI args being parsed (logical_date=None)
+        method_params = {
+            "trigger_dag_run": {
+                "dag_run_id": None,
+                "data_interval_start": None,
+                "data_interval_end": None,
+                "logical_date": None,  # User did not provide --logical-date
+                "run_after": None,
+                "conf": None,
+                "note": None,
+                "partition_key": None,
+            }
+        }
+
+        # Step 2: Apply the defaulting logic (from cli_config.py lines 622-630)
+        datamodel = TriggerDAGRunPostBody
+        datamodel_param_name = "trigger_dag_run"
+
+        if (
+            datamodel.__name__ == "TriggerDAGRunPostBody"
+            and "logical_date" in method_params[datamodel_param_name]
+            and method_params[datamodel_param_name]["logical_date"] is None
+        ):
+            method_params[datamodel_param_name]["logical_date"] = datetime.now(timezone.utc)
+
+        # Step 3: Create the Pydantic model (what happens in the actual code)
+        trigger_body = datamodel.model_validate(method_params[datamodel_param_name])
+
+        # Step 4: Verify logical_date was set to now
+        assert trigger_body.logical_date is not None, "logical_date should be defaulted to now"
+        assert isinstance(trigger_body.logical_date, datetime)
+
+        # Verify it's close to current time (within 5 seconds)
+        time_diff = abs((datetime.now(timezone.utc) - trigger_body.logical_date).total_seconds())
+        assert time_diff < 5, f"logical_date should be close to now, but diff is {time_diff} seconds"
+
+        # Also verify timezone is UTC
+        assert trigger_body.logical_date.tzinfo is not None, "logical_date should have timezone info"